From 3400375fb3964dbfdb825400a5d4ebe777efe5fa Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Thu, 1 Oct 2026 18:06:09 +0530 Subject: [PATCH] refactor(acp): split the promise turn from the session lifecycle (#52493) --- packages/cli/src/acp/agent.ts | 27 ++- packages/cli/src/acp/catalog.ts | 8 +- packages/cli/src/acp/config-option.ts | 30 +++ packages/cli/src/acp/connection.ts | 24 +- packages/cli/src/acp/error.ts | 16 +- packages/cli/src/acp/promise.ts | 17 ++ packages/cli/src/acp/service.ts | 310 ++++++-------------------- packages/cli/src/acp/sessions.ts | 102 ++++----- packages/cli/src/acp/turn.ts | 203 +++++++++++++++++ packages/cli/test/acp/catalog.test.ts | 133 +++++------ packages/cli/test/acp/session.test.ts | 31 ++- 11 files changed, 487 insertions(+), 414 deletions(-) create mode 100644 packages/cli/src/acp/promise.ts create mode 100644 packages/cli/src/acp/turn.ts diff --git a/packages/cli/src/acp/agent.ts b/packages/cli/src/acp/agent.ts index 31a8cc9fcd..f906585caf 100644 --- a/packages/cli/src/acp/agent.ts +++ b/packages/cli/src/acp/agent.ts @@ -9,10 +9,14 @@ import { type Stream, } from "@agentclientprotocol/sdk" import type { OpenCodeClient } from "@opencode/client/promise" -import { Cause, Deferred, Effect, type Scope } from "effect" +import { Cause, Deferred, Effect, Ref, type Scope } from "effect" import { ACPCatalog } from "./catalog" +import { ACPConnection } from "./connection" import { ACPError } from "./error" +import { ACPPromise } from "./promise" import { ACPService } from "./service" +import { ACPSessions } from "./sessions" +import { ACPTurn } from "./turn" // Untraced so request spans parent to the caller's span instead of a setup span that has already ended. export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stream: Stream) { @@ -28,8 +32,10 @@ export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stre const handler = Effect.fn(name)( (ctx: AgentHandlerContext) => Deferred.await(ready).pipe(Effect.flatMap((service) => call(service, ctx))), - // Catalog failures arrive typed and are classified like promise rejections. - Effect.catch(ACPError.classify), + Effect.catchTags({ + ACPCatalogLoadError: (error) => ACPPromise.classify(error.cause), + ACPCatalogNotReadyError: (error) => Effect.die(error), + }), Effect.mapError((error) => (error instanceof RequestError ? error : ACPError.toRequestError(error))), Effect.tapCauseIf(Cause.hasDies, (cause) => Effect.logError("ACP request failed", cause)), Effect.catchDefect((defect) => Effect.fail(ACPError.toRequestError(ACPError.fromUnknown(defect)))), @@ -76,7 +82,7 @@ export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stre ) request( "session/close", - handle((service, ctx) => ACPError.promise(() => service.closeSession(ctx.params))), + handle((service, ctx) => service.closeSession(ctx.params)), ) request( "session/fork", @@ -94,14 +100,21 @@ export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stre // `stopReason: "cancelled"`. request( "session/prompt", - handle((service, ctx) => ACPError.promise(() => service.prompt(ctx.params, ctx.signal))), + handle((service, ctx) => ACPPromise.promise(() => service.prompt(ctx.params, ctx.signal))), ) notification( "session/cancel", - handle((service, ctx) => ACPError.promise(() => service.cancel(ctx.params))), + handle((service, ctx) => ACPPromise.promise(() => service.cancel(ctx.params))), ) const connection = app.connect(stream) - yield* Deferred.succeed(ready, yield* ACPService.make({ client, connection, catalog, run })) + const promiseConnection = ACPConnection.promise(connection) + const sessions = yield* ACPSessions.make({ client, connection: ACPConnection.make(connection), catalog }) + const capabilities = yield* Ref.make({ childSessionUpdates: false }) + const turn = ACPTurn.make({ client, connection: promiseConnection, sessions, catalog, capabilities, run }) + yield* Deferred.succeed( + ready, + ACPService.make({ client, connection: promiseConnection, catalog, sessions, capabilities, turn }), + ) return connection }) diff --git a/packages/cli/src/acp/catalog.ts b/packages/cli/src/acp/catalog.ts index 1f550e11b9..0182e577bc 100644 --- a/packages/cli/src/acp/catalog.ts +++ b/packages/cli/src/acp/catalog.ts @@ -3,12 +3,18 @@ import { FSUtil } from "@opencode/util/fs-util" import { Context, Deferred, Effect, Exit, Schedule, Schema, Semaphore, Stream, SubscriptionRef } from "effect" import type { ConfigOptionProvider } from "./config-option" +// ACP runs these itself; they take precedence over server commands with the same name. +export const builtinCommands = new Map([ + ["compact", { description: "Compact the session", start: "compaction" as const }], +]) + export type Catalog = { readonly providers: ConfigOptionProvider[] readonly models: ModelInfo[] readonly defaultModel: ModelRef readonly modes: Array<{ id: string; name: string; description?: string }> readonly defaultModeID: string + /** Server commands, without those shadowed by a built-in. */ readonly commands: CommandInfo[] } @@ -176,7 +182,7 @@ const read = Effect.fnUntraced(function* (client: OpenCodeClient, cwd: string) { }, modes: agents.map((agent) => ({ id: agent.id, name: agent.name, description: agent.description })), defaultModeID: defaultAgent.id, - commands: commandResult.data, + commands: commandResult.data.filter((command) => !builtinCommands.has(command.name)), } satisfies Catalog }) diff --git a/packages/cli/src/acp/config-option.ts b/packages/cli/src/acp/config-option.ts index fe3642613a..80355d0dad 100644 --- a/packages/cli/src/acp/config-option.ts +++ b/packages/cli/src/acp/config-option.ts @@ -1,4 +1,6 @@ import type { SessionConfigOption } from "@agentclientprotocol/sdk" +import type { ModelRef } from "@opencode/client/promise" +import { builtinCommands, type Catalog } from "./catalog" export const DEFAULT_VARIANT_VALUE = "default" @@ -25,6 +27,34 @@ export type ModelSelection = { variant?: string } +/** A session's model and mode. Unset fields follow the catalog defaults. */ +export type Selection = { + readonly model?: ModelRef + readonly modeID?: string +} + +export function currentModel(catalog: Catalog, selection: Selection) { + return selection.model ?? catalog.defaultModel +} + +export function configOptions(catalog: Catalog, selection: Selection) { + const model = currentModel(catalog, selection) + return buildConfigOptions({ + providers: catalog.providers, + currentModel: { providerID: model.providerID, modelID: model.id }, + currentVariant: model.variant, + modes: catalog.modes, + currentModeId: selection.modeID ?? catalog.defaultModeID, + }) +} + +export function availableCommands(catalog: Catalog) { + return [ + ...catalog.commands.map((command) => ({ name: command.name, description: command.description ?? "" })), + ...Array.from(builtinCommands, ([name, command]) => ({ name, description: command.description })), + ] +} + export function buildConfigOptions(input: { providers: readonly ConfigOptionProvider[] currentModel: ModelSelection["model"] diff --git a/packages/cli/src/acp/connection.ts b/packages/cli/src/acp/connection.ts index 876d7dfa89..87965ef58b 100644 --- a/packages/cli/src/acp/connection.ts +++ b/packages/cli/src/acp/connection.ts @@ -8,27 +8,19 @@ import { type SessionNotification, } from "@agentclientprotocol/sdk" import { Context, type Effect } from "effect" -import { ACPError } from "./error" - -type Failure = ACPError.Error | RequestError +import type { ACPError } from "./error" +import { ACPPromise } from "./promise" export interface Interface { - readonly sessionUpdate: (params: SessionNotification) => Effect.Effect - /** Interrupting the request cancels it on the client. */ - readonly requestPermission: (params: RequestPermissionRequest) => Effect.Effect - readonly extNotification: (method: string, params: Record) => Effect.Effect + readonly sessionUpdate: (params: SessionNotification) => Effect.Effect } export class Service extends Context.Service()("@opencode/cli/acp/Connection") {} -export function service(connection: AgentConnection) { +export function make(connection: AgentConnection) { return Service.of({ - sessionUpdate: (params) => ACPError.promise(() => connection.client.notify(methods.client.session.update, params)), - requestPermission: (params) => - ACPError.promise((signal) => - connection.client.request(methods.client.session.requestPermission, params, { cancellationSignal: signal }), - ), - extNotification: (method, params) => ACPError.promise(() => connection.client.notify(method, params)), + sessionUpdate: (params) => + ACPPromise.promise(() => connection.client.notify(methods.client.session.update, params)), }) } @@ -39,8 +31,8 @@ export type Connection = { extNotification?(method: string, params: Record): Promise } -/** Promise view for the turn and permission code until they run as effects. */ -export function make(connection: AgentConnection): Connection { +/** Promise view for the turn, permission, and replay code until they run as effects. */ +export function promise(connection: AgentConnection): Connection { return { signal: connection.signal, sessionUpdate: (params) => connection.client.notify(methods.client.session.update, params), diff --git a/packages/cli/src/acp/error.ts b/packages/cli/src/acp/error.ts index 7362b322dc..6ffc0c8874 100644 --- a/packages/cli/src/acp/error.ts +++ b/packages/cli/src/acp/error.ts @@ -1,7 +1,5 @@ import { RequestError } from "@agentclientprotocol/sdk" -import { ClientError } from "@opencode/client/promise" -import { Effect, Schema } from "effect" -import { ACPCatalog } from "./catalog" +import { Schema } from "effect" export class SessionNotFoundError extends Schema.TaggedError()("ACPSessionNotFoundError", { sessionId: Schema.String, @@ -103,18 +101,6 @@ export function toRequestError(error: Error): RequestError { return exhaustive } -/** Runs a promise, keeping ACP failures typed. Any other rejection is a defect. */ -export const promise = (evaluate: (signal: AbortSignal) => Promise) => - Effect.tryPromise({ try: evaluate, catch: (cause) => cause }).pipe(Effect.catch(classify)) - -export function classify(cause: unknown): Effect.Effect { - // A catalog load failure is classified by the client error that caused it. - if (cause instanceof ACPCatalog.LoadError) return classify(cause.cause) - if (cause instanceof RequestError || is(cause)) return Effect.fail(cause) - if (cause instanceof ClientError && cause.reason === "Transport") return Effect.fail(new ServerUnavailableError()) - return Effect.die(cause) -} - export function fromUnknown(error: unknown, service?: string) { const errorName = error instanceof Error ? error.name : undefined return new ServiceFailureError({ safeMessage: "Internal service failure", service, errorName }) diff --git a/packages/cli/src/acp/promise.ts b/packages/cli/src/acp/promise.ts new file mode 100644 index 0000000000..56f7a33b76 --- /dev/null +++ b/packages/cli/src/acp/promise.ts @@ -0,0 +1,17 @@ +import { RequestError } from "@agentclientprotocol/sdk" +import { ClientError } from "@opencode/client/promise" +import { Effect } from "effect" +import { ACPError } from "./error" + +/** Runs a promise, keeping ACP failures typed. Any other rejection is a defect. */ +export const promise = (evaluate: (signal: AbortSignal) => Promise) => + Effect.tryPromise({ try: evaluate, catch: (cause) => cause }).pipe(Effect.catch(classify)) + +export function classify(cause: unknown): Effect.Effect { + if (cause instanceof RequestError || ACPError.is(cause)) return Effect.fail(cause) + if (cause instanceof ClientError && cause.reason === "Transport") + return Effect.fail(new ACPError.ServerUnavailableError()) + return Effect.die(cause) +} + +export * as ACPPromise from "./promise" diff --git a/packages/cli/src/acp/service.ts b/packages/cli/src/acp/service.ts index 45273e634b..411bcd6124 100644 --- a/packages/cli/src/acp/service.ts +++ b/packages/cli/src/acp/service.ts @@ -1,9 +1,8 @@ -import { isSessionNotFoundError, type CommandInfo, type ModelRef, type OpenCodeClient } from "@opencode/client/promise" +import { isSessionNotFoundError, type ModelRef, type OpenCodeClient } from "@opencode/client/promise" import { FSUtil } from "@opencode/util/fs-util" -import { Effect, Option, Ref, Stream, type Scope } from "effect" +import { Effect, Option, Ref, Stream } from "effect" import { withTimestampedFallback } from "@opencode/util/session-title-fallback" import type { - AgentConnection, AuthenticateRequest, AuthenticateResponse, AuthMethod, @@ -33,34 +32,17 @@ import type { SetSessionModeResponse, } from "@agentclientprotocol/sdk" import { OPENCODE_VERSION } from "../version" -import { SessionMessage } from "@opencode/schema/session-message" import type { ACPCatalog, Catalog } from "./catalog" -import { DEFAULT_VARIANT_VALUE, parseModelSelection } from "./config-option" -import { ACPConnection } from "./connection" -import { promptContentToParts } from "./content" -import { - ChildSessionUpdateMethod, - ChildSessionUpdatesCapability, - replayMessages, - streamTurn, - type ChildSessionUpdate, - type TurnControl, - type TurnStart, -} from "./event" +import { configOptions, currentModel, DEFAULT_VARIANT_VALUE, parseModelSelection } from "./config-option" +import type { ACPConnection } from "./connection" +import { ChildSessionUpdatesCapability, replayMessages } from "./event" import { ACPError } from "./error" -import { ACPSessions, builtinCommands, type Attached } from "./sessions" +import { ACPPromise } from "./promise" +import type { ACPSessions, Attached } from "./sessions" +import type { ACPTurn } from "./turn" export const AuthMethodID = "opencode-login" -type PreparedPrompt = { - readonly start: TurnStart - readonly text: string - readonly files: Array<{ readonly uri: string; readonly name?: string }> - readonly synthetic: ReadonlyArray - readonly slash?: { readonly name: string; readonly args: string } - readonly command?: CommandInfo -} - export type Failure = ACPError.Error | RequestError | ACPCatalog.Error export interface Interface { @@ -71,32 +53,27 @@ export interface Interface { readonly listSessions: (input: ListSessionsRequest) => Effect.Effect readonly deleteSession: (input: DeleteSessionRequest) => Effect.Effect readonly resumeSession: (input: ResumeSessionRequest) => Effect.Effect + readonly closeSession: (input: CloseSessionRequest) => Effect.Effect readonly forkSession: (input: ForkSessionRequest) => Effect.Effect readonly setSessionConfigOption: ( input: SetSessionConfigOptionRequest, ) => Effect.Effect readonly setSessionMode: (input: SetSessionModeRequest) => Effect.Effect - closeSession(input: CloseSessionRequest): Promise prompt(input: PromptRequest, signal?: AbortSignal): Promise cancel(input: CancelNotification): Promise } -export const make = Effect.fnUntraced(function* (input: { +export function make(input: { readonly client: OpenCodeClient - readonly connection: AgentConnection + /** Replay still writes through the promise view. */ + readonly connection: ACPConnection.Connection readonly catalog: ACPCatalog.Interface - readonly run: (effect: Effect.Effect) => Promise -}) { - const connection = ACPConnection.service(input.connection) - // The turn still runs on promises. - const turnConnection = ACPConnection.make(input.connection) - const sessions = yield* ACPSessions.make({ client: input.client, connection, catalog: input.catalog }) - const capabilities = yield* Ref.make({ childSessionUpdates: false }) - const active = new Map }>() - - const configOptions = Effect.fnUntraced(function* (attached: Attached) { - const catalog = yield* input.catalog.get(attached.cwd) - return ACPSessions.configOptions(catalog, yield* Ref.get(attached.selection)) + readonly sessions: ACPSessions.Interface + readonly capabilities: Ref.Ref<{ readonly childSessionUpdates: boolean }> + readonly turn: ACPTurn.Interface +}): Interface { + const currentOptions = Effect.fnUntraced(function* (attached: Attached) { + return configOptions(yield* input.catalog.get(attached.cwd), yield* Ref.get(attached.selection)) }) // A selection the catalog has not seen may be new on the server, so reload once before rejecting it. @@ -108,80 +85,61 @@ export const make = Effect.fnUntraced(function* (input: { } const selectOption = Effect.fnUntraced(function* (attached: Attached, configId: string, value: string) { - if (configId === "mode") return yield* selectMode(attached, value) - if (configId !== "model" && configId !== "effort") return yield* new ACPError.InvalidConfigOptionError({ configId }) const catalog = yield* input.catalog.get(attached.cwd) - const current = ACPSessions.currentModel(catalog, yield* Ref.get(attached.selection)) - const model = - configId === "model" - ? yield* requireModel(catalog, value, current) - : yield* requireEffort(catalog, value, current) + const current = currentModel(catalog, yield* Ref.get(attached.selection)) + switch (configId) { + case "model": + return yield* selectModel(attached, yield* requireModel(catalog, value, current)) + case "effort": + return yield* selectModel(attached, yield* requireEffort(catalog, value, current)) + case "mode": + return yield* selectMode(attached, value) + default: + return yield* new ACPError.InvalidConfigOptionError({ configId }) + } + }) + + const selectModel = Effect.fnUntraced(function* (attached: Attached, model: ModelRef) { yield* Ref.update(attached.selection, (selection) => ({ ...selection, model })) - yield* ACPError.promise(() => input.client.session.switchModel({ sessionID: attached.id, model })) + yield* ACPPromise.promise(() => input.client.session.switchModel({ sessionID: attached.id, model })) }) const selectMode = Effect.fnUntraced(function* (attached: Attached, modeID: string) { const catalog = yield* input.catalog.get(attached.cwd) if (!catalog.modes.some((mode) => mode.id === modeID)) return yield* new ACPError.InvalidModeError({ mode: modeID }) yield* Ref.update(attached.selection, (selection) => ({ ...selection, modeID })) - yield* ACPError.promise(() => input.client.session.switchAgent({ sessionID: attached.id, agent: modeID })) + yield* ACPPromise.promise(() => input.client.session.switchAgent({ sessionID: attached.id, agent: modeID })) }) const getSession = Effect.fnUntraced(function* (sessionID: string, cwd: string) { - const session = yield* ACPError.promise(() => input.client.session.get({ sessionID }).catch(notFound(sessionID))) + const session = yield* ACPPromise.promise(() => + input.client.session.get({ sessionID }).catch((error) => { + if (isSessionNotFoundError(error)) throw new ACPError.SessionNotFoundError({ sessionId: sessionID }) + throw error + }), + ) if (FSUtil.resolve(cwd) !== FSUtil.resolve(session.location.directory)) return yield* new ACPError.SessionDirectoryMismatchError({ sessionId: sessionID, cwd }) return session }) const replay = (attached: Attached) => - Stream.paginate(Option.none(), (cursor) => - ACPError.promise(() => - Option.isSome(cursor) - ? input.client.message.list({ sessionID: attached.id, limit: 200, cursor: cursor.value }) + Stream.paginate(undefined, (cursor: string | undefined) => + ACPPromise.promise(() => + cursor + ? input.client.message.list({ sessionID: attached.id, limit: 200, cursor }) : input.client.message.list({ sessionID: attached.id, limit: 200, order: "asc" }), - ).pipe( - Effect.map( - (page) => [page.data, Option.fromNullishOr(page.cursor.next).pipe(Option.map(Option.some))] as const, - ), - ), + ).pipe(Effect.map((page) => [page.data, Option.fromNullishOr(page.cursor.next)] as const)), ).pipe( Stream.runCollect, Effect.flatMap((messages) => - ACPError.promise(() => replayMessages(turnConnection, attached.id, attached.cwd, messages)), + ACPPromise.promise(() => replayMessages(input.connection, attached.id, attached.cwd, messages)), ), ) - const sendUsageUpdate = Effect.fnUntraced(function* (attached: Attached, used: number | undefined) { - if (!used) return - const catalog = yield* input.catalog.get(attached.cwd) - const current = ACPSessions.currentModel(catalog, yield* Ref.get(attached.selection)) - const model = catalog.models.find((item) => item.providerID === current.providerID && item.id === current.id) - if (!model?.limit.context) return - const info = yield* ACPError.promise(() => input.client.session.get({ sessionID: attached.id })) - yield* connection.sessionUpdate({ - sessionId: attached.id, - update: { - sessionUpdate: "usage_update", - used, - size: model.limit.context, - cost: { amount: info.cost, currency: "USD" }, - }, - }) - }) - - const cancelTurn = (sessionID: string) => { - const turn = active.get(sessionID) - if (turn) { - turn.control.cancelled = true - turn.control.admission.abort() - } - return input.client.session.interrupt({ sessionID }) - } - return { initialize: Effect.fnUntraced(function* (params) { - yield* Ref.set(capabilities, { + yield* Ref.set(input.capabilities, { childSessionUpdates: params.clientCapabilities?._meta?.[ChildSessionUpdatesCapability] === true, }) const authMethod: AuthMethod = { @@ -216,20 +174,20 @@ export const make = Effect.fnUntraced(function* (input: { // Load before creating so a catalog failure leaves no session behind. Agent and model stay unset // so the server resolves its defaults after plugins activate. yield* input.catalog.get(params.cwd) - const created = yield* ACPError.promise(() => + const created = yield* ACPPromise.promise(() => input.client.session.create({ location: { directory: params.cwd } }), ) - const attached = yield* sessions.attach(created, params.cwd, params.mcpServers) - return { sessionId: attached.id, configOptions: yield* configOptions(attached) } + const attached = yield* input.sessions.attach(created, params.cwd, params.mcpServers) + return { sessionId: attached.id, configOptions: yield* currentOptions(attached) } }), loadSession: Effect.fnUntraced(function* (params) { const session = yield* getSession(params.sessionId, params.cwd) - const attached = yield* sessions.attach(session, session.location.directory, params.mcpServers) + const attached = yield* input.sessions.attach(session, session.location.directory, params.mcpServers) yield* replay(attached) - return { configOptions: yield* configOptions(attached) } + return { configOptions: yield* currentOptions(attached) } }), listSessions: Effect.fnUntraced(function* (params) { - const page = yield* ACPError.promise(() => + const page = yield* ACPPromise.promise(() => input.client.session.list({ ...(params.cwd ? { directory: params.cwd } : {}), order: "desc", @@ -248,159 +206,47 @@ export const make = Effect.fnUntraced(function* (input: { } }), deleteSession: Effect.fnUntraced(function* (params) { - yield* ACPError.promise(() => - input.client.session.remove({ sessionID: params.sessionId }).catch(notFound(params.sessionId)), - ).pipe(Effect.catchTag("ACPSessionNotFoundError", () => Effect.void)) - yield* sessions.detach(params.sessionId) + yield* ACPPromise.promise(() => + input.client.session.remove({ sessionID: params.sessionId }).catch((error) => { + if (!isSessionNotFoundError(error)) throw error + }), + ) + yield* input.sessions.detach(params.sessionId) return {} }), resumeSession: Effect.fnUntraced(function* (params) { const session = yield* getSession(params.sessionId, params.cwd) - const attached = yield* sessions.attach(session, session.location.directory, params.mcpServers ?? []) - return { configOptions: yield* configOptions(attached) } + const attached = yield* input.sessions.attach(session, session.location.directory, params.mcpServers ?? []) + return { configOptions: yield* currentOptions(attached) } + }), + closeSession: Effect.fnUntraced(function* (params) { + yield* input.turn.close(params.sessionId) + yield* input.sessions.detach(params.sessionId) + return {} }), forkSession: Effect.fnUntraced(function* (params) { - const forked = yield* ACPError.promise(() => input.client.session.fork({ sessionID: params.sessionId })) - const attached = yield* sessions.attach(forked, forked.location.directory, params.mcpServers ?? []) + const forked = yield* ACPPromise.promise(() => input.client.session.fork({ sessionID: params.sessionId })) + const attached = yield* input.sessions.attach(forked, forked.location.directory, params.mcpServers ?? []) yield* replay(attached) - return { sessionId: attached.id, configOptions: yield* configOptions(attached) } + return { sessionId: attached.id, configOptions: yield* currentOptions(attached) } }), setSessionConfigOption: Effect.fnUntraced(function* (params) { - const attached = yield* sessions.require(params.sessionId) + const attached = yield* input.sessions.require(params.sessionId) const value = params.value if (typeof value !== "string") return yield* new ACPError.InvalidConfigOptionError({ configId: params.configId }) yield* withReload(attached, selectOption(attached, params.configId, value)) - return { configOptions: yield* configOptions(attached) } + return { configOptions: yield* currentOptions(attached) } }), setSessionMode: Effect.fnUntraced(function* (params) { - const attached = yield* sessions.require(params.sessionId) + const attached = yield* input.sessions.require(params.sessionId) yield* withReload(attached, selectMode(attached, params.modeId)) return {} }), - closeSession: async (params) => { - const turn = active.get(params.sessionId) - await cancelTurn(params.sessionId).catch((error) => { - if (!isSessionNotFoundError(error)) throw error - }) - await turn?.turn.catch(() => {}) - await input.run(sessions.detach(params.sessionId)) - return {} - }, - prompt: async (params, signal) => { - // Read everything first so the active check and registration below stay synchronous. - const resolved = await input.run( - Effect.gen(function* () { - const attached = yield* sessions.require(params.sessionId) - return { - attached, - catalog: yield* input.catalog.get(attached.cwd), - childSessionUpdates: (yield* Ref.get(capabilities)).childSessionUpdates, - } - }), - ) - const state = resolved.attached - 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(resolved.catalog, params.prompt, messageID) - const control: TurnControl = { cancelled: false, admission: new AbortController() } - const extNotification = turnConnection.extNotification - const childSessionUpdate = - resolved.childSessionUpdates && extNotification - ? (update: ChildSessionUpdate) => extNotification(ChildSessionUpdateMethod, update).then(() => {}) - : undefined - // A `$/cancel_request` for this prompt behaves like `session/cancel` for its turn. - const cancel = () => void cancelTurn(state.id).catch(() => {}) - const turn = streamTurn({ - client: input.client, - connection: turnConnection, - sessionID: state.id, - cwd: state.cwd, - start: prepared.start, - action: prepared.command !== undefined, - control, - connectionSignal: turnConnection.signal, - sessionSignal: state.signal, - submit: (signal) => submitPrompt(input.client, state, prepared, signal), - ...(childSessionUpdate ? { childSessionUpdate } : {}), - }) - .then(async (result) => { - await input.run(sendUsageUpdate(state, result.contextTokens)).catch(() => {}) - return result.response - }) - .finally(() => { - signal?.removeEventListener("abort", cancel) - if (active.get(state.id)?.control === control) active.delete(state.id) - }) - active.set(state.id, { control, turn }) - signal?.addEventListener("abort", cancel, { once: true }) - // The cancel may already be buffered behind the awaits above. - if (signal?.aborted) cancel() - return turn - }, - cancel: async (params) => { - await cancelTurn(params.sessionId).catch(() => {}) - }, - } satisfies Interface -}) - -function notFound(sessionID: string) { - return (error: unknown): never => { - throw isSessionNotFoundError(error) ? new ACPError.SessionNotFoundError({ sessionId: sessionID }) : error + prompt: input.turn.prompt, + cancel: input.turn.cancel, } } -function preparePrompt(catalog: Catalog, prompt: PromptRequest["prompt"], messageID: string): PreparedPrompt { - const parts = promptContentToParts(prompt) - const visible = parts.filter((part) => part.type !== "text" || (!part.synthetic && !part.ignored)) - const synthetic = parts.flatMap((part) => (part.type === "text" && part.synthetic ? [part.text] : [])) - const text = visible.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("\n") - const files = visible.flatMap((part) => (part.type === "file" ? [{ uri: part.url, name: part.filename }] : [])) - const slash = detectSlashCommand(text) - const command = - slash && !builtinCommands.has(slash.name) ? catalog.commands.find((item) => item.name === slash.name) : undefined - const start = turnStart(messageID, slash) - return { start, text, files, synthetic, slash, command } -} - -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, - }) - } - if (prompt.start.type === "compaction") return client.session.compact({ sessionID: session.id, id: prompt.start.id }) - if (prompt.command) { - return client.session.command( - { - sessionID: session.id, - name: prompt.command.name, - text: prompt.slash?.args ?? "", - files: prompt.files, - delivery: "steer", - }, - { signal }, - ) - } - return client.session.prompt( - { sessionID: session.id, id: prompt.start.id, text: prompt.text, files: prompt.files, delivery: "steer" }, - { signal }, - ) -} - -function turnStart(messageID: string, slash: PreparedPrompt["slash"]): TurnStart { - if (slash && builtinCommands.get(slash.name)?.start === "compaction") return { type: "compaction", id: messageID } - return { type: "input", id: messageID } -} - const requireModel = Effect.fnUntraced(function* (catalog: Catalog, modelID: string, current: ModelRef) { const selected = parseModelSelection(modelID, catalog.providers) const model = catalog.models.find( @@ -426,12 +272,4 @@ const requireEffort = Effect.fnUntraced(function* (catalog: Catalog, effort: str return { ...current, variant: effort } satisfies ModelRef }) -function detectSlashCommand(text: string): { readonly name: string; readonly args: string } | undefined { - const value = text.trim() - if (!value.startsWith("/")) return undefined - const [name, ...rest] = value.slice(1).split(/\s+/) - if (!name) return undefined - return { name, args: rest.join(" ").trim() } -} - export * as ACPService from "./service" diff --git a/packages/cli/src/acp/sessions.ts b/packages/cli/src/acp/sessions.ts index 5f50aadca2..4ecb4e0321 100644 --- a/packages/cli/src/acp/sessions.ts +++ b/packages/cli/src/acp/sessions.ts @@ -1,22 +1,12 @@ import { isDeepStrictEqual } from "node:util" import type { McpServer, RequestError } from "@agentclientprotocol/sdk" -import type { ModelRef, OpenCodeClient, SessionInfo } from "@opencode/client/promise" -import { Cause, Context, Effect, Exit, Ref, Scope, Stream } from "effect" +import type { OpenCodeClient, SessionInfo } from "@opencode/client/promise" +import { Context, Effect, Exit, Ref, Scope, Stream } from "effect" import type { ACPCatalog, Catalog } from "./catalog" -import { buildConfigOptions } from "./config-option" +import { availableCommands, configOptions, type Selection } from "./config-option" import type { ACPConnection } from "./connection" import { ACPError } from "./error" - -// ACP runs these itself; they take precedence over server commands with the same name. -export const builtinCommands = new Map([ - ["compact", { description: "Compact the session", start: "compaction" as const }], -]) - -/** Unset fields follow the server defaults. */ -export type Selection = { - readonly model?: ModelRef - readonly modeID?: string -} +import { ACPPromise } from "./promise" export type Attached = { readonly id: string @@ -29,7 +19,8 @@ export type Attached = { export interface Interface { /** * Attaches a session in its own scope, closing any previous attachment of the same ID. The scope follows the - * cwd's catalog and pushes config option and command updates while it is open. + * cwd's catalog and pushes config option and command updates while it is open. A failed attach leaves the + * session detached. */ readonly attach: ( session: SessionInfo, @@ -43,28 +34,22 @@ export interface Interface { export class Service extends Context.Service()("@opencode/cli/acp/Sessions") {} +type Entry = { readonly attached: Attached; readonly scope: Scope.Closeable } + export const make = Effect.fnUntraced(function* (input: { readonly client: OpenCodeClient readonly connection: ACPConnection.Interface readonly catalog: ACPCatalog.Interface }) { const scope = yield* Effect.scope - const sessions = new Map() + const sessions = new Map() // Kept across re-attachment so resuming with the same servers does not add them again. const registeredMcp = new Map>() const sendCommands = (sessionID: string, catalog: Catalog) => input.connection.sessionUpdate({ sessionId: sessionID, - update: { - sessionUpdate: "available_commands_update", - availableCommands: [ - ...catalog.commands - .filter((command) => !builtinCommands.has(command.name)) - .map((command) => ({ name: command.name, description: command.description ?? "" })), - ...Array.from(builtinCommands, ([name, command]) => ({ name, description: command.description })), - ], - }, + update: { sessionUpdate: "available_commands_update", availableCommands: availableCommands(catalog) }, }) const changed = Effect.fnUntraced(function* (attached: Attached, previous: Catalog, next: Catalog) { @@ -91,58 +76,68 @@ export const make = Effect.fnUntraced(function* (input: { const key = `${server.name}:${stableStringify(config)}` if (registered.has(key)) return Effect.void registered.add(key) - return ACPError.promise(() => + return ACPPromise.promise(() => input.client.mcp.add({ server: server.name, location: { directory: attached.cwd }, config }), ).pipe( - // An interrupted add still completes on the server, so only a failed one is forgotten. - Effect.tapCause((cause) => - Cause.hasInterruptsOnly(cause) ? Effect.void : Effect.sync(() => registered.delete(key)), - ), + Effect.onError(() => Effect.sync(() => registered.delete(key))), + Effect.uninterruptible, ) }), { concurrency: "unbounded", discard: true }, ) }) + const remove = (sessionID: string, entry: Entry) => + Effect.suspend(() => { + if (sessions.get(sessionID) === entry) { + sessions.delete(sessionID) + registeredMcp.delete(sessionID) + } + return Scope.close(entry.scope, Exit.void) + }) + return Service.of({ attach: Effect.fn("cli.acp.sessions.attach")(function* (session, cwd, mcpServers) { const current = yield* input.catalog.get(cwd) const abort = new AbortController() - const attached: Attached = { - id: session.id, - cwd, - selection: yield* Ref.make({ model: session.model, modeID: session.agent }), - signal: abort.signal, + const entry: Entry = { + attached: { + id: session.id, + cwd, + selection: yield* Ref.make({ model: session.model, modeID: session.agent }), + signal: abort.signal, + }, + scope: Scope.forkUnsafe(scope), } // Swap synchronously so concurrent attaches of one ID cannot both keep a scope. - const sessionScope = Scope.forkUnsafe(scope) const replaced = sessions.get(session.id) - sessions.set(session.id, { attached, scope: sessionScope }) + sessions.set(session.id, entry) if (replaced) yield* Scope.close(replaced.scope, Exit.void) yield* Scope.addFinalizer( - sessionScope, + entry.scope, Effect.sync(() => abort.abort()), ) + yield* registerMcp(entry.attached, mcpServers).pipe( + 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( Stream.runFoldEffect( () => current, (previous, next) => next === previous ? Effect.succeed(previous) - : changed(attached, previous, next).pipe(Effect.ignore, Effect.as(next)), + : changed(entry.attached, previous, next).pipe(Effect.ignore, Effect.as(next)), ), Effect.ignore, - Effect.forkIn(sessionScope), + Effect.forkIn(entry.scope), ) - yield* registerMcp(attached, mcpServers) - yield* sendCommands(attached.id, yield* input.catalog.get(cwd)) - return attached + return entry.attached }), detach: Effect.fn("cli.acp.sessions.detach")(function* (sessionID) { const entry = sessions.get(sessionID) - sessions.delete(sessionID) - registeredMcp.delete(sessionID) - if (entry) yield* Scope.close(entry.scope, Exit.void) + if (entry) yield* remove(sessionID, entry) }), require: Effect.fn("cli.acp.sessions.require")(function* (sessionID) { const entry = sessions.get(sessionID) @@ -152,21 +147,6 @@ export const make = Effect.fnUntraced(function* (input: { }) }) -export function currentModel(catalog: Catalog, selection: Selection) { - return selection.model ?? catalog.defaultModel -} - -export function configOptions(catalog: Catalog, selection: Selection) { - const model = currentModel(catalog, selection) - return buildConfigOptions({ - providers: catalog.providers, - currentModel: { providerID: model.providerID, modelID: model.id }, - currentVariant: model.variant, - modes: catalog.modes, - currentModeId: selection.modeID ?? catalog.defaultModeID, - }) -} - function mcpConfig(server: McpServer) { if ("type" in server) { if (server.type === "acp") throw new Error("MCP-over-ACP is not supported") diff --git a/packages/cli/src/acp/turn.ts b/packages/cli/src/acp/turn.ts new file mode 100644 index 0000000000..1b4738fe26 --- /dev/null +++ b/packages/cli/src/acp/turn.ts @@ -0,0 +1,203 @@ +import type { CancelNotification, PromptRequest, PromptResponse, RequestError } from "@agentclientprotocol/sdk" +import { isSessionNotFoundError, type CommandInfo, type OpenCodeClient } from "@opencode/client/promise" +import { SessionMessage } from "@opencode/schema/session-message" +import { Effect, Ref, type Scope } from "effect" +import { builtinCommands, type ACPCatalog, type Catalog } from "./catalog" +import { currentModel } from "./config-option" +import type { ACPConnection } from "./connection" +import { promptContentToParts } from "./content" +import { ACPError } from "./error" +import { + ChildSessionUpdateMethod, + streamTurn, + type ChildSessionUpdate, + type TurnControl, + type TurnStart, +} from "./event" +import { ACPPromise } from "./promise" +import type { ACPSessions, Attached } from "./sessions" + +type PreparedPrompt = { + readonly start: TurnStart + readonly text: string + readonly files: Array<{ readonly uri: string; readonly name?: string }> + readonly synthetic: ReadonlyArray + readonly slash?: { readonly name: string; readonly args: string } + readonly command?: CommandInfo +} + +export interface Interface { + prompt(input: PromptRequest, signal?: AbortSignal): Promise + cancel(input: CancelNotification): Promise + /** Cancels the session's active turn and waits for it to settle. */ + readonly close: (sessionID: string) => Effect.Effect +} + +export function make(input: { + readonly client: OpenCodeClient + readonly connection: ACPConnection.Connection + readonly sessions: ACPSessions.Interface + readonly catalog: ACPCatalog.Interface + readonly capabilities: Ref.Ref<{ readonly childSessionUpdates: boolean }> + readonly run: (effect: Effect.Effect) => Promise +}): Interface { + const active = new Map }>() + + const cancelTurn = (sessionID: string) => { + const turn = active.get(sessionID) + if (turn) { + turn.control.cancelled = true + turn.control.admission.abort() + } + return input.client.session.interrupt({ sessionID }) + } + + const sendUsageUpdate = async (state: Attached, used?: number) => { + if (!used) return + const model = await input.run( + Effect.gen(function* () { + const catalog = yield* input.catalog.get(state.cwd) + const current = currentModel(catalog, yield* Ref.get(state.selection)) + return catalog.models.find((item) => item.providerID === current.providerID && item.id === current.id) + }), + ) + if (!model?.limit.context) return + const info = await input.client.session.get({ sessionID: state.id }) + await input.connection.sessionUpdate({ + sessionId: state.id, + update: { + sessionUpdate: "usage_update", + used, + size: model.limit.context, + cost: { amount: info.cost, currency: "USD" }, + }, + }) + } + + return { + prompt: async (params, signal) => { + // Read everything first so the active check and registration below stay synchronous. + const resolved = await input.run( + Effect.gen(function* () { + const attached = yield* input.sessions.require(params.sessionId) + return { + attached, + catalog: yield* input.catalog.get(attached.cwd), + childSessionUpdates: (yield* Ref.get(input.capabilities)).childSessionUpdates, + } + }), + ) + const state = resolved.attached + 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(resolved.catalog, params.prompt, messageID) + const control: TurnControl = { cancelled: false, admission: new AbortController() } + const extNotification = input.connection.extNotification + const childSessionUpdate = + resolved.childSessionUpdates && extNotification + ? (update: ChildSessionUpdate) => extNotification(ChildSessionUpdateMethod, update).then(() => {}) + : undefined + // A `$/cancel_request` for this prompt behaves like `session/cancel` for its turn. + const cancel = () => void cancelTurn(state.id).catch(() => {}) + const turn = streamTurn({ + client: input.client, + connection: input.connection, + sessionID: state.id, + cwd: state.cwd, + start: prepared.start, + action: prepared.command !== undefined, + control, + connectionSignal: input.connection.signal, + sessionSignal: state.signal, + submit: (signal) => submitPrompt(input.client, state, prepared, signal), + ...(childSessionUpdate ? { childSessionUpdate } : {}), + }) + .then(async (result) => { + await sendUsageUpdate(state, result.contextTokens).catch(() => {}) + return result.response + }) + .finally(() => { + signal?.removeEventListener("abort", cancel) + if (active.get(state.id)?.control === control) active.delete(state.id) + }) + active.set(state.id, { control, turn }) + signal?.addEventListener("abort", cancel, { once: true }) + // The cancel may already be buffered behind the awaits above. + if (signal?.aborted) cancel() + return turn + }, + cancel: async (params) => { + await cancelTurn(params.sessionId).catch(() => {}) + }, + close: Effect.fnUntraced(function* (sessionID) { + const turn = active.get(sessionID) + yield* ACPPromise.promise(() => + cancelTurn(sessionID).catch((error) => { + if (!isSessionNotFoundError(error)) throw error + }), + ) + if (turn) yield* Effect.promise(() => turn.turn.catch(() => {})) + }), + } +} + +function preparePrompt(catalog: Catalog, prompt: PromptRequest["prompt"], messageID: string): PreparedPrompt { + const parts = promptContentToParts(prompt) + const visible = parts.filter((part) => part.type !== "text" || (!part.synthetic && !part.ignored)) + const synthetic = parts.flatMap((part) => (part.type === "text" && part.synthetic ? [part.text] : [])) + const text = visible.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("\n") + const files = visible.flatMap((part) => (part.type === "file" ? [{ uri: part.url, name: part.filename }] : [])) + const slash = detectSlashCommand(text) + const command = slash ? catalog.commands.find((item) => item.name === slash.name) : undefined + const start = turnStart(messageID, slash) + return { start, text, files, synthetic, slash, command } +} + +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, + }) + } + if (prompt.start.type === "compaction") return client.session.compact({ sessionID: session.id, id: prompt.start.id }) + if (prompt.command) { + return client.session.command( + { + sessionID: session.id, + name: prompt.command.name, + text: prompt.slash?.args ?? "", + files: prompt.files, + delivery: "steer", + }, + { signal }, + ) + } + return client.session.prompt( + { sessionID: session.id, id: prompt.start.id, text: prompt.text, files: prompt.files, delivery: "steer" }, + { signal }, + ) +} + +function turnStart(messageID: string, slash: PreparedPrompt["slash"]): TurnStart { + if (slash && builtinCommands.get(slash.name)?.start === "compaction") return { type: "compaction", id: messageID } + return { type: "input", id: messageID } +} + +function detectSlashCommand(text: string): { readonly name: string; readonly args: string } | undefined { + const value = text.trim() + if (!value.startsWith("/")) return undefined + const [name, ...rest] = value.slice(1).split(/\s+/) + if (!name) return undefined + return { name, args: rest.join(" ").trim() } +} + +export * as ACPTurn from "./turn" diff --git a/packages/cli/test/acp/catalog.test.ts b/packages/cli/test/acp/catalog.test.ts index 822c85a067..8d2530bd8d 100644 --- a/packages/cli/test/acp/catalog.test.ts +++ b/packages/cli/test/acp/catalog.test.ts @@ -118,104 +118,71 @@ describe("acp catalog and config options over the wire", () => { const session = await acp.newSession() acp.server.catalog.models = [testModel, secondModel] acp.server.catalog.agents = [buildAgent, planAgent, configured] + const set = (configId: string, value: string) => + acp.request("session/set_config_option", { sessionId: session.sessionId, configId, value }) + const initialModelReads = modelReads(acp) - const model = await acp.request("session/set_config_option", { - sessionId: session.sessionId, - configId: "model", - value: "test/second-model", - }) + const model = await set("model", "test/second-model") + const reloadedModelReads = modelReads(acp) + const missingModel = await rpcError(set("model", "test/missing-model")) + const missingModelReads = modelReads(acp) await acp.request("session/set_mode", { sessionId: session.sessionId, modeId: "copilot-build" }) const reads = agentReads(acp) - const missing = await rpcError( - acp.request("session/set_config_option", { sessionId: session.sessionId, configId: "mode", value: "missing" }), - ) + const missing = await rpcError(set("mode", "missing")) expect(currentValue(model, "model")).toBe("test/second-model") + expect([reloadedModelReads, missingModelReads]).toEqual([initialModelReads + 1, initialModelReads + 2]) + expect(missingModel).toMatchObject({ code: -32602, data: { modelId: "test/missing-model" } }) expect(acp.server.selections).toContainEqual({ sessionID: session.sessionId, agent: "copilot-build" }) expect(missing).toMatchObject({ code: -32602, data: { mode: "missing" } }) expect(agentReads(acp)).toBeGreaterThan(reads) }) - test("reloads the catalog once for an unseen model and selects it", async () => { + test.each([ + [ + "a sibling session closes", + async (acp: Wire) => { + const closed = await acp.newSession() + const open = await acp.newSession() + await acp.request("session/close", { sessionId: closed.sessionId }) + return open.sessionId + }, + ], + ...(["session/load", "session/resume"] as const).map( + (method) => + [ + `${method} re-attaches the session`, + async (acp: Wire) => { + const session = await acp.newSession() + const params = { cwd: "/workspace", sessionId: session.sessionId, mcpServers: [] } + await acp.request(method, params) + await acp.request(method, params) + return session.sessionId + }, + ] as const, + ), + ])("pushes exactly one update per catalog change after %s", async (_, setup) => { await using acp = await startWire() acp.server.catalog.models = [testModel] await acp.initialize() - const session = await acp.newSession() - acp.server.catalog.models = [testModel, secondModel] - const set = (value: string) => - acp.request("session/set_config_option", { sessionId: session.sessionId, configId: "model", value }) - const reads = modelReads(acp) - - const selected = await set("test/second-model") - expect(modelReads(acp)).toBe(reads + 1) - expect(await rpcError(set("test/missing-model"))).toMatchObject({ code: -32602 }) - - expect(currentValue(selected, "model")).toBe("test/second-model") - expect(modelReads(acp)).toBe(reads + 2) - expect(acp.server.selections).toEqual([ - { sessionID: session.sessionId, model: { providerID: "test", id: secondModel.id } }, - ]) - }) - - test("stops catalog updates for a closed session while other sessions in the cwd keep them", async () => { - await using acp = await startWire() - acp.server.catalog.models = [testModel] - await acp.initialize() - const closed = await acp.newSession() - const open = await acp.newSession() - await acp.request("session/close", { sessionId: closed.sessionId }) + const sessionId = await setup(acp) const since = acp.updates.length - await change(acp, open.sessionId, "config_option_update", () => { + await change(acp, sessionId, "config_option_update", () => { acp.server.catalog.models = [testModel, secondModel] acp.server.send(ephemeralEvent("model.updated", {})) }) - await change(acp, open.sessionId, "available_commands_update", () => { + await change(acp, sessionId, "available_commands_update", () => { acp.server.catalog.commands = [reviewCommand, { name: "ship", description: "Ship it" }] acp.server.send(ephemeralEvent("command.updated", {}, { directory: "/workspace" })) }) - await change(acp, open.sessionId, "config_option_update", () => { - acp.server.catalog.agents = [buildAgent] - acp.server.send(ephemeralEvent("agent.updated", {}, { directory: "/workspace" })) - }) expect(updateKinds(acp, since)).toEqual([ - [open.sessionId, "config_option_update"], - [open.sessionId, "available_commands_update"], - [open.sessionId, "config_option_update"], + [sessionId, "config_option_update"], + [sessionId, "available_commands_update"], ]) }) - test.each(["session/load", "session/resume"] as const)( - "pushes one update per catalog change after %s re-attaches a session", - async (method) => { - await using acp = await startSession() - const params = { cwd: "/workspace", sessionId: acp.sessionId, mcpServers: [] } - await acp.request(method, params) - await acp.request(method, params) - const since = acp.updates.length - - await change(acp, acp.sessionId, "config_option_update", () => { - acp.server.catalog.models = [testModel] - acp.server.send(ephemeralEvent("model.updated", {})) - }) - await change(acp, acp.sessionId, "available_commands_update", () => { - acp.server.catalog.commands = [reviewCommand, { name: "ship", description: "Ship it" }] - acp.server.send(ephemeralEvent("command.updated", {}, { directory: "/workspace" })) - }) - await change(acp, acp.sessionId, "config_option_update", () => { - acp.server.catalog.agents = [buildAgent] - acp.server.send(ephemeralEvent("agent.updated", {}, { directory: "/workspace" })) - }) - - expect(updateKinds(acp, since)).toEqual([ - [acp.sessionId, "config_option_update"], - [acp.sessionId, "available_commands_update"], - [acp.sessionId, "config_option_update"], - ]) - }, - ) - test.each(["empty", "missing the default"])( "retries when the model list is %s but the default is ready", async (initial) => { @@ -292,20 +259,32 @@ describe("acp catalog and config options over the wire", () => { await using acp = await startSession() const advertised = await acp.waitForUpdate((item) => commandNames(item) !== undefined) - acp.server.catalog.commands = [reviewCommand, { name: "compact", description: "Server compact" }] + acp.server.catalog.commands = [ + reviewCommand, + { name: "compact", description: "Server compact" }, + { name: "ship", description: "Ship it" }, + ] acp.server.send(ephemeralEvent("command.updated", {}, { directory: "/workspace" })) const replaced = await acp.waitForUpdate((item) => item !== advertised && commandNames(item) !== undefined) const compacted = await acp.prompt(acp.sessionId, "/compact") - expect([advertised, replaced].map((item) => item.update)).toEqual( - Array.from({ length: 2 }, () => ({ + expect([advertised, replaced].map((item) => item.update)).toEqual([ + { sessionUpdate: "available_commands_update", availableCommands: [ { name: "review", description: "Review changes" }, { name: "compact", description: "Compact the session" }, ], - })), - ) + }, + { + sessionUpdate: "available_commands_update", + availableCommands: [ + { name: "review", description: "Review changes" }, + { name: "ship", description: "Ship it" }, + { name: "compact", description: "Compact the session" }, + ], + }, + ]) expect(compacted.stopReason).toBe("end_turn") expect(acp.server.submissions.map((item) => item.kind)).toEqual(["compact"]) }) diff --git a/packages/cli/test/acp/session.test.ts b/packages/cli/test/acp/session.test.ts index d80e6eced2..5a55e7630e 100644 --- a/packages/cli/test/acp/session.test.ts +++ b/packages/cli/test/acp/session.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test" import type { McpServer } from "@agentclientprotocol/sdk" import { currentValue } from "./select-options" -import { makeSession, rpcError, secondModel, startSession, startWire } from "./wire-fixture" +import { ephemeralEvent, makeSession, rpcError, secondModel, startSession, startWire, testModel } from "./wire-fixture" describe("acp session lifecycle over the wire", () => { test("initialize advertises capabilities and terminal auth only when the client asks", async () => { @@ -226,4 +226,33 @@ describe("acp session lifecycle over the wire", () => { config: { type: "remote", url: "https://example.com/mcp", headers: { Authorization: "Bearer x" }, oauth: false }, }) }) + test("leaves a session detached when re-attaching it fails", async () => { + const broken: McpServer = { name: "broken", command: "bun", args: [], env: [] } + await using acp = await startWire({ + fetch: (request) => + request.method === "PUT" && request.path === "/api/experimental/mcp/broken" + ? new Response(null, { status: 500 }) + : undefined, + }) + await acp.initialize() + const failed = await acp.newSession() + const other = await acp.newSession() + + expect( + await rpcError( + acp.request("session/resume", { cwd: "/workspace", sessionId: failed.sessionId, mcpServers: [broken] }), + ), + ).toMatchObject({ code: -32603 }) + const since = acp.updates.length + acp.server.catalog.models = [testModel] + acp.server.send(ephemeralEvent("model.updated", {})) + await acp.until(() => acp.updates.length > since, "config options for the attached session") + + expect(acp.updates.slice(since).map((item) => item.sessionId)).toEqual([other.sessionId]) + expect( + await rpcError( + acp.request("session/set_config_option", { sessionId: failed.sessionId, configId: "mode", value: "plan" }), + ), + ).toMatchObject({ code: -32602, data: { sessionId: failed.sessionId } }) + }) })