feat(acp): follow model and agent changes from other clients (#52499)

packages/cli/src/acp/service.ts

@@ -99,6 +99,7 @@ export function make(input: {
     }
   })
 
+  // Both selectors update the selection before switching on the server, so the echoed event diffs to no change.
   const selectModel = Effect.fnUntraced(function* (attached: Attached, model: ModelRef) {
     yield* Ref.update(attached.selection, (selection) => ({ ...selection, model }))
     yield* ACPPromise.promise(() => input.client.session.switchModel({ sessionID: attached.id, model }))

packages/cli/src/acp/sessions.ts

@@ -1,7 +1,7 @@
 import { isDeepStrictEqual } from "node:util"
 import type { McpServer, RequestError } from "@agentclientprotocol/sdk"
-import type { OpenCodeClient, SessionInfo } from "@opencode/client/promise"
-import { Context, Effect, Exit, Ref, Scope, Stream } from "effect"
+import type { OpenCodeClient, OpenCodeEvent, SessionInfo } from "@opencode/client/promise"
+import { Context, Deferred, Effect, Exit, Queue, Ref, Scope, Stream } from "effect"
 import type { ACPCatalog, Catalog } from "./catalog"
 import { availableCommands, configOptions, type Selection } from "./config-option"
 import type { ACPConnection } from "./connection"
@@ -34,7 +34,14 @@ export interface Interface {
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/cli/acp/Sessions") {}
 
-type Entry = { readonly attached: Attached; readonly scope: Scope.Closeable }
+type Entry = {
+  readonly attached: Attached
+  readonly scope: Scope.Closeable
+  /** Selection changes from other clients, applied by the session's fold. */
+  readonly selected: Queue.Queue<Selection>
+}
+
+type SelectedEvent = Extract<OpenCodeEvent, { type: "session.model.selected" | "session.agent.selected" }>
 
 export const make = Effect.fnUntraced(function* (input: {
   readonly client: OpenCodeClient
@@ -45,6 +52,27 @@ export const make = Effect.fnUntraced(function* (input: {
   const sessions = new Map<string, Entry>()
   // Kept across re-attachment so resuming with the same servers does not add them again.
   const registeredMcp = new Map<string, Set<string>>()
+  const connected = yield* Deferred.make<void>()
+
+  // Subscribe before any attach so a switch right after `sessions.set` reaches the session.
+  yield* Stream.fromAsyncIterable(input.client.event.subscribe(), (cause) => cause).pipe(
+    Stream.tap((event) => (event.type === "server.connected" ? Deferred.succeed(connected, undefined) : Effect.void)),
+    Stream.filter(
+      (event): event is SelectedEvent =>
+        event.type === "session.model.selected" || event.type === "session.agent.selected",
+    ),
+    Stream.runForEach((event) => {
+      const entry = sessions.get(event.data.sessionID)
+      if (!entry) return Effect.void
+      return Queue.offer(
+        entry.selected,
+        event.type === "session.model.selected" ? { model: event.data.model } : { modeID: event.data.agent },
+      )
+    }),
+    Effect.ignore,
+    Effect.ensuring(Deferred.succeed(connected, undefined)),
+    Effect.forkScoped,
+  )
 
   const sendCommands = (sessionID: string, catalog: Catalog) =>
     input.connection.sessionUpdate({
@@ -52,9 +80,9 @@ export const make = Effect.fnUntraced(function* (input: {
       update: { sessionUpdate: "available_commands_update", availableCommands: availableCommands(catalog) },
     })
 
-  const changed = Effect.fnUntraced(function* (attached: Attached, previous: Catalog, next: Catalog) {
-    const selection = yield* Ref.get(attached.selection)
-    const options = configOptions(next, selection)
+  const changed = Effect.fnUntraced(function* (attached: Attached, previous: Catalog, next: Catalog, patch: Selection) {
+    const selection = yield* Ref.getAndUpdate(attached.selection, (current) => ({ ...current, ...patch }))
+    const options = configOptions(next, { ...selection, ...patch })
     if (!isDeepStrictEqual(options, configOptions(previous, selection))) {
       yield* input.connection.sessionUpdate({
         sessionId: attached.id,
@@ -98,6 +126,7 @@ export const make = Effect.fnUntraced(function* (input: {
 
   return Service.of({
     attach: Effect.fn("cli.acp.sessions.attach")(function* (session, cwd, mcpServers) {
+      yield* Deferred.await(connected)
       const current = yield* input.catalog.get(cwd)
       const abort = new AbortController()
       const entry: Entry = {
@@ -108,6 +137,7 @@ export const make = Effect.fnUntraced(function* (input: {
           signal: abort.signal,
         },
         scope: Scope.forkUnsafe(scope),
+        selected: yield* Queue.unbounded<Selection>(),
       }
       // Swap synchronously so concurrent attaches of one ID cannot both keep a scope.
       const replaced = sessions.get(session.id)
@@ -121,14 +151,18 @@ export const make = Effect.fnUntraced(function* (input: {
         Effect.andThen(sendCommands(session.id, current)),
         Effect.onError(() => remove(session.id, entry)),
       )
-      // `changes` emits the latest catalog first, so a reload since `current` is still pushed.
-      yield* input.catalog.changes(cwd).pipe(
+      // `changes` emits the latest catalog first, so a reload since `current` is still pushed. One fold applies
+      // catalog and selection changes so pushes leave the client on the latest pair.
+      yield* Stream.merge(
+        input.catalog.changes(cwd).pipe(Stream.map((catalog) => ({ catalog, patch: {} }))),
+        Stream.fromQueue(entry.selected).pipe(Stream.map((patch) => ({ catalog: undefined, patch }))),
+      ).pipe(
         Stream.runFoldEffect(
           () => current,
-          (previous, next) =>
-            next === previous
-              ? Effect.succeed(previous)
-              : changed(entry.attached, previous, next).pipe(Effect.ignore, Effect.as(next)),
+          (previous, step) => {
+            const next = step.catalog ?? previous
+            return changed(entry.attached, previous, next, step.patch).pipe(Effect.ignore, Effect.as(next))
+          },
         ),
         Effect.ignore,
         Effect.forkIn(entry.scope),

packages/cli/test/acp/selection.test.ts

@@ -0,0 +1,202 @@
+import { describe, expect, test } from "bun:test"
+import type { SessionConfigOption } from "@agentclientprotocol/sdk"
+import { currentValue, selectValues } from "./select-options"
+import {
+  durableEvent,
+  ephemeralEvent,
+  secondModel,
+  startSession,
+  startWire,
+  stepEnded,
+  testModel,
+  turn,
+  type Wire,
+} from "./wire-fixture"
+
+describe("acp follows model and agent selections from other clients", () => {
+  test("pushes an external model switch with its variant and uses it for the next prompt", async () => {
+    await using acp = await startSession({
+      onPrompt: ({ sessionID, id }) => turn(sessionID, id, stepEnded(sessionID, "msg_assistant")),
+    })
+
+    acp.server.send(
+      durableEvent("session.model.selected", {
+        sessionID: acp.sessionId,
+        model: { providerID: "test", id: secondModel.id, variant: "low" },
+        previous: { providerID: "test", id: "test-model" },
+      }),
+    )
+    const [options] = await optionUpdates(acp, acp.sessionId, 1)
+    await acp.prompt(acp.sessionId, "hello")
+
+    expect(values(options)).toEqual({ model: "test/second-model", effort: "low", mode: "build" })
+    expect(await acp.waitForUpdate((item) => item.update.sessionUpdate === "usage_update")).toMatchObject({
+      update: { size: secondModel.limit.context },
+    })
+  })
+
+  test("shows the default effort when the external switch carries no variant", async () => {
+    await using acp = await startSession()
+    await acp.request("session/set_config_option", { sessionId: acp.sessionId, configId: "effort", value: "high" })
+
+    acp.server.send(
+      durableEvent("session.model.selected", {
+        sessionID: acp.sessionId,
+        model: { providerID: "test", id: secondModel.id },
+      }),
+    )
+
+    expect(values((await optionUpdates(acp, acp.sessionId, 1))[0])).toEqual({
+      model: "test/second-model",
+      effort: "default",
+      mode: "build",
+    })
+  })
+
+  test("pushes an external agent switch as the mode", async () => {
+    await using acp = await startSession()
+
+    acp.server.send(
+      durableEvent("session.agent.selected", { sessionID: acp.sessionId, agent: "plan", previous: "build" }),
+    )
+
+    expect(values((await optionUpdates(acp, acp.sessionId, 1))[0])).toEqual({
+      model: "test/test-model",
+      effort: "default",
+      mode: "plan",
+    })
+  })
+
+  test("does not echo its own model and mode switches", async () => {
+    await using acp = await startSession({
+      // The server publishes the selection event before it answers the switch.
+      fetch: (request) => {
+        if (request.method !== "POST") return undefined
+        if (request.path.endsWith("/model"))
+          acp.server.send(
+            durableEvent("session.model.selected", {
+              sessionID: acp.sessionId,
+              model: { providerID: "test", id: secondModel.id, variant: "medium" },
+            }),
+          )
+        if (request.path.endsWith("/agent"))
+          acp.server.send(durableEvent("session.agent.selected", { sessionID: acp.sessionId, agent: "plan" }))
+        return undefined
+      },
+    })
+    const set = (configId: string, value: string) =>
+      acp.request("session/set_config_option", { sessionId: acp.sessionId, configId, value })
+
+    await set("model", "test/second-model/medium")
+    await acp.request("session/set_mode", { sessionId: acp.sessionId, modeId: "plan" })
+    acp.server.send(durableEvent("session.agent.selected", { sessionID: acp.sessionId, agent: "build" }))
+    const updates = await optionUpdates(acp, acp.sessionId, 1)
+
+    expect(updates.map(values)).toEqual([{ model: "test/second-model", effort: "medium", mode: "build" }])
+  })
+
+  test("ignores selections for other sessions", async () => {
+    await using acp = await startSession()
+    const other = await acp.newSession()
+
+    acp.server.send(durableEvent("session.agent.selected", { sessionID: "ses_unattached", agent: "plan" }))
+    acp.server.send(durableEvent("session.agent.selected", { sessionID: other.sessionId, agent: "plan" }))
+    acp.server.send(
+      durableEvent("session.model.selected", {
+        sessionID: acp.sessionId,
+        model: { providerID: "test", id: secondModel.id },
+      }),
+    )
+    const updates = await optionUpdates(acp, acp.sessionId, 1)
+
+    expect(updates.map(values)).toEqual([{ model: "test/second-model", effort: "default", mode: "build" }])
+  })
+
+  test("stops following a closed session and follows it once after re-attaching", async () => {
+    await using acp = await startSession()
+    const other = await acp.newSession()
+    await acp.request("session/close", { sessionId: acp.sessionId })
+
+    acp.server.send(durableEvent("session.agent.selected", { sessionID: acp.sessionId, agent: "plan" }))
+    acp.server.send(durableEvent("session.agent.selected", { sessionID: other.sessionId, agent: "plan" }))
+    await optionUpdates(acp, other.sessionId, 1)
+    expect(configUpdates(acp, acp.sessionId)).toEqual([])
+
+    const stored = acp.server.sessions.get(acp.sessionId)
+    if (!stored) throw new Error(`missing stored session ${acp.sessionId}`)
+    acp.server.sessions.set(acp.sessionId, { ...stored, agent: "plan" })
+    const resume = () => acp.request("session/resume", { sessionId: acp.sessionId, cwd: "/workspace" })
+    const resumed = await resume()
+    await resume()
+    acp.server.send(
+      durableEvent("session.model.selected", {
+        sessionID: acp.sessionId,
+        model: { providerID: "test", id: secondModel.id },
+      }),
+    )
+    await optionUpdates(acp, acp.sessionId, 1)
+    acp.server.send(durableEvent("session.agent.selected", { sessionID: acp.sessionId, agent: "build" }))
+    await optionUpdates(acp, acp.sessionId, 2)
+
+    expect(currentValue(resumed, "mode")).toBe("plan")
+    expect(configUpdates(acp, acp.sessionId).map(values)).toEqual([
+      { model: "test/second-model", effort: "default", mode: "plan" },
+      { model: "test/second-model", effort: "default", mode: "build" },
+    ])
+  })
+
+  test("ends on the latest catalog and selection when both change together", async () => {
+    await using acp = await startWire()
+    acp.server.catalog.models = [testModel]
+    await acp.initialize()
+    const { sessionId } = await acp.newSession()
+
+    acp.server.catalog.models = [testModel, secondModel]
+    acp.server.send(
+      ephemeralEvent("model.updated", {}),
+      durableEvent("session.model.selected", {
+        sessionID: sessionId,
+        model: { providerID: "test", id: secondModel.id },
+      }),
+    )
+    await acp.until(
+      () => configUpdates(acp, sessionId).some((options) => selectValues(options, "model").length === 2),
+      "the reloaded catalog",
+    )
+    acp.server.send(durableEvent("session.agent.selected", { sessionID: sessionId, agent: "plan" }))
+    const updates = await acp.until(() => {
+      const updates = configUpdates(acp, sessionId)
+      return values(updates.at(-1)).mode === "plan" && updates
+    }, "the sentinel update")
+
+    expect(
+      updates.slice(-2).map((options) => ({ ...values(options), models: selectValues(options, "model") })),
+    ).toEqual(
+      ["build", "plan"].map((mode) => ({
+        model: "test/second-model",
+        effort: "default",
+        mode,
+        models: ["test/second-model", "test/test-model"],
+      })),
+    )
+  })
+})
+
+function configUpdates(acp: Wire, sessionId: string) {
+  return acp.updates.flatMap((item) =>
+    item.sessionId === sessionId && item.update.sessionUpdate === "config_option_update"
+      ? [item.update.configOptions]
+      : [],
+  )
+}
+
+function optionUpdates(acp: Wire, sessionId: string, count: number) {
+  return acp.until(() => {
+    const updates = configUpdates(acp, sessionId)
+    return updates.length >= count && updates
+  }, `${count} config_option_update for ${sessionId}`)
+}
+
+function values(options: SessionConfigOption[] | undefined) {
+  return Object.fromEntries((options ?? []).map((option) => [option.id, option.currentValue]))
+}