fix(cli): fail closed acp coordination

packages/cli/src/acp/event.ts

@@ -22,6 +22,9 @@ type Connection = Pick<AgentSideConnection, "sessionUpdate" | "requestPermission
 
 export type TurnControl = {
   cancelled: boolean
+  admitted?: boolean
+  interrupting?: boolean
+  stream?: AbortController
   readonly admission: AbortController
 }
 
@@ -69,6 +72,44 @@ function emptyToolState(): ToolState {
   return { name: "tool", input: {}, metadata: {}, content: [] }
 }
 
+async function openEventStream(input: {
+  readonly client: OpenCodeClient
+  readonly streamController: AbortController
+  readonly admission: AbortController
+  readonly connectionSignal?: AbortSignal
+  readonly sessionSignal?: AbortSignal
+}) {
+  const connectionAbort = () => {
+    input.streamController.abort()
+    input.admission.abort()
+  }
+  const sessionAbort = () => {
+    input.streamController.abort()
+    input.admission.abort()
+  }
+  let stream: AsyncIterator<EventSubscribeOutput> | undefined
+  let opened = false
+  const close = async () => {
+    input.streamController.abort()
+    input.connectionSignal?.removeEventListener("abort", connectionAbort)
+    input.sessionSignal?.removeEventListener("abort", sessionAbort)
+    await stream?.return?.(undefined).catch(() => {})
+  }
+  try {
+    input.connectionSignal?.addEventListener("abort", connectionAbort, { once: true })
+    input.sessionSignal?.addEventListener("abort", sessionAbort, { once: true })
+    if (input.connectionSignal?.aborted) connectionAbort()
+    if (input.sessionSignal?.aborted) sessionAbort()
+    stream = input.client.event.subscribe({ signal: input.streamController.signal })[Symbol.asyncIterator]()
+    const connected = await stream.next()
+    if (connected.done) throw new Error("event stream disconnected before prompt admission")
+    opened = true
+    return { stream, close }
+  } finally {
+    if (!opened) await close()
+  }
+}
+
 export async function streamTurn(input: {
   readonly client: OpenCodeClient
   readonly connection: Connection
@@ -83,12 +124,17 @@ export async function streamTurn(input: {
   readonly connectionSignal?: AbortSignal
   readonly sessionSignal?: AbortSignal
 }): Promise<PromptResponse> {
-  const streamController = new AbortController()
-  const connectionAbort = () => streamController.abort()
-  input.connectionSignal?.addEventListener("abort", connectionAbort, { once: true })
-  const stream = input.client.event.subscribe({ signal: streamController.signal })[Symbol.asyncIterator]()
-  const connected = await stream.next()
-  if (connected.done) throw new Error("event stream disconnected before prompt admission")
+  const streamController = input.control.stream ?? new AbortController()
+  input.control.stream = streamController
+  const opened = await openEventStream({
+    client: input.client,
+    streamController,
+    admission: input.control.admission,
+    connectionSignal: input.connectionSignal,
+    sessionSignal: input.sessionSignal,
+  })
+  const stream = opened.stream
+  const closeStream = opened.close
 
   const control = input.control
   let started = false
@@ -173,6 +219,7 @@ export async function streamTurn(input: {
       if (!eventSessionID || (eventSessionID !== input.sessionID && !child)) continue
       if (matchesStart(event, input.start)) {
         started = true
+        control.admitted = true
         continue
       }
       if (!started) continue
@@ -335,34 +382,38 @@ export async function streamTurn(input: {
     return "interrupted" as const
   }
 
-  const completed = consume("turn")
-  const closeStream = async () => {
-    streamController.abort()
-    input.connectionSignal?.removeEventListener("abort", connectionAbort)
-    input.sessionSignal?.removeEventListener("abort", connectionAbort)
-    await stream.return?.(undefined).catch(() => {})
-  }
+  const completed = consume("turn").then(
+    (value) => ({ success: true as const, value }),
+    (error) => ({ success: false as const, error }),
+  )
+  const submitted = input.submit(control.admission.signal).then(
+    () => ({ success: true as const }),
+    (error) => ({ success: false as const, error }),
+  )
   try {
-    await input.submit(control.admission.signal).catch((error) => {
-      if (!control.cancelled) throw error
-    })
-    if (input.action) {
-      streamController.abort()
-      await completed.catch(() => {})
-      return response(undefined, undefined, "succeeded", control.cancelled, undefined)
-    }
-    if (control.cancelled) {
-      await input.client.session.interrupt({ sessionID: input.sessionID }).catch(() => {})
-      if (!started) {
-        streamController.abort()
-        await completed.catch(() => {})
-        return response(undefined, undefined, "interrupted", true, undefined)
+    const first = await Promise.race([
+      submitted.then((result) => ({ source: "admission" as const, result })),
+      completed.then((result) => ({ source: "stream" as const, result })),
+    ])
+    const completion = await (async () => {
+      if (first.source === "stream") {
+        control.admission.abort()
+        return first.result
       }
-    }
-    const terminal = await completed
+      if (!first.result.success && !control.cancelled) throw first.result.error
+      if (first.result.success) control.admitted = true
+      if (input.action) {
+        streamController.abort()
+        await completed
+        return { success: true as const, value: "succeeded" as const }
+      }
+      if (control.cancelled && !started) streamController.abort()
+      return completed
+    })()
+    if (!completion.success && !control.cancelled) throw completion.error
+    const terminal = completion.success ? completion.value : "interrupted"
     if (input.childSessionUpdate && openChildren.size > 0 && !input.sessionSignal?.aborted) {
       handedOff = true
-      input.sessionSignal?.addEventListener("abort", connectionAbort, { once: true })
       void consume("background")
         .catch(() => {})
         .finally(closeStream)
@@ -381,7 +432,6 @@ export async function streamTurn(input: {
     )
   } catch (error) {
     streamController.abort()
-    await completed.catch(() => {})
     throw error
   } finally {
     if (!handedOff) await closeStream()

packages/cli/src/acp/service.ts

@@ -9,6 +9,7 @@ import {
   type SkillInfo,
 } from "@opencode-ai/client/promise"
 import { withTimestampedFallback } from "@opencode-ai/util/session-title-fallback"
+import { Effect, Semaphore } from "effect"
 import type {
   AgentSideConnection,
   AuthenticateRequest,
@@ -78,6 +79,27 @@ type Attached = {
   modeID: string
 }
 
+type ActiveTurn = {
+  readonly state: Attached
+  readonly control: TurnControl
+  readonly stopped: Promise<void>
+  readonly resolveStopped: () => void
+}
+
+type RegisteredMcp = {
+  readonly server: string
+  readonly config: ReturnType<typeof mcpConfig>
+}
+
+class McpRollbackError extends AggregateError {
+  readonly servers: ReadonlyArray<string>
+
+  constructor(primary: unknown, rollback: unknown, servers: ReadonlyArray<string>) {
+    super([primary, rollback], "ACP attachment failed and MCP rollback did not complete", { cause: primary })
+    this.servers = servers
+  }
+}
+
 type PreparedPrompt = {
   readonly start: TurnStart
   readonly text: string
@@ -104,11 +126,42 @@ export interface Interface {
   cancel(input: CancelNotification): Promise<void>
 }
 
+type KeyedLock<Key> = <A>(key: Key, operation: () => Promise<A>) => Promise<A>
+
+function makeKeyedLock<Key>(): KeyedLock<Key> {
+  const entries = new Map<Key, { readonly semaphore: Semaphore.Semaphore; users: number }>()
+  return async <A>(key: Key, operation: () => Promise<A>) => {
+    const current = entries.get(key)
+    const entry = current ?? { semaphore: Semaphore.makeUnsafe(1), users: 0 }
+    if (!current) entries.set(key, entry)
+    // Count holders and waiters so cleanup cannot split one key across two locks.
+    entry.users++
+    const result = await Effect.runPromise(
+      entry.semaphore.withPermit(
+        Effect.promise(() =>
+          operation().then(
+            (value) => ({ success: true as const, value }),
+            (error) => ({ success: false as const, error }),
+          ),
+        ),
+      ),
+    ).finally(() => {
+      entry.users--
+      if (entry.users === 0) entries.delete(key)
+    })
+    if (!result.success) throw result.error
+    return result.value
+  }
+}
+
 export function make(input: { readonly client: OpenCodeClient; readonly connection: 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 registeredMcp = new Map<string, Map<string, RegisteredMcp>>()
+  const uncertainMcp = new Map<string, Set<string>>()
+  const active = new Map<string, ActiveTurn>()
+  const withSessionLock = makeKeyedLock<string>()
+  const withLocationLock = makeKeyedLock<string>()
   const capabilities = { writeTextFile: false, childSessionUpdates: false }
 
   const catalog = (cwd: string) => {
@@ -128,15 +181,50 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
     throw new ACPError.SessionNotFoundError({ sessionId: sessionID })
   }
 
-  const detach = (sessionID: string) => {
-    sessions.get(sessionID)?.abort.abort()
-    sessions.delete(sessionID)
-    registeredMcp.delete(sessionID)
+  const retire = (state: Attached) => {
+    state.abort.abort()
+    const turn = active.get(state.id)
+    if (turn?.state !== state) return
+    turn.control.admission.abort()
+    active.delete(state.id)
   }
 
-  const attach = async (session: SessionInfo, cwd: string, mcpServers: readonly McpServer[]) => {
+  const detach = (sessionID: string) => {
+    const state = sessions.get(sessionID)
+    if (state) retire(state)
+    sessions.delete(sessionID)
+  }
+
+  const invalidateLocation = (cwd: string, servers: ReadonlyArray<string>) => {
+    sessions.forEach((state, sessionID) => {
+      if (state.cwd !== cwd) return
+      retire(state)
+      sessions.delete(sessionID)
+    })
+    const uncertain = new Set([...(registeredMcp.get(cwd)?.keys() ?? []), ...servers])
+    if (uncertain.size > 0) uncertainMcp.set(cwd, uncertain)
+    registeredMcp.delete(cwd)
+  }
+
+  const reconcileLocation = async (cwd: string) => {
+    const uncertain = uncertainMcp.get(cwd)
+    if (!uncertain) return
+    const removed = await Promise.allSettled(
+      [...uncertain].map((server) => input.client.mcp.remove({ server, location: { directory: cwd } })),
+    )
+    const failures = removed.flatMap((result) => (result.status === "rejected" ? [result.reason] : []))
+    if (failures.length > 0) throw new AggregateError(failures, "Failed to reconcile uncertain MCP configuration")
+    uncertainMcp.delete(cwd)
+  }
+
+  // Lifecycle operations acquire Session ID before entering this Location transaction.
+  const attach = async (
+    session: SessionInfo,
+    cwd: string,
+    mcpServers: readonly McpServer[],
+    replayHistory: boolean,
+  ) => {
     const currentCatalog = await catalog(cwd)
-    sessions.get(session.id)?.abort.abort()
     const state: Attached = {
       id: session.id,
       cwd,
@@ -145,25 +233,56 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
       model: session.model ?? currentCatalog.defaultModel,
       modeID: session.agent ?? currentCatalog.defaultModeID,
     }
-    sessions.set(session.id, state)
-    await registerMcpServers(input.client, registeredMcp, state, mcpServers)
-    await input.connection.sessionUpdate({
-      sessionId: state.id,
-      update: {
-        sessionUpdate: "available_commands_update",
-        availableCommands: [
-          ...state.catalog.commands,
-          ...state.catalog.skills.filter(
-            (skill) => !state.catalog.commands.some((command) => command.name === skill.name),
-          ),
-        ].map((command) => ({ name: command.name, description: command.description ?? "" })),
-      },
+    return withLocationLock(cwd, async () => {
+      const history = replayHistory ? await messages(input.client, state.id) : undefined
+      await reconcileLocation(cwd)
+      const registration = await registerMcpServers(
+        input.client,
+        registeredMcp.get(cwd) ?? new Map(),
+        state,
+        mcpServers,
+      ).catch((error) => {
+        state.abort.abort()
+        if (error instanceof McpRollbackError) invalidateLocation(cwd, error.servers)
+        throw error
+      })
+      return input.connection
+        .sessionUpdate({
+          sessionId: state.id,
+          update: {
+            sessionUpdate: "available_commands_update",
+            availableCommands: [
+              ...state.catalog.commands,
+              ...state.catalog.skills.filter(
+                (skill) => !state.catalog.commands.some((command) => command.name === skill.name),
+              ),
+            ].map((command) => ({ name: command.name, description: command.description ?? "" })),
+          },
+        })
+        .then(async () => {
+          if (history) await replayMessages(input.connection, state.id, state.cwd, history)
+          const previous = sessions.get(session.id)
+          if (previous) retire(previous)
+          sessions.set(session.id, state)
+          registeredMcp.set(cwd, registration.registered)
+          return state
+        })
+        .catch(async (error) => {
+          state.abort.abort()
+          const rollback = await registration.rollback().then(
+            () => ({ success: true as const }),
+            (failure) => ({ success: false as const, failure }),
+          )
+          if (!rollback.success) {
+            invalidateLocation(cwd, registration.changed)
+            throw new McpRollbackError(error, rollback.failure, registration.changed)
+          }
+          throw error
+        })
+    }).catch((error) => {
+      state.abort.abort()
+      throw error
     })
-    return state
-  }
-
-  const replay = async (state: Attached) => {
-    await replayMessages(input.connection, state.id, state.cwd, await messages(input.client, state.id))
   }
 
   const configOptions = (state: Attached) =>
@@ -213,15 +332,15 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
         agent: currentCatalog.defaultModeID,
         model: currentCatalog.defaultModel,
       })
-      const state = await attach(created, params.cwd, params.mcpServers)
+      const state = await withSessionLock(created.id, () => attach(created, params.cwd, params.mcpServers, false))
       return { sessionId: state.id, configOptions: configOptions(state) }
     },
-    loadSession: async (params) => {
-      const session = await getSession(input.client, params.sessionId)
-      const state = await attach(session, session.location.directory, params.mcpServers)
-      await replay(state)
-      return { configOptions: configOptions(state) }
-    },
+    loadSession: (params) =>
+      withSessionLock(params.sessionId, async () => {
+        const session = await getSession(input.client, params.sessionId)
+        const state = await attach(session, session.location.directory, params.mcpServers, true)
+        return { configOptions: configOptions(state) }
+      }),
     listSessions: async (params) => {
       const page = await input.client.session.list({
         ...(params.cwd ? { directory: params.cwd } : {}),
@@ -239,86 +358,109 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
         ...(page.cursor.next ? { nextCursor: page.cursor.next } : {}),
       }
     },
-    deleteSession: async (params) => {
-      await input.client.session.remove({ sessionID: params.sessionId }).catch((error) => {
-        if (!isSessionNotFoundError(error)) throw error
-      })
-      detach(params.sessionId)
-      return {}
-    },
-    resumeSession: async (params) => {
-      const session = await getSession(input.client, params.sessionId)
-      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(() => {})
-      return {}
-    },
+    deleteSession: (params) =>
+      withSessionLock(params.sessionId, async () => {
+        await input.client.session.remove({ sessionID: params.sessionId }).catch((error) => {
+          if (!isSessionNotFoundError(error)) throw error
+        })
+        detach(params.sessionId)
+        return {}
+      }),
+    resumeSession: (params) =>
+      withSessionLock(params.sessionId, async () => {
+        const session = await getSession(input.client, params.sessionId)
+        const state = await attach(session, session.location.directory, params.mcpServers ?? [], false)
+        return { configOptions: configOptions(state) }
+      }),
+    closeSession: (params) =>
+      withSessionLock(params.sessionId, async () => {
+        const turn = active.get(params.sessionId)
+        if (turn) {
+          turn.control.cancelled = true
+          turn.control.interrupting = true
+        }
+        detach(params.sessionId)
+        await input.client.session.interrupt({ sessionID: params.sessionId }).catch(() => {})
+        return {}
+      }),
     forkSession: async (params) => {
-      const forked = await input.client.session.fork({
-        sessionID: params.sessionId,
-        boundary: { type: "through" },
+      const forked = await withSessionLock(params.sessionId, () =>
+        input.client.session.fork({
+          sessionID: params.sessionId,
+          boundary: { type: "through" },
+        }),
+      )
+      const state = await withSessionLock(forked.id, async () => {
+        return attach(forked, forked.location.directory, params.mcpServers ?? [], true)
       })
-      const state = await attach(forked, forked.location.directory, params.mcpServers ?? [])
-      await replay(state)
       return { sessionId: state.id, configOptions: configOptions(state) }
     },
     setSessionConfigOption: async (params) => {
-      const state = await requireSession(params.sessionId)
       if (typeof params.value !== "string") throw new ACPError.InvalidConfigOptionError({ configId: params.configId })
-      switch (params.configId) {
-        case "model": {
-          const selected = requireModel(state.catalog, params.value)
-          state.model = selected
-          await input.client.session.switchModel({ sessionID: state.id, model: selected })
-          break
+      return withSessionLock(params.sessionId, async () => {
+        const state = await requireSession(params.sessionId)
+        switch (params.configId) {
+          case "model": {
+            const selected = requireModel(state.catalog, params.value)
+            await input.client.session.switchModel({ sessionID: state.id, model: selected })
+            state.model = selected
+            break
+          }
+          case "effort": {
+            const model = state.catalog.models.find(
+              (item) => item.providerID === state.model.providerID && item.id === state.model.id,
+            )
+            if (!model?.variants.some((variant) => variant.id === params.value))
+              throw new ACPError.InvalidEffortError({ effort: params.value })
+            const selected = { ...state.model, variant: params.value }
+            await input.client.session.switchModel({ sessionID: state.id, model: selected })
+            state.model = selected
+            break
+          }
+          case "mode":
+            await selectMode(input.client, state, params.value)
+            break
+          default:
+            throw new ACPError.InvalidConfigOptionError({ configId: params.configId })
         }
-        case "effort": {
-          const model = state.catalog.models.find(
-            (item) => item.providerID === state.model.providerID && item.id === state.model.id,
-          )
-          if (!model?.variants.some((variant) => variant.id === params.value))
-            throw new ACPError.InvalidEffortError({ effort: params.value })
-          state.model = { ...state.model, variant: params.value }
-          await input.client.session.switchModel({ sessionID: state.id, model: state.model })
-          break
-        }
-        case "mode":
-          await selectMode(input.client, state, params.value)
-          break
-        default:
-          throw new ACPError.InvalidConfigOptionError({ configId: params.configId })
-      }
-      return { configOptions: configOptions(state) }
+        return { configOptions: configOptions(state) }
+      })
     },
     setSessionMode: async (params) => {
-      await selectMode(input.client, await requireSession(params.sessionId), params.modeId)
+      await withSessionLock(params.sessionId, async () => {
+        const state = await requireSession(params.sessionId)
+        await selectMode(input.client, state, params.modeId)
+      })
       return {}
     },
     prompt: async (params) => {
-      const state = await requireSession(params.sessionId)
-      if (active.has(state.id)) {
-        throw new ACPError.ServiceFailureError({
-          safeMessage: `Session already has an active ACP prompt: ${state.id}`,
-          service: "session",
-        })
-      }
-      const messageID = SessionMessage.ID.create()
-      const prepared = preparePrompt(state.catalog, params.prompt, messageID)
-      const control: TurnControl = { cancelled: false, admission: new AbortController() }
+      const acquired = await withSessionLock(params.sessionId, async () => {
+        const state = await requireSession(params.sessionId)
+        if (active.has(state.id)) {
+          throw new ACPError.ServiceFailureError({
+            safeMessage: `Session already has an active ACP prompt: ${state.id}`,
+            service: "session",
+          })
+        }
+        const prepared = preparePrompt(state.catalog, params.prompt, SessionMessage.ID.create())
+        const control: TurnControl = {
+          cancelled: false,
+          admission: new AbortController(),
+          stream: new AbortController(),
+        }
+        const stopped = Promise.withResolvers<void>()
+        const turn = { state, control, stopped: stopped.promise, resolveStopped: () => stopped.resolve() }
+        active.set(state.id, turn)
+        return { state, control, prepared, turn }
+      })
+      const state = acquired.state
+      const control = acquired.control
+      const prepared = acquired.prepared
       const extNotification = input.connection.extNotification
       const childSessionUpdate =
         capabilities.childSessionUpdates && extNotification
           ? (update: ChildSessionUpdate) => extNotification(ChildSessionUpdateMethod, update).then(() => {})
           : undefined
-      active.set(state.id, control)
       const response = await streamTurn({
         client: input.client,
         connection: input.connection,
@@ -333,18 +475,49 @@ export function make(input: { readonly client: OpenCodeClient; readonly connecti
         submit: (signal) => submitPrompt(input.client, state, prepared, signal),
         ...(childSessionUpdate ? { childSessionUpdate } : {}),
       }).finally(() => {
-        if (active.get(state.id) === control) active.delete(state.id)
+        if (active.get(state.id) === acquired.turn) active.delete(state.id)
+        acquired.turn.resolveStopped()
       })
       await sendUsageUpdate(input.client, input.connection, state, response.usage?.totalTokens).catch(() => {})
       return response
     },
     cancel: async (params) => {
-      const current = active.get(params.sessionId)
-      if (current) {
-        current.cancelled = true
-        current.admission.abort()
+      const cancellation = await withSessionLock(params.sessionId, async () => {
+        const current = active.get(params.sessionId)
+        if (!current) return undefined
+        current.control.cancelled = true
+        current.control.admission.abort()
+        if (!current.control.admitted) {
+          current.control.stream?.abort()
+          return { current, interrupt: false as const }
+        }
+        if (current.control.interrupting) return { current, interrupt: false as const }
+        current.control.interrupting = true
+        return { current, interrupt: true as const }
+      })
+      if (!cancellation) return
+      const current = cancellation.current
+      if (!cancellation.interrupt) {
+        await current.stopped
+        return
       }
-      await input.client.session.interrupt({ sessionID: params.sessionId }).catch(() => {})
+      const interrupted = await input.client.session.interrupt({ sessionID: params.sessionId }).catch(async (error) => {
+        await withSessionLock(params.sessionId, async () => {
+          if (active.get(params.sessionId) === current) current.control.interrupting = false
+        })
+        throw error
+      })
+      if (!interrupted.interrupted) {
+        await withSessionLock(params.sessionId, async () => {
+          if (active.get(params.sessionId) === current) current.control.interrupting = false
+        })
+        throw new ACPError.ServiceFailureError({
+          safeMessage: `Failed to interrupt active ACP prompt: ${params.sessionId}`,
+          service: "session",
+        })
+      }
+      current.control.stream?.abort()
+      await current.stopped
     },
   }
 }
@@ -364,16 +537,21 @@ function preparePrompt(catalog: Catalog, prompt: PromptRequest["prompt"], messag
 
 async function submitPrompt(client: OpenCodeClient, session: Attached, prompt: PreparedPrompt, signal: AbortSignal) {
   if (prompt.synthetic.length > 0) {
-    await client.session.synthetic({
-      sessionID: session.id,
-      text: prompt.synthetic.join("\n\n"),
-      description: "ACP embedded context",
-      delivery: "steer",
-      resume: false,
-    })
+    await client.session.synthetic(
+      {
+        sessionID: session.id,
+        text: prompt.synthetic.join("\n\n"),
+        description: "ACP embedded context",
+        delivery: "steer",
+        resume: false,
+      },
+      { signal },
+    )
   }
-  if (prompt.start.type === "compaction") return client.session.compact({ sessionID: session.id, id: prompt.start.id })
-  if (prompt.skill) return client.session.skill({ sessionID: session.id, id: prompt.start.id, skill: prompt.skill.id })
+  if (prompt.start.type === "compaction")
+    return client.session.compact({ sessionID: session.id, id: prompt.start.id }, { signal })
+  if (prompt.skill)
+    return client.session.skill({ sessionID: session.id, id: prompt.start.id, skill: prompt.skill.id }, { signal })
   if (prompt.command) {
     return client.session.command(
       {
@@ -461,8 +639,8 @@ function requireModel(catalog: Catalog, modelID: string): ModelRef {
 
 async function selectMode(client: OpenCodeClient, state: Attached, modeID: string) {
   if (!state.catalog.modes.some((mode) => mode.id === modeID)) throw new ACPError.InvalidModeError({ mode: modeID })
-  state.modeID = modeID
   await client.session.switchAgent({ sessionID: state.id, agent: modeID })
+  state.modeID = modeID
 }
 
 async function getSession(client: OpenCodeClient, sessionID: string) {
@@ -487,26 +665,59 @@ async function messages(client: OpenCodeClient, sessionID: string) {
 
 async function registerMcpServers(
   client: OpenCodeClient,
-  registered: Map<string, Set<string>>,
+  current: Map<string, RegisteredMcp>,
   session: Attached,
   servers: readonly McpServer[],
 ) {
-  const current = registered.get(session.id) ?? new Set<string>()
-  registered.set(session.id, current)
-  await Promise.all(
-    servers.flatMap((server) => {
+  const requested = new Map(
+    servers.map((server) => {
       const config = mcpConfig(server)
-      const key = `${server.name}:${stableStringify(config)}`
-      if (current.has(key)) return []
-      current.add(key)
-      return [
-        client.mcp.add({ server: server.name, location: { directory: session.cwd }, config }).catch((error) => {
-          current.delete(key)
-          throw error
-        }),
-      ]
+      return [server.name, { server: server.name, config }] as const
     }),
   )
+  const changed = [...requested.values()].filter(
+    (entry) => stableStringify(current.get(entry.server)?.config) !== stableStringify(entry.config),
+  )
+  const registered = new Map(current)
+  changed.forEach((entry) => registered.set(entry.server, entry))
+  const rollback = async () => {
+    const results = await Promise.allSettled(
+      changed.map((entry) => {
+        const previous = current.get(entry.server)
+        if (previous) {
+          return client.mcp.add({
+            server: previous.server,
+            location: { directory: session.cwd },
+            config: previous.config,
+          })
+        }
+        return client.mcp.remove({ server: entry.server, location: { directory: session.cwd } })
+      }),
+    )
+    const failures = results.flatMap((result) => (result.status === "rejected" ? [result.reason] : []))
+    if (failures.length > 0) throw new AggregateError(failures, "Failed to roll back MCP configuration")
+  }
+  const additions = await Promise.allSettled(
+    changed.map((entry) =>
+      client.mcp.add({ server: entry.server, location: { directory: session.cwd }, config: entry.config }),
+    ),
+  )
+  const failure = additions.find((result): result is PromiseRejectedResult => result.status === "rejected")
+  if (failure) {
+    const rollbackResult = await rollback().then(
+      () => ({ success: true as const }),
+      (error) => ({ success: false as const, error }),
+    )
+    if (!rollbackResult.success) {
+      throw new McpRollbackError(
+        failure.reason,
+        rollbackResult.error,
+        changed.map((entry) => entry.server),
+      )
+    }
+    throw failure.reason
+  }
+  return { registered, rollback, changed: changed.map((entry) => entry.server) }
 }
 
 function mcpConfig(server: McpServer) {

packages/cli/test/acp/event-behavior.test.ts

@@ -624,6 +624,7 @@ describe("acp event behavior", () => {
     try {
       await withTimeout(submitted.promise, "cancel test prompt was not admitted")
       control.cancelled = true
+      control.interrupting = true
       control.admission.abort()
       expect(await fixture.client.session.interrupt({ sessionID: "ses_cancel" })).toEqual({ interrupted: true })
 
@@ -669,7 +670,7 @@ describe("acp event behavior", () => {
 
       const response = await withTimeout(result, "pre-admission cancellation did not terminate")
       expect(response).toMatchObject({ stopReason: "cancelled" })
-      expect(fixture.requests.filter((request) => request.path.endsWith("/interrupt"))).toHaveLength(1)
+      expect(fixture.requests.filter((request) => request.path.endsWith("/interrupt"))).toHaveLength(0)
     } finally {
       control.cancelled = true
       control.admission.abort()
@@ -678,6 +679,169 @@ describe("acp event behavior", () => {
     }
   })
 
+  test("observes stream failure while cancelled admission settles", async () => {
+    const fixture = createSseFixture()
+
+    try {
+      for (let index = 0; index < 20; index++) {
+        const submitted = Promise.withResolvers<void>()
+        const session = new AbortController()
+        const result = streamTurn({
+          client: fixture.client,
+          connection: recordingConnection([]),
+          sessionID: `ses_abort_stress_${index}`,
+          cwd: "/workspace",
+          start: { type: "input", id: `input_abort_stress_${index}` },
+          writeTextFile: false,
+          control: { cancelled: false, admission: new AbortController() },
+          sessionSignal: session.signal,
+          submit: (signal) =>
+            new Promise<void>((resolve) => {
+              submitted.resolve()
+              signal.addEventListener(
+                "abort",
+                () => {
+                  void Bun.sleep(5).then(resolve)
+                },
+                { once: true },
+              )
+            }),
+        })
+        const observed = result.catch((error: unknown) => error)
+
+        await submitted.promise
+        session.abort()
+        expect(await withTimeout(observed, "cancelled stress turn did not settle")).toBeInstanceOf(Error)
+      }
+    } finally {
+      await fixture.stop()
+    }
+  })
+
+  test("connection abort stops a turn with hanging admission", async () => {
+    const fixture = createSseFixture()
+    const submitted = Promise.withResolvers<void>()
+    const admissionAborted = Promise.withResolvers<void>()
+    const connection = new AbortController()
+    const result = streamTurn({
+      client: fixture.client,
+      connection: recordingConnection([]),
+      connectionSignal: connection.signal,
+      sessionID: "ses_connection_abort",
+      cwd: "/workspace",
+      start: { type: "input", id: "input_connection_abort" },
+      writeTextFile: false,
+      control: { cancelled: false, admission: new AbortController() },
+      submit: (signal) =>
+        new Promise<void>(() => {
+          submitted.resolve()
+          signal.addEventListener("abort", () => admissionAborted.resolve(), { once: true })
+        }),
+    })
+    const observed = result.catch((error: unknown) => error)
+
+    try {
+      await submitted.promise
+      connection.abort()
+
+      await withTimeout(admissionAborted.promise, "connection abort did not cancel admission")
+      expect(await withTimeout(observed, "connection-aborted turn did not settle")).toBeInstanceOf(Error)
+    } finally {
+      await fixture.stop()
+    }
+  })
+
+  test("cleans stream setup when attachment is already aborted", async () => {
+    const fixture = createSseFixture()
+    const session = new AbortController()
+    let submitted = false
+    session.abort()
+
+    try {
+      const failure = await streamTurn({
+        client: fixture.client,
+        connection: recordingConnection([]),
+        sessionSignal: session.signal,
+        sessionID: "ses_preaborted",
+        cwd: "/workspace",
+        start: { type: "input", id: "input_preaborted" },
+        writeTextFile: false,
+        control: { cancelled: false, admission: new AbortController() },
+        submit: async () => {
+          submitted = true
+        },
+      }).catch((error: unknown) => error)
+
+      expect(failure).toBeInstanceOf(Error)
+      expect(submitted).toBe(false)
+      expect(fixture.streamCount()).toBe(0)
+    } finally {
+      await fixture.stop()
+    }
+  })
+
+  test("preserves admission failure when admission settles first", async () => {
+    const fixture = createSseFixture()
+    const expected = new Error("admission failed first")
+
+    try {
+      const failure = await streamTurn({
+        client: fixture.client,
+        connection: recordingConnection([]),
+        sessionID: "ses_admission_first",
+        cwd: "/workspace",
+        start: { type: "input", id: "input_admission_first" },
+        writeTextFile: false,
+        control: { cancelled: false, admission: new AbortController() },
+        submit: () => Promise.reject(expected),
+      }).catch((error: unknown) => error)
+
+      expect(failure).toBe(expected)
+    } finally {
+      await fixture.stop()
+    }
+  })
+
+  test("preserves stream failure when stream settles before admission", async () => {
+    const fixture = createSseFixture()
+    const connection = new AbortController()
+    const submitted = Promise.withResolvers<void>()
+    const lateAdmission = new Error("admission failed later")
+    const result = streamTurn({
+      client: fixture.client,
+      connection: recordingConnection([]),
+      connectionSignal: connection.signal,
+      sessionID: "ses_stream_first",
+      cwd: "/workspace",
+      start: { type: "input", id: "input_stream_first" },
+      writeTextFile: false,
+      control: { cancelled: false, admission: new AbortController() },
+      submit: (signal) =>
+        new Promise<void>((_, reject) => {
+          submitted.resolve()
+          signal.addEventListener(
+            "abort",
+            () => {
+              void Bun.sleep(20).then(() => reject(lateAdmission))
+            },
+            { once: true },
+          )
+        }),
+    })
+    const observed = result.catch((error: unknown) => error)
+
+    try {
+      await submitted.promise
+      connection.abort()
+
+      const failure = await withTimeout(observed, "stream-first failure did not settle")
+      expect(failure).toBeInstanceOf(Error)
+      expect(failure).not.toBe(lateAdmission)
+    } finally {
+      await fixture.stop()
+    }
+  })
+
   test("cancels unsupported session forms so execution can continue", async () => {
     const fixture = createSseFixture({
       onPrompt({ id, send }) {

packages/cli/test/acp/service-directory.test.ts

@@ -181,6 +181,180 @@ describe("acp service directory behavior", () => {
     expect(invalidConfig).toMatchObject({ _tag: "ACPInvalidConfigOptionError" })
   })
 
+  test("keeps the last confirmed config after switch requests are rejected", async () => {
+    let rejected: "model" | "effort" | "config-mode" | "mode" | undefined
+    await using fixture = makeACPFixture({
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_rejected_config") })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_rejected_config/model") {
+          if (rejected === "model" || rejected === "effort") return new Response(null, { status: 409 })
+          return new Response(null, { status: 204 })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_rejected_config/agent") {
+          if (rejected === "config-mode" || rejected === "mode") return new Response(null, { status: 409 })
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    rejected = "model"
+    const modelFailure = await fixture.service
+      .setSessionConfigOption({ sessionId: session.sessionId, configId: "model", value: "test/second-model" })
+      .catch((error: unknown) => error)
+    rejected = undefined
+    const afterModelFailure = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "high",
+    })
+
+    rejected = "effort"
+    const effortFailure = await fixture.service
+      .setSessionConfigOption({ sessionId: session.sessionId, configId: "effort", value: "default" })
+      .catch((error: unknown) => error)
+    rejected = undefined
+    const afterEffortFailure = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "mode",
+      value: "plan",
+    })
+
+    rejected = "config-mode"
+    const configModeFailure = await fixture.service
+      .setSessionConfigOption({ sessionId: session.sessionId, configId: "mode", value: "build" })
+      .catch((error: unknown) => error)
+    rejected = undefined
+    const afterConfigModeFailure = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "model",
+      value: "test/test-model",
+    })
+
+    rejected = "mode"
+    const modeFailure = await fixture.service
+      .setSessionMode({ sessionId: session.sessionId, modeId: "build" })
+      .catch((error: unknown) => error)
+    rejected = undefined
+    const afterModeFailure = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "high",
+    })
+
+    expect([modelFailure, effortFailure, configModeFailure, modeFailure]).toEqual([
+      expect.any(Error),
+      expect.any(Error),
+      expect.any(Error),
+      expect.any(Error),
+    ])
+    expect(currentValue(afterModelFailure, "model")).toBe("test/test-model")
+    expect(currentValue(afterModelFailure, "effort")).toBe("high")
+    expect(currentValue(afterEffortFailure, "effort")).toBe("high")
+    expect(currentValue(afterConfigModeFailure, "mode")).toBe("plan")
+    expect(currentValue(afterModeFailure, "mode")).toBe("plan")
+  })
+
+  test("orders overlapping model and effort switches for one session", async () => {
+    const modelStarted = Promise.withResolvers<void>()
+    const releaseModel = Promise.withResolvers<void>()
+    let serverModel = "test/test-model/default"
+    await using fixture = makeACPFixture({
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_ordered_model") })
+        }
+        if (request.method !== "POST" || request.path !== "/api/session/ses_ordered_model/model") return undefined
+        if (serverModel === "test/test-model/default") {
+          serverModel = "test/second-model"
+          modelStarted.resolve()
+          return releaseModel.promise.then(() => new Response(null, { status: 204 }))
+        }
+        serverModel = "test/second-model/medium"
+        return new Response(null, { status: 204 })
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    const model = fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "model",
+      value: "test/second-model",
+    })
+    await modelStarted.promise
+    const effort = fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "medium",
+    })
+    releaseModel.resolve()
+    await model
+    const result = await effort
+
+    expect(serverModel).toBe("test/second-model/medium")
+    expect(currentValue(result, "model")).toBe("test/second-model")
+    expect(currentValue(result, "effort")).toBe("medium")
+    expect(
+      fixture.requests
+        .filter((request) => request.path === "/api/session/ses_ordered_model/model")
+        .map((request) => request.body),
+    ).toEqual([
+      { model: { providerID: "test", id: "second-model" } },
+      { model: { providerID: "test", id: "second-model", variant: "medium" } },
+    ])
+  })
+
+  test("orders overlapping config-mode and setSessionMode switches for one session", async () => {
+    const configModeStarted = Promise.withResolvers<void>()
+    const releaseConfigMode = Promise.withResolvers<void>()
+    let serverMode = "build"
+    await using fixture = makeACPFixture({
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_ordered_mode") })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_ordered_mode/model") {
+          return new Response(null, { status: 204 })
+        }
+        if (request.method !== "POST" || request.path !== "/api/session/ses_ordered_mode/agent") return undefined
+        if (serverMode !== "build") {
+          serverMode = "build"
+          return new Response(null, { status: 204 })
+        }
+        serverMode = "plan"
+        configModeStarted.resolve()
+        return releaseConfigMode.promise.then(() => new Response(null, { status: 204 }))
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    const configMode = fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "mode",
+      value: "plan",
+    })
+    await configModeStarted.promise
+    const mode = fixture.service.setSessionMode({ sessionId: session.sessionId, modeId: "build" })
+    releaseConfigMode.resolve()
+    await Promise.all([configMode, mode])
+    const result = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "model",
+      value: "test/test-model",
+    })
+
+    expect(serverMode).toBe("build")
+    expect(currentValue(result, "mode")).toBe("build")
+    expect(
+      fixture.requests
+        .filter((request) => request.path === "/api/session/ses_ordered_mode/agent")
+        .map((request) => request.body),
+    ).toEqual([{ agent: "plan" }, { agent: "build" }])
+  })
+
   test("converts MCP configs and deduplicates registrations per session and config", async () => {
     const local: McpServer = {
       name: "tools",

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

@@ -30,6 +30,8 @@ type FixtureHandler = (
 
 type FixtureOptions = {
   readonly fetch?: FixtureHandler
+  readonly clientFetch?: (...args: Parameters<typeof fetch>) => ReturnType<typeof fetch>
+  readonly sessionUpdate?: AgentSideConnection["sessionUpdate"]
   readonly models?: readonly ModelInfo[]
   readonly defaultModel?: ModelInfo
   readonly agents?: readonly AgentInfo[]
@@ -127,13 +129,15 @@ export function makeACPFixture(options: FixtureOptions = {}) {
   const requests: FixtureRequest[] = []
   const updates: Parameters<AgentSideConnection["sessionUpdate"]>[0][] = []
   const encoder = new TextEncoder()
-  let eventController: ReadableStreamDefaultController<Uint8Array> | undefined
+  const eventControllers = new Set<ReadableStreamDefaultController<Uint8Array>>()
   const models = options.models ?? [testModel, secondModel]
   const context: FixtureContext = {
     requests,
     send(event) {
-      if (!eventController) throw new Error("ACP fixture has no active event stream")
-      eventController.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`))
+      if (eventControllers.size === 0) throw new Error("ACP fixture has no active event stream")
+      eventControllers.forEach((controller) => {
+        controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`))
+      })
     },
   }
   const server = Bun.serve({
@@ -158,11 +162,11 @@ export function makeACPFixture(options: FixtureOptions = {}) {
           new ReadableStream<Uint8Array>({
             start(value) {
               controller = value
-              eventController = value
+              eventControllers.add(value)
               context.send({ id: "evt_connected", type: "server.connected", data: {} })
             },
             cancel() {
-              if (eventController === controller) eventController = undefined
+              if (controller) eventControllers.delete(controller)
             },
           }),
           { headers: { "content-type": "text/event-stream" } },
@@ -185,10 +189,14 @@ export function makeACPFixture(options: FixtureOptions = {}) {
     },
   })
   const service = ACPService.make({
-    client: OpenCode.make({ baseUrl: server.url.toString() }),
+    client: OpenCode.make({
+      baseUrl: server.url.toString(),
+      ...(options.clientFetch ? { fetch: Object.assign(options.clientFetch, { preconnect: fetch.preconnect }) } : {}),
+    }),
     connection: {
       sessionUpdate: async (update) => {
         updates.push(update)
+        await options.sessionUpdate?.(update)
       },
       requestPermission: async () => ({ outcome: { outcome: "cancelled" } }),
     },
@@ -198,8 +206,10 @@ export function makeACPFixture(options: FixtureOptions = {}) {
     service,
     requests,
     updates,
+    send: (event: unknown) => context.send(event),
     async [Symbol.asyncDispose]() {
-      eventController?.close()
+      eventControllers.forEach((controller) => controller.close())
+      eventControllers.clear()
       await server.stop(true)
     },
   }

packages/cli/test/acp/service-lifecycle.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, test } from "bun:test"
 import type { SessionConfigOption } from "@agentclientprotocol/sdk"
 import { makeACPFixture, makeSession, secondModel } from "./service-fixture"
+import { withTimeout } from "./sse-fixture"
 
 describe("acp service lifecycle", () => {
   test("does not persist the first catalog variant when no explicit default exists", async () => {
@@ -172,6 +173,833 @@ describe("acp service lifecycle", () => {
     })
   })
 
+  test("loads authoritative config after an overlapping switch completes", async () => {
+    const switchStarted = Promise.withResolvers<void>()
+    const releaseSwitch = Promise.withResolvers<void>()
+    const nativeFetch = fetch
+    let switching = true
+    let serverModel = makeSession("server").model
+    await using fixture = makeACPFixture({
+      clientFetch(input, init) {
+        const url = new URL(input instanceof Request ? input.url : input.toString())
+        if (init?.method !== "GET" || url.pathname !== "/api/session/ses_load_during_switch") {
+          return nativeFetch(input, init)
+        }
+        return Promise.resolve(Response.json({ data: makeSession("ses_load_during_switch", { model: serverModel }) }))
+      },
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_load_during_switch", { model: serverModel }) })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_load_during_switch/model") {
+          if (!switching) return new Response(null, { status: 204 })
+          switching = false
+          switchStarted.resolve()
+          return releaseSwitch.promise.then(() => {
+            serverModel = { providerID: "test", id: secondModel.id }
+            return new Response(null, { status: 204 })
+          })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_load_during_switch/message") {
+          return Response.json({ data: [], cursor: {} })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    const switched = fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "model",
+      value: "test/second-model",
+    })
+    await switchStarted.promise
+    const loaded = fixture.service.loadSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [] })
+    releaseSwitch.resolve()
+    await switched
+    const result = await loaded
+    const effort = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "medium",
+    })
+
+    expect(currentValue(result, "model")).toBe("test/second-model")
+    expect(currentValue(effort, "effort")).toBe("medium")
+  })
+
+  test("detaches and resumes after an overlapping switch completes", async () => {
+    const switchStarted = Promise.withResolvers<void>()
+    const releaseSwitch = Promise.withResolvers<void>()
+    const nativeFetch = fetch
+    let switching = true
+    let serverModel = makeSession("server").model
+    await using fixture = makeACPFixture({
+      clientFetch(input, init) {
+        const url = new URL(input instanceof Request ? input.url : input.toString())
+        if (init?.method !== "GET" || url.pathname !== "/api/session/ses_resume_during_switch") {
+          return nativeFetch(input, init)
+        }
+        return Promise.resolve(Response.json({ data: makeSession("ses_resume_during_switch", { model: serverModel }) }))
+      },
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_resume_during_switch", { model: serverModel }) })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_resume_during_switch/model") {
+          if (!switching) return new Response(null, { status: 204 })
+          switching = false
+          switchStarted.resolve()
+          return releaseSwitch.promise.then(() => {
+            serverModel = { providerID: "test", id: secondModel.id }
+            return new Response(null, { status: 204 })
+          })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_resume_during_switch/interrupt") {
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    const switched = fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "model",
+      value: "test/second-model",
+    })
+    await switchStarted.promise
+    const closed = fixture.service.closeSession({ sessionId: session.sessionId })
+    const resumed = fixture.service.resumeSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [] })
+    releaseSwitch.resolve()
+    await Promise.all([switched, closed])
+    const result = await resumed
+    const effort = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "medium",
+    })
+
+    expect(currentValue(result, "model")).toBe("test/second-model")
+    expect(currentValue(effort, "effort")).toBe("medium")
+  })
+
+  test("forks from authoritative parent config after an overlapping switch completes", async () => {
+    const switchStarted = Promise.withResolvers<void>()
+    const releaseSwitch = Promise.withResolvers<void>()
+    const nativeFetch = fetch
+    let serverModel = makeSession("server").model
+    await using fixture = makeACPFixture({
+      clientFetch(input, init) {
+        const url = new URL(input instanceof Request ? input.url : input.toString())
+        if (init?.method !== "POST" || url.pathname !== "/api/session/ses_fork_parent/fork") {
+          return nativeFetch(input, init)
+        }
+        return Promise.resolve(
+          Response.json({ data: makeSession("ses_fork_child", { model: serverModel, agent: "build" }) }),
+        )
+      },
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_fork_parent", { model: serverModel }) })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_fork_parent/model") {
+          switchStarted.resolve()
+          return releaseSwitch.promise.then(() => {
+            serverModel = { providerID: "test", id: secondModel.id }
+            return new Response(null, { status: 204 })
+          })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_fork_child/message") {
+          return Response.json({ data: [], cursor: {} })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    const switched = fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "model",
+      value: "test/second-model",
+    })
+    await switchStarted.promise
+    const forked = fixture.service.forkSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [] })
+    releaseSwitch.resolve()
+    await switched
+    const result = await forked
+
+    expect(result.sessionId).toBe("ses_fork_child")
+    expect(currentValue(result, "model")).toBe("test/second-model")
+  })
+
+  test("keeps the prior attachment when staged MCP, command, or replay setup fails", async () => {
+    let phase: "mcp" | "commands" | "replay" | "success" = "mcp"
+    await using fixture = makeACPFixture({
+      sessionUpdate: async () => {
+        if (phase === "commands") throw new Error("command publication failed")
+      },
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_attach_transaction") })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_attach_transaction") {
+          return Response.json({
+            data: makeSession("ses_attach_transaction", {
+              model: { providerID: secondModel.providerID, id: secondModel.id },
+            }),
+          })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_attach_transaction/model") {
+          return new Response(null, { status: 204 })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_attach_transaction/message") {
+          if (phase === "replay") return new Response(null, { status: 500 })
+          return Response.json({ data: [], cursor: {} })
+        }
+        if (request.method === "PUT" && request.path === "/api/mcp/docs" && phase === "mcp") {
+          return new Response(null, { status: 500 })
+        }
+        if (request.method === "PUT" && request.path.startsWith("/api/mcp/")) {
+          return new Response(null, { status: 204 })
+        }
+        if (request.method === "DELETE" && request.path.startsWith("/api/mcp/")) {
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    const tools = { name: "tools", command: "bun", args: ["tools.ts"], env: [] }
+    const docs = { name: "docs", command: "bun", args: ["docs.ts"], env: [] }
+
+    const mcpFailure = await fixture.service
+      .loadSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [tools, docs] })
+      .catch((error: unknown) => error)
+    const afterMcpFailure = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "high",
+    })
+
+    phase = "commands"
+    const commandFailure = await fixture.service
+      .resumeSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [tools] })
+      .catch((error: unknown) => error)
+    const afterCommandFailure = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "default",
+    })
+
+    phase = "replay"
+    const replayFailure = await fixture.service
+      .loadSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [tools] })
+      .catch((error: unknown) => error)
+    const afterReplayFailure = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "high",
+    })
+
+    phase = "success"
+    const attached = await fixture.service.resumeSession({
+      cwd: "/workspace",
+      sessionId: session.sessionId,
+      mcpServers: [tools],
+    })
+
+    expect([mcpFailure, commandFailure, replayFailure]).toEqual([
+      expect.any(Error),
+      expect.any(Error),
+      expect.any(Error),
+    ])
+    expect(currentValue(afterMcpFailure, "model")).toBe("test/test-model")
+    expect(currentValue(afterCommandFailure, "model")).toBe("test/test-model")
+    expect(currentValue(afterReplayFailure, "model")).toBe("test/test-model")
+    expect(currentValue(attached, "model")).toBe("test/second-model")
+    expect(
+      fixture.requests
+        .filter((request) => request.path.startsWith("/api/mcp/"))
+        .map((request) => `${request.method} ${request.path}`),
+    ).toEqual([
+      "PUT /api/mcp/tools",
+      "PUT /api/mcp/docs",
+      "DELETE /api/mcp/tools",
+      "DELETE /api/mcp/docs",
+      "PUT /api/mcp/tools",
+      "DELETE /api/mcp/tools",
+      "PUT /api/mcp/tools",
+    ])
+  })
+
+  test("serializes MCP attachment transactions for sessions in the same location", async () => {
+    const firstStaged = Promise.withResolvers<void>()
+    const releaseFirst = Promise.withResolvers<void>()
+    let racing = false
+    let created = 0
+    let installed: unknown
+    await using fixture = makeACPFixture({
+      sessionUpdate: async (update) => {
+        if (!racing || update.sessionId !== "ses_location_1") return
+        firstStaged.resolve()
+        await releaseFirst.promise
+        throw new Error("first publication failed")
+      },
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          created++
+          return Response.json({ data: makeSession(`ses_location_${created}`) })
+        }
+        if (request.method === "GET" && request.path.startsWith("/api/session/ses_location_")) {
+          return Response.json({ data: makeSession(request.path.split("/").at(-1) ?? "missing") })
+        }
+        if (request.method === "PUT" && request.path === "/api/mcp/shared") {
+          installed = request.body
+          return new Response(null, { status: 204 })
+        }
+        if (request.method === "DELETE" && request.path === "/api/mcp/shared") {
+          installed = undefined
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const first = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    const second = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    racing = true
+
+    const failed = fixture.service
+      .resumeSession({
+        cwd: "/workspace",
+        sessionId: first.sessionId,
+        mcpServers: [{ name: "shared", command: "bun", args: ["first.ts"], env: [] }],
+      })
+      .catch((error: unknown) => error)
+    await firstStaged.promise
+    const succeeded = fixture.service.resumeSession({
+      cwd: "/workspace",
+      sessionId: second.sessionId,
+      mcpServers: [{ name: "shared", command: "bun", args: ["second.ts"], env: [] }],
+    })
+    releaseFirst.resolve()
+
+    expect(await failed).toBeInstanceOf(Error)
+    await succeeded
+    expect(installed).toEqual({
+      config: { type: "local", command: ["bun", "second.ts"], environment: {} },
+    })
+    expect(
+      fixture.requests
+        .filter((request) => request.path === "/api/mcp/shared")
+        .map((request) => ({ method: request.method, body: request.body })),
+    ).toEqual([
+      {
+        method: "PUT",
+        body: { config: { type: "local", command: ["bun", "first.ts"], environment: {} } },
+      },
+      { method: "DELETE", body: undefined },
+      {
+        method: "PUT",
+        body: { config: { type: "local", command: ["bun", "second.ts"], environment: {} } },
+      },
+    ])
+  })
+
+  test("invalidates a location when MCP rollback fails", async () => {
+    let failing = false
+    let created = 0
+    await using fixture = makeACPFixture({
+      sessionUpdate: async (update) => {
+        if (failing && update.sessionId === "ses_rollback_1") throw new Error("command publication failed")
+      },
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          created++
+          return Response.json({ data: makeSession(`ses_rollback_${created}`) })
+        }
+        if (request.method === "GET" && request.path.startsWith("/api/session/ses_rollback_")) {
+          return Response.json({ data: makeSession(request.path.split("/").at(-1) ?? "missing") })
+        }
+        if (request.method === "PUT" && request.path === "/api/mcp/tools") {
+          return new Response(null, { status: 204 })
+        }
+        if (request.method === "DELETE" && request.path === "/api/mcp/tools") {
+          return new Response(null, { status: failing ? 500 : 204 })
+        }
+        if (request.method === "POST" && request.path.endsWith("/model")) {
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const first = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    const second = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    failing = true
+
+    const failure = await fixture.service
+      .resumeSession({
+        cwd: "/workspace",
+        sessionId: first.sessionId,
+        mcpServers: [{ name: "tools", command: "bun", args: ["tools.ts"], env: [] }],
+      })
+      .catch((error: unknown) => error)
+    const missing = await Promise.all(
+      [first, second].map((session) =>
+        fixture.service
+          .setSessionConfigOption({ sessionId: session.sessionId, configId: "effort", value: "high" })
+          .catch((error: unknown) => error),
+      ),
+    )
+
+    expect(failure).toBeInstanceOf(AggregateError)
+    expect(failure).toMatchObject({ message: "ACP attachment failed and MCP rollback did not complete" })
+    expect(missing).toEqual([
+      expect.objectContaining({ _tag: "ACPSessionNotFoundError" }),
+      expect.objectContaining({ _tag: "ACPSessionNotFoundError" }),
+    ])
+
+    failing = false
+    const recovered = await fixture.service.resumeSession({
+      cwd: "/workspace",
+      sessionId: first.sessionId,
+      mcpServers: [],
+    })
+
+    expect(currentValue(recovered, "model")).toBe("test/test-model")
+    expect(
+      fixture.requests.filter((request) => request.path === "/api/mcp/tools").map((request) => request.method),
+    ).toEqual(["PUT", "DELETE", "DELETE"])
+  })
+
+  test("allows MCP attachment transactions in distinct locations to proceed concurrently", async () => {
+    const firstStaged = Promise.withResolvers<void>()
+    const releaseFirst = Promise.withResolvers<void>()
+    let racing = false
+    let created = 0
+    await using fixture = makeACPFixture({
+      sessionUpdate: async (update) => {
+        if (!racing || update.sessionId !== "ses_distinct_1") return
+        firstStaged.resolve()
+        await releaseFirst.promise
+      },
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          created++
+          const cwd = created === 1 ? "/first" : "/second"
+          return Response.json({ data: makeSession(`ses_distinct_${created}`, { cwd }) })
+        }
+        if (request.method === "GET" && request.path.startsWith("/api/session/ses_distinct_")) {
+          const id = request.path.split("/").at(-1) ?? "missing"
+          return Response.json({ data: makeSession(id, { cwd: id.endsWith("1") ? "/first" : "/second" }) })
+        }
+        if (request.method === "PUT" && request.path === "/api/mcp/shared") {
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const first = await fixture.service.newSession({ cwd: "/first", mcpServers: [] })
+    const second = await fixture.service.newSession({ cwd: "/second", mcpServers: [] })
+    racing = true
+
+    const blocked = fixture.service.resumeSession({
+      cwd: "/first",
+      sessionId: first.sessionId,
+      mcpServers: [{ name: "shared", command: "bun", args: ["first.ts"], env: [] }],
+    })
+    await firstStaged.promise
+    const concurrent = fixture.service.resumeSession({
+      cwd: "/second",
+      sessionId: second.sessionId,
+      mcpServers: [{ name: "shared", command: "bun", args: ["second.ts"], env: [] }],
+    })
+
+    await withTimeout(concurrent, "distinct location transaction was blocked")
+    releaseFirst.resolve()
+    await blocked
+  })
+
+  test("allows config switches for distinct sessions to proceed concurrently", async () => {
+    const firstStarted = Promise.withResolvers<void>()
+    const releaseFirst = Promise.withResolvers<void>()
+    let created = 0
+    await using fixture = makeACPFixture({
+      fetch(request) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          created++
+          return Response.json({ data: makeSession(`ses_concurrent_${created}`) })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_concurrent_1/model") {
+          firstStarted.resolve()
+          return releaseFirst.promise.then(() => new Response(null, { status: 204 }))
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_concurrent_2/model") {
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const first = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    const second = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    const blocked = fixture.service.setSessionConfigOption({
+      sessionId: first.sessionId,
+      configId: "effort",
+      value: "high",
+    })
+    await firstStarted.promise
+    const concurrent = await fixture.service.setSessionConfigOption({
+      sessionId: second.sessionId,
+      configId: "effort",
+      value: "high",
+    })
+    releaseFirst.resolve()
+    await blocked
+
+    expect(currentValue(concurrent, "effort")).toBe("high")
+  })
+
+  test("load replacement cancels prompt ownership acquired before streaming", async () => {
+    await using fixture = makeACPFixture({
+      fetch(request, context) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_prompt_load") })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_prompt_load") {
+          return Response.json({ data: makeSession("ses_prompt_load") })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_prompt_load/message") {
+          return Response.json({ data: [], cursor: {} })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_prompt_load/prompt") {
+          const id = requestField(request.body, "id")
+          context.send({
+            id: `evt_${id}`,
+            type: "session.inbox.delivered",
+            data: { sessionID: "ses_prompt_load", inboxID: id },
+          })
+          if (requestField(request.body, "text") === "second") {
+            context.send({
+              id: "evt_second_complete",
+              type: "session.execution.succeeded",
+              data: { sessionID: "ses_prompt_load" },
+            })
+          }
+          return Response.json({ data: {} })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    const prompt = fixture.service.prompt({
+      sessionId: session.sessionId,
+      prompt: [{ type: "text", text: "first" }],
+    })
+    const stoppedPrompt = prompt.then(
+      () => "resolved",
+      () => "rejected",
+    )
+    const loaded = fixture.service.loadSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [] })
+    await loaded
+    const stopped = await withTimeout(stoppedPrompt, "replaced prompt did not stop")
+    const second = await fixture.service.prompt({
+      sessionId: session.sessionId,
+      prompt: [{ type: "text", text: "second" }],
+    })
+
+    expect(stopped).toBe("rejected")
+    expect(second.stopReason).toBe("end_turn")
+  })
+
+  test("replacement permits a new prompt while the retired request does not settle", async () => {
+    const firstStarted = Promise.withResolvers<void>()
+    const firstAborted = Promise.withResolvers<void>()
+    const nativeFetch = fetch
+    let prompts = 0
+    await using fixture = makeACPFixture({
+      clientFetch(input, init) {
+        const url = new URL(input instanceof Request ? input.url : input.toString())
+        if (init?.method !== "POST" || url.pathname !== "/api/session/ses_prompt_handoff/prompt") {
+          return nativeFetch(input, init)
+        }
+        prompts++
+        if (prompts > 1) return nativeFetch(input, init)
+        firstStarted.resolve()
+        init.signal?.addEventListener("abort", () => firstAborted.resolve(), { once: true })
+        return new Promise<Response>(() => {})
+      },
+      fetch(request, context) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_prompt_handoff") })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_prompt_handoff") {
+          return Response.json({ data: makeSession("ses_prompt_handoff") })
+        }
+        if (request.method === "GET" && request.path === "/api/session/ses_prompt_handoff/message") {
+          return Response.json({ data: [], cursor: {} })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_prompt_handoff/prompt") {
+          const id = requestField(request.body, "id")
+          context.send({
+            id: `evt_${id}`,
+            type: "session.inbox.delivered",
+            data: { sessionID: "ses_prompt_handoff", inboxID: id },
+          })
+          context.send({
+            id: "evt_handoff_complete",
+            type: "session.execution.succeeded",
+            data: { sessionID: "ses_prompt_handoff" },
+          })
+          return Response.json({ data: {} })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    void fixture.service
+      .prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "never settles" }] })
+      .catch(() => undefined)
+    await firstStarted.promise
+
+    await fixture.service.loadSession({ cwd: "/workspace", sessionId: session.sessionId, mcpServers: [] })
+    await firstAborted.promise
+    const current = await fixture.service.prompt({
+      sessionId: session.sessionId,
+      prompt: [{ type: "text", text: "new owner" }],
+    })
+
+    expect(current.stopReason).toBe("end_turn")
+    expect(prompts).toBe(2)
+  })
+
+  test("cancel permits a new prompt while the cancelled request does not settle", async () => {
+    const firstStarted = Promise.withResolvers<void>()
+    const firstAborted = Promise.withResolvers<void>()
+    const nativeFetch = fetch
+    let prompts = 0
+    await using fixture = makeACPFixture({
+      clientFetch(input, init) {
+        const url = new URL(input instanceof Request ? input.url : input.toString())
+        if (init?.method !== "POST" || url.pathname !== "/api/session/ses_prompt_cancel_handoff/prompt") {
+          return nativeFetch(input, init)
+        }
+        prompts++
+        if (prompts > 1) return nativeFetch(input, init)
+        firstStarted.resolve()
+        init.signal?.addEventListener("abort", () => firstAborted.resolve(), { once: true })
+        return new Promise<Response>(() => {})
+      },
+      fetch(request, context) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_prompt_cancel_handoff") })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_prompt_cancel_handoff/interrupt") {
+          return Response.json({ interrupted: true })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_prompt_cancel_handoff/prompt") {
+          const id = requestField(request.body, "id")
+          context.send({
+            id: `evt_${id}`,
+            type: "session.inbox.delivered",
+            data: { sessionID: "ses_prompt_cancel_handoff", inboxID: id },
+          })
+          context.send({
+            id: "evt_cancel_handoff_complete",
+            type: "session.execution.succeeded",
+            data: { sessionID: "ses_prompt_cancel_handoff" },
+          })
+          return Response.json({ data: {} })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    void fixture.service
+      .prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "never settles" }] })
+      .catch(() => undefined)
+    await firstStarted.promise
+
+    await fixture.service.cancel({ sessionId: session.sessionId })
+    await firstAborted.promise
+    const current = await fixture.service.prompt({
+      sessionId: session.sessionId,
+      prompt: [{ type: "text", text: "new owner" }],
+    })
+
+    expect(current.stopReason).toBe("end_turn")
+    expect(prompts).toBe(2)
+  })
+
+  test("failed authoritative interruption retains sole prompt ownership", async () => {
+    const admitted = Promise.withResolvers<void>()
+    const afterFailure = Promise.withResolvers<void>()
+    let interruptFails = true
+    let prompts = 0
+    await using fixture = makeACPFixture({
+      sessionUpdate: async (update) => {
+        if (
+          update.update.sessionUpdate === "agent_message_chunk" &&
+          update.update.content.type === "text" &&
+          update.update.content.text === "before cancel"
+        ) {
+          admitted.resolve()
+        }
+        if (
+          update.update.sessionUpdate === "agent_message_chunk" &&
+          update.update.content.type === "text" &&
+          update.update.content.text === "after failure"
+        ) {
+          afterFailure.resolve()
+        }
+      },
+      fetch(request, context) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_interrupt_owner") })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_interrupt_owner/prompt") {
+          prompts++
+          const id = requestField(request.body, "id")
+          context.send({
+            id: `evt_${id}`,
+            type: "session.inbox.delivered",
+            data: { sessionID: "ses_interrupt_owner", inboxID: id },
+          })
+          context.send({
+            id: `evt_text_${prompts}`,
+            type: "session.text.delta",
+            data: {
+              sessionID: "ses_interrupt_owner",
+              assistantMessageID: `msg_assistant_${prompts}`,
+              ordinal: 0,
+              delta: prompts === 1 ? "before cancel" : "replacement",
+            },
+          })
+          if (prompts > 1) {
+            context.send({
+              id: "evt_replacement_complete",
+              type: "session.execution.succeeded",
+              data: { sessionID: "ses_interrupt_owner" },
+            })
+          }
+          return Response.json({ data: {} })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_interrupt_owner/interrupt") {
+          if (interruptFails) return new Response(null, { status: 500 })
+          context.send({
+            id: "evt_owner_interrupted",
+            type: "session.execution.interrupted",
+            data: { sessionID: "ses_interrupt_owner", reason: "user" },
+          })
+          return Response.json({ interrupted: true })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    const first = fixture.service
+      .prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "first" }] })
+      .catch((error: unknown) => error)
+    await admitted.promise
+
+    const interruptionFailure = await fixture.service.cancel({ sessionId: session.sessionId }).catch((error) => error)
+    const replacementFailure = await fixture.service
+      .prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "must wait" }] })
+      .catch((error: unknown) => error)
+    fixture.updates.length = 0
+    fixture.send({
+      id: "evt_after_failed_interrupt",
+      type: "session.text.delta",
+      data: {
+        sessionID: "ses_interrupt_owner",
+        assistantMessageID: "msg_assistant_1",
+        ordinal: 1,
+        delta: "after failure",
+      },
+    })
+    await afterFailure.promise
+    const activeStream = fixture.requests.filter((request) => request.path === "/api/event").length
+
+    expect(interruptionFailure).toBeInstanceOf(Error)
+    expect(replacementFailure).toMatchObject({ _tag: "ACPServiceFailureError" })
+    expect(activeStream).toBe(1)
+    expect(
+      fixture.updates.flatMap((update) =>
+        update.update.sessionUpdate === "agent_message_chunk" && update.update.content.type === "text"
+          ? [update.update.content.text]
+          : [],
+      ),
+    ).toEqual(["after failure"])
+
+    interruptFails = false
+    await fixture.service.cancel({ sessionId: session.sessionId })
+    await first
+    const replacement = await fixture.service.prompt({
+      sessionId: session.sessionId,
+      prompt: [{ type: "text", text: "replacement" }],
+    })
+
+    expect(replacement.stopReason).toBe("end_turn")
+    expect(prompts).toBe(2)
+    expect(
+      fixture.updates.flatMap((update) =>
+        update.update.sessionUpdate === "agent_message_chunk" && update.update.content.type === "text"
+          ? [update.update.content.text]
+          : [],
+      ),
+    ).toEqual(["after failure", "replacement"])
+  })
+
+  test("delete detaches and cancels an active foreground prompt", async () => {
+    const promptStarted = Promise.withResolvers<void>()
+    await using fixture = makeACPFixture({
+      fetch(request, context) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_prompt_delete") })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_prompt_delete/prompt") {
+          const id = requestField(request.body, "id")
+          context.send({
+            id: `evt_${id}`,
+            type: "session.inbox.delivered",
+            data: { sessionID: "ses_prompt_delete", inboxID: id },
+          })
+          promptStarted.resolve()
+          return Response.json({ data: {} })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_prompt_delete/model") {
+          return new Response(null, { status: 204 })
+        }
+        if (request.method === "DELETE" && request.path === "/api/session/ses_prompt_delete") {
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+    const prompt = fixture.service.prompt({
+      sessionId: session.sessionId,
+      prompt: [{ type: "text", text: "running" }],
+    })
+    const stoppedPrompt = prompt.then(
+      () => "resolved",
+      () => "rejected",
+    )
+    await promptStarted.promise
+
+    const configured = await fixture.service.setSessionConfigOption({
+      sessionId: session.sessionId,
+      configId: "effort",
+      value: "high",
+    })
+    await fixture.service.deleteSession({ sessionId: session.sessionId })
+    const stopped = await withTimeout(stoppedPrompt, "deleted session prompt did not stop")
+
+    expect(currentValue(configured, "effort")).toBe("high")
+    expect(stopped).toBe("rejected")
+  })
+
   test("lists server-backed pages and forwards cwd and cursor", async () => {
     const firstPage = Array.from({ length: 100 }, (_, index) =>
       makeSession(`ses_${100 - index}`, {
@@ -251,11 +1079,7 @@ describe("acp service lifecycle", () => {
     expect(await fixture.service.closeSession({ sessionId: "missing" })).toEqual({})
     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/missing/interrupt",
-    ])
+    ).toEqual(["/api/session/ses_lifecycle/interrupt", "/api/session/missing/interrupt"])
   })
 
   test("deletes sessions from backing and local storage", async () => {
@@ -289,3 +1113,10 @@ describe("acp service lifecycle", () => {
 function currentValue(result: { readonly configOptions?: readonly SessionConfigOption[] | null }, id: string) {
   return result.configOptions?.find((option) => option.id === id)?.currentValue
 }
+
+function requestField(value: unknown, key: string) {
+  if (!value || typeof value !== "object") throw new Error(`Missing request ${key}`)
+  const field = Reflect.get(value, key)
+  if (typeof field !== "string") throw new Error(`Missing request ${key}`)
+  return field
+}

packages/cli/test/acp/service-usage.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, test } from "bun:test"
 import { makeACPFixture, makeSession, secondModel, type FixtureContext, type FixtureRequest } from "./service-fixture"
+import { withTimeout } from "./sse-fixture"
 
 describe("acp service prompt routing and usage", () => {
   test("routes slash commands, skills, and compact through their session endpoints", async () => {
@@ -69,6 +70,87 @@ describe("acp service prompt routing and usage", () => {
     expect(fixture.requests.some((request) => request.path === "/api/session/ses_routes/prompt")).toBe(false)
   })
 
+  test("forwards admission signals through every prompt route", async () => {
+    const nativeFetch = fetch
+    const signalled: string[] = []
+    await using fixture = makeACPFixture({
+      clientFetch(input, init) {
+        const url = new URL(input instanceof Request ? input.url : input.toString())
+        if (
+          ["synthetic", "prompt", "skill", "compact", "command"].some((suffix) => url.pathname.endsWith(`/${suffix}`))
+        ) {
+          if (init?.signal instanceof AbortSignal) signalled.push(url.pathname.split("/").at(-1) ?? "")
+        }
+        return nativeFetch(input, init)
+      },
+      fetch(request, context) {
+        if (request.method === "POST" && request.path === "/api/session") {
+          return Response.json({ data: makeSession("ses_signals") })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_signals/synthetic") {
+          return Response.json({ data: {} })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_signals/prompt") {
+          const id = requestID(request)
+          completeTurn(context, "ses_signals", {
+            id: id.replace(/^msg_/, "evt_"),
+            type: "session.inbox.delivered",
+            data: { sessionID: "ses_signals", inboxID: id },
+          })
+          return Response.json({ data: {} })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_signals/skill") {
+          const id = requestID(request)
+          completeTurn(context, "ses_signals", {
+            id: id.replace(/^msg_/, "evt_"),
+            type: "session.skill.activated",
+            data: { sessionID: "ses_signals", skill: "verify" },
+          })
+          return new Response(null, { status: 204 })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_signals/compact") {
+          const id = requestID(request)
+          completeTurn(context, "ses_signals", {
+            id: `evt_${id}`,
+            type: "session.inbox.delivered",
+            data: { sessionID: "ses_signals", inboxID: id },
+          })
+          return Response.json({ data: {} })
+        }
+        if (request.method === "POST" && request.path === "/api/session/ses_signals/command") {
+          return new Response(null, { status: 204 })
+        }
+        return undefined
+      },
+    })
+    const session = await fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })
+
+    await withTimeout(
+      fixture.service.prompt({
+        sessionId: session.sessionId,
+        prompt: [
+          { type: "text", text: "context", annotations: { audience: ["assistant"] } },
+          { type: "text", text: "hello" },
+        ],
+      }),
+      "synthetic prompt did not complete",
+    )
+    await withTimeout(
+      fixture.service.prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "/verify" }] }),
+      "skill prompt did not complete",
+    )
+    await withTimeout(
+      fixture.service.prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "/compact" }] }),
+      "compact prompt did not complete",
+    )
+    await withTimeout(
+      fixture.service.prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "/review" }] }),
+      "command prompt did not complete",
+    )
+
+    expect(signalled).toEqual(["synthetic", "prompt", "skill", "compact", "command"])
+  })
+
   test("returns turn usage and publishes current context usage with cumulative session cost", async () => {
     const assistantTokens = {
       input: 100,

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

@@ -166,6 +166,7 @@ export function createSseFixture(options: FixtureOptions = {}) {
     messages,
     requests,
     send,
+    streamCount: () => streams.size,
     async stop() {
       for (const stream of streams) {
         try {