From 7e42a897bc6dbbba580392d779e47137038bcafb Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Thu, 1 Oct 2026 18:45:52 +0530 Subject: [PATCH] feat(acp): follow model and agent changes from other clients (#52499) --- packages/cli/src/acp/service.ts | 1 + packages/cli/src/acp/sessions.ts | 58 +++++-- packages/cli/test/acp/selection.test.ts | 202 ++++++++++++++++++++++++ 3 files changed, 249 insertions(+), 12 deletions(-) create mode 100644 packages/cli/test/acp/selection.test.ts diff --git a/packages/cli/src/acp/service.ts b/packages/cli/src/acp/service.ts index 411bcd61244..0fe9872c33d 100644 --- a/packages/cli/src/acp/service.ts +++ b/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 })) diff --git a/packages/cli/src/acp/sessions.ts b/packages/cli/src/acp/sessions.ts index 4ecb4e0321b..9ae6e9f82aa 100644 --- a/packages/cli/src/acp/sessions.ts +++ b/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()("@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 +} + +type SelectedEvent = Extract 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() // Kept across re-attachment so resuming with the same servers does not add them again. const registeredMcp = new Map>() + const connected = yield* Deferred.make() + + // 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(), } // 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), diff --git a/packages/cli/test/acp/selection.test.ts b/packages/cli/test/acp/selection.test.ts new file mode 100644 index 00000000000..98129b0bf82 --- /dev/null +++ b/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])) +}