fix(acp): propagate request cancellation and close sessions cleanly (#49563)

2026-10-16 10:30:01 +08:00 · 2026-09-17 20:16:07 +05:30
parent e32de90d88
commit 19781a4447
13 changed files with 362 additions and 123 deletions
@@ -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<A>(promise: Promise<A>) {
@@ -52,6 +38,7 @@ async function run<A>(promise: Promise<A>) {
 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 ||
@@ -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<void>
+  requestPermission(params: RequestPermissionRequest, options?: SendRequestOptions): Promise<RequestPermissionResponse>
+  writeTextFile?(params: WriteTextFileRequest, options?: SendRequestOptions): Promise<WriteTextFileResponse>
+  extNotification?(method: string, params: Record<string, unknown>): Promise<void>
+}
+
+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"
@@ -5,6 +5,11 @@ export class SessionNotFoundError extends Schema.TaggedError<SessionNotFoundErro
  sessionId: Schema.String,
 }) {}

+export class SessionDirectoryMismatchError extends Schema.TaggedError<SessionDirectoryMismatchError>()(
+  "ACPSessionDirectoryMismatchError",
+  { sessionId: Schema.String, cwd: Schema.String },
+) {}
+
 export class InvalidConfigOptionError extends Schema.TaggedError<InvalidConfigOptionError>()(
  "ACPInvalidConfigOptionError",
  { configId: Schema.String },
@@ -37,6 +42,7 @@ export class ServiceFailureError extends Schema.TaggedError<ServiceFailureError>

 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":
@@ -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<AgentSideConnection, "sessionUpdate" | "requestPermission"> &
-  Partial<Pick<AgentSideConnection, "writeTextFile">>
+type Connection = Pick<ACPConnection.Connection, "sessionUpdate" | "requestPermission" | "writeTextFile">

 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<AgentSideConnection, "sessionUpdate">,
+  connection: Pick<Connection, "sessionUpdate">,
  sessionID: string,
  cwd: string,
  messages: readonly SessionMessageInfo[],
@@ -426,7 +429,7 @@ export async function replayMessages(
 }

 async function replayMessage(
-  connection: Pick<AgentSideConnection, "sessionUpdate">,
+  connection: Pick<Connection, "sessionUpdate">,
  sessionID: string,
  cwd: string,
  message: SessionMessageInfo,
@@ -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<EventSubscribeOutput, { type: "permission.asked" }>
-type Connection = Pick<AgentSideConnection, "requestPermission"> & Partial<Pick<AgentSideConnection, "writeTextFile">>
+type Connection = Pick<ACPConnection.Connection, "requestPermission" | "writeTextFile">
 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<Pick<AgentSideConnection, "writeTextFile">>
+  readonly connection: Pick<ACPConnection.Connection, "writeTextFile">
  readonly writeTextFile: boolean
  readonly sessionID: string
  readonly cwd: string
  readonly toolName: string
  readonly toolInput: ToolInput
  readonly metadata: Readonly<Record<string, unknown>>
+  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 },
+      )
    }),
  )
 }
@@ -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<AgentSideConnection, "sessionUpdate" | "requestPermission"> &
-  Partial<Pick<AgentSideConnection, "writeTextFile" | "extNotification" | "signal">>
-
 type Catalog = {
  readonly providers: ConfigOptionProvider[]
  readonly models: ModelInfo[]
@@ -102,15 +100,18 @@ export interface Interface {
  forkSession(input: ForkSessionRequest): Promise<ForkSessionResponse>
  setSessionConfigOption(input: SetSessionConfigOptionRequest): Promise<SetSessionConfigOptionResponse>
  setSessionMode(input: SetSessionModeRequest): Promise<SetSessionModeResponse>
-  prompt(input: PromptRequest): Promise<PromptResponse>
+  prompt(input: PromptRequest, signal?: AbortSignal): Promise<PromptResponse>
  cancel(input: CancelNotification): Promise<void>
 }

-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<string, Attached>()
  const catalogs = new Map<string, Promise<Catalog>>()
  const registeredMcp = new Map<string, Set<string>>()
-  const active = new Map<string, TurnControl>()
+  const active = new Map<string, { readonly control: TurnControl; readonly turn: Promise<PromptResponse> }>()
  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,
@@ -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.
@@ -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<void>()
+  const release = Promise.withResolvers<string>()
+  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<PromptResponse>("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<PromptResponse>("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<PromptResponse>("session/prompt", {
+      sessionId: closing.sessionId,
+      prompt: [{ type: "text", text: "hold" }],
+    })
+    await model.started
+    const order: string[] = []
+    const close = acp
+      .request<CloseSessionResponse>("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<PromptResponse>("session/prompt", {
+        sessionId: other.sessionId,
+        prompt: [{ type: "text", text: "still here" }],
+      }),
+    )
+    expect(next.stopReason).toBe("end_turn")
+  }, 60_000)
+})
@@ -51,6 +51,12 @@ describe("acp lifecycle subprocess", () => {
    )

    expect(selectConfigOption(loaded.configOptions, "model")?.category).toBe("model")
+    const mismatched = await acp.request<LoadSessionResponse>("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 () => {
@@ -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<AgentSideConnection["sessionUpdate"]>[0]
-type Connection = Pick<AgentSideConnection, "sessionUpdate" | "requestPermission"> &
-  Partial<Pick<AgentSideConnection, "writeTextFile">>
+type SessionUpdateParams = SessionNotification
+type Connection = Pick<ACPConnection.Connection, "sessionUpdate" | "requestPermission" | "writeTextFile">
 type Fixture = ReturnType<typeof createSseFixture>

 describe("acp permission behavior", () => {
  test("does not sync edits when writeTextFile was not advertised", async () => {
-    const writes: Parameters<AgentSideConnection["writeTextFile"]>[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<AgentSideConnection["writeTextFile"]>[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<AgentSideConnection["writeTextFile"]>[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<AbortSignal | undefined>()
+    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<RequestPermissionResponse>((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") {
@@ -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<AgentSideConnection["sessionUpdate"]>[0][] = []
+  const updates: SessionNotification[] = []
  const encoder = new TextEncoder()
  let eventController: ReadableStreamDefaultController<Uint8Array> | undefined
  const models = options.models ?? [testModel, secondModel]
@@ -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",
@@ -46,6 +46,11 @@ type Waiter = {

 export type AcpProcess = {
  readonly request: <T>(method: string, params?: unknown) => Promise<JsonRpcResponse<T>>
+  readonly send: <T>(
+    method: string,
+    params?: unknown,
+  ) => { readonly id: number; readonly response: Promise<JsonRpcResponse<T>> }
+  readonly notify: (method: string, params: unknown) => Promise<void>
  readonly waitForNotification: <T>(
    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<string> } = {},
+) {
  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<string, string | undefined> }):
    })
  }

-  return {
-    async request<T>(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<unknown>) => {
+    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 = <T>(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<T>(response)) throw new Error(`Invalid ACP response: ${JSON.stringify(response)}`)
      return response
-    },
+    })
+    return { id, response }
+  }
+
+  return {
+    request: <T>(method: string, params?: unknown) => send<T>(method, params).response,
+    send,
+    notify: (method: string, params: unknown) => write({ jsonrpc: "2.0", method, params }),
    async waitForNotification<T>(method: string, predicate: (params: T) => boolean, timeoutMs = 20_000) {
      const notification = await take(
        (message) => isNotification<T>(message) && message.method === method && predicate(message.params),