diff --git a/packages/cli/src/acp/agent.ts b/packages/cli/src/acp/agent.ts index 8433fb61e38..31a8cc9fcd4 100644 --- a/packages/cli/src/acp/agent.ts +++ b/packages/cli/src/acp/agent.ts @@ -8,10 +8,9 @@ import { type AgentRequestMethod, type Stream, } from "@agentclientprotocol/sdk" -import { ClientError, type OpenCodeClient } from "@opencode/client/promise" -import { Cause, Effect, type Scope } from "effect" +import type { OpenCodeClient } from "@opencode/client/promise" +import { Cause, Deferred, Effect, type Scope } from "effect" import { ACPCatalog } from "./catalog" -import { ACPConnection } from "./connection" import { ACPError } from "./error" import { ACPService } from "./service" @@ -19,11 +18,18 @@ import { ACPService } from "./service" export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stream: Stream) { const run = Effect.runPromiseWith(yield* Effect.context()) const catalog = yield* ACPCatalog.make(client) + // Requests can dispatch once the stream's read loop yields, which may be before the service below is built. + const ready = yield* Deferred.make() const handle = - (call: (ctx: AgentHandlerContext) => Effect.Effect) => + ( + call: (service: ACPService.Interface, ctx: AgentHandlerContext) => Effect.Effect, + ) => (name: string) => { const handler = Effect.fn(name)( - call, + (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.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)))), @@ -42,78 +48,63 @@ export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stre request( "initialize", - handle((ctx) => promise(() => service.initialize(ctx.params))), + handle((service, ctx) => service.initialize(ctx.params)), ) request( "authenticate", - handle((ctx) => promise(() => service.authenticate(ctx.params))), + handle((service, ctx) => service.authenticate(ctx.params)), ) request( "session/new", - handle((ctx) => promise(() => service.newSession(ctx.params))), + handle((service, ctx) => service.newSession(ctx.params)), ) request( "session/load", - handle((ctx) => promise(() => service.loadSession(ctx.params))), + handle((service, ctx) => service.loadSession(ctx.params)), ) request( "session/list", - handle((ctx) => promise(() => service.listSessions(ctx.params))), + handle((service, ctx) => service.listSessions(ctx.params)), ) request( "session/delete", - handle((ctx) => promise(() => service.deleteSession(ctx.params))), + handle((service, ctx) => service.deleteSession(ctx.params)), ) request( "session/resume", - handle((ctx) => promise(() => service.resumeSession(ctx.params))), + handle((service, ctx) => service.resumeSession(ctx.params)), ) request( "session/close", - handle((ctx) => promise(() => service.closeSession(ctx.params))), + handle((service, ctx) => ACPError.promise(() => service.closeSession(ctx.params))), ) request( "session/fork", - handle((ctx) => promise(() => service.forkSession(ctx.params))), + handle((service, ctx) => service.forkSession(ctx.params)), ) request( "session/set_config_option", - handle((ctx) => promise(() => service.setSessionConfigOption(ctx.params))), + handle((service, ctx) => service.setSessionConfigOption(ctx.params)), ) request( "session/set_mode", - handle((ctx) => promise(() => service.setSessionMode(ctx.params))), + handle((service, ctx) => service.setSessionMode(ctx.params)), ) // The SDK signal is passed through rather than interrupting the fiber: a cancelled turn still resolves with // `stopReason: "cancelled"`. request( "session/prompt", - handle((ctx) => promise(() => service.prompt(ctx.params, ctx.signal))), + handle((service, ctx) => ACPError.promise(() => service.prompt(ctx.params, ctx.signal))), ) notification( "session/cancel", - handle((ctx) => promise(() => service.cancel(ctx.params))), + handle((service, ctx) => ACPError.promise(() => service.cancel(ctx.params))), ) const connection = app.connect(stream) - // Inbound dispatch starts after the stream's async read loop yields, so handlers never observe this before assignment. - const service = ACPService.make({ client, connection: ACPConnection.make(connection), catalog, run }) + yield* Deferred.succeed(ready, yield* ACPService.make({ client, connection, catalog, run })) return connection }) const spanName = (method: string) => `cli.acp.${method.replaceAll("/", ".")}` -const promise = (evaluate: () => Promise) => - Effect.tryPromise({ - try: evaluate, - // A catalog load failure is classified by the client error that caused it. - catch: (cause) => (cause instanceof ACPCatalog.LoadError ? cause.cause : cause), - }).pipe( - Effect.catch((cause) => { - 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 ACP from "./agent" diff --git a/packages/cli/src/acp/catalog.ts b/packages/cli/src/acp/catalog.ts index d9c608cdd8d..1f550e11b95 100644 --- a/packages/cli/src/acp/catalog.ts +++ b/packages/cli/src/acp/catalog.ts @@ -1,6 +1,6 @@ import type { CommandInfo, ModelInfo, ModelRef, OpenCodeClient, OpenCodeEvent } from "@opencode/client/promise" import { FSUtil } from "@opencode/util/fs-util" -import { Context, Deferred, Effect, Exit, Schedule, Schema, Scope, Semaphore, Stream, SubscriptionRef } from "effect" +import { Context, Deferred, Effect, Exit, Schedule, Schema, Semaphore, Stream, SubscriptionRef } from "effect" import type { ConfigOptionProvider } from "./config-option" export type Catalog = { @@ -133,52 +133,6 @@ export const make = Effect.fnUntraced(function* (client: OpenCodeClient) { }) }) -export type Live = { - readonly cwd: string - readonly current: Catalog -} - -/** Temporary adapter for the promise-based `ACPService` until sessions consume the service directly. */ -export function promise(input: { - readonly catalog: Interface - readonly run: (effect: Effect.Effect) => Promise - readonly changed: (live: Live, previous: Catalog) => Promise -}) { - const lives = new Map() - return { - get: (cwd: string) => - input.run( - Effect.gen(function* () { - const current = yield* input.catalog.get(cwd) - const key = FSUtil.resolve(cwd) - const existing = lives.get(key) - if (existing) return existing - const live: Live = { - cwd, - // A loaded entry's read never suspends. - get current() { - return Effect.runSync(input.catalog.get(cwd)) - }, - } - lives.set(key, live) - yield* input.catalog.changes(cwd).pipe( - Stream.runFoldEffect( - () => current, - (previous, next) => - next === previous - ? Effect.succeed(previous) - : Effect.promise(() => input.changed(live, previous).catch(() => {})).pipe(Effect.as(next)), - ), - Effect.ignore, - Effect.forkScoped({ startImmediately: true }), - ) - return live - }), - ), - reload: (live: Live) => input.run(input.catalog.reload(live.cwd)), - } -} - const load = (client: OpenCodeClient, cwd: string) => read(client, cwd).pipe( // Some providers discover models in the background after plugin startup begins. diff --git a/packages/cli/src/acp/connection.ts b/packages/cli/src/acp/connection.ts index c1b4d0c7b64..876d7dfa895 100644 --- a/packages/cli/src/acp/connection.ts +++ b/packages/cli/src/acp/connection.ts @@ -1,11 +1,36 @@ import { methods, type AgentConnection, + type RequestError, type RequestPermissionRequest, type RequestPermissionResponse, type SendRequestOptions, type SessionNotification, } from "@agentclientprotocol/sdk" +import { Context, type Effect } from "effect" +import { ACPError } from "./error" + +type Failure = ACPError.Error | RequestError + +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 +} + +export class Service extends Context.Service()("@opencode/cli/acp/Connection") {} + +export function service(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)), + }) +} export type Connection = { readonly signal?: AbortSignal @@ -14,6 +39,7 @@ 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 { return { signal: connection.signal, diff --git a/packages/cli/src/acp/error.ts b/packages/cli/src/acp/error.ts index 6ffc0c88743..7362b322dcd 100644 --- a/packages/cli/src/acp/error.ts +++ b/packages/cli/src/acp/error.ts @@ -1,5 +1,7 @@ import { RequestError } from "@agentclientprotocol/sdk" -import { Schema } from "effect" +import { ClientError } from "@opencode/client/promise" +import { Effect, Schema } from "effect" +import { ACPCatalog } from "./catalog" export class SessionNotFoundError extends Schema.TaggedError()("ACPSessionNotFoundError", { sessionId: Schema.String, @@ -101,6 +103,18 @@ 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/service.ts b/packages/cli/src/acp/service.ts index 7276b25b984..45273e634b7 100644 --- a/packages/cli/src/acp/service.ts +++ b/packages/cli/src/acp/service.ts @@ -1,16 +1,9 @@ -import { isDeepStrictEqual } from "node:util" -import { - isSessionNotFoundError, - type CommandInfo, - type ModelRef, - type OpenCodeClient, - type SessionInfo, - type SessionMessageInfo, -} from "@opencode/client/promise" +import { isSessionNotFoundError, type CommandInfo, type ModelRef, type OpenCodeClient } from "@opencode/client/promise" import { FSUtil } from "@opencode/util/fs-util" -import type { Effect, Scope } from "effect" +import { Effect, Option, Ref, Stream, type Scope } from "effect" import { withTimestampedFallback } from "@opencode/util/session-title-fallback" import type { + AgentConnection, AuthenticateRequest, AuthenticateResponse, AuthMethod, @@ -27,11 +20,11 @@ import type { ListSessionsResponse, LoadSessionRequest, LoadSessionResponse, - McpServer, NewSessionRequest, NewSessionResponse, PromptRequest, PromptResponse, + RequestError, ResumeSessionRequest, ResumeSessionResponse, SetSessionConfigOptionRequest, @@ -41,9 +34,9 @@ import type { } from "@agentclientprotocol/sdk" import { OPENCODE_VERSION } from "../version" import { SessionMessage } from "@opencode/schema/session-message" -import { ACPCatalog, type Catalog } from "./catalog" -import { buildConfigOptions, DEFAULT_VARIANT_VALUE, parseModelSelection } from "./config-option" -import type { ACPConnection } from "./connection" +import type { ACPCatalog, Catalog } from "./catalog" +import { DEFAULT_VARIANT_VALUE, parseModelSelection } from "./config-option" +import { ACPConnection } from "./connection" import { promptContentToParts } from "./content" import { ChildSessionUpdateMethod, @@ -55,22 +48,10 @@ import { type TurnStart, } from "./event" import { ACPError } from "./error" +import { ACPSessions, builtinCommands, type Attached } from "./sessions" export const AuthMethodID = "opencode-login" -// ACP runs these itself; they take precedence over server commands with the same name. -const builtinCommands = new Map([["compact", { description: "Compact the session", start: "compaction" as const }]]) - -// Model and mode are unset while the session follows the server defaults. -type Attached = { - readonly id: string - readonly cwd: string - readonly abort: AbortController - readonly catalog: ACPCatalog.Live - model?: ModelRef - modeID?: string -} - type PreparedPrompt = { readonly start: TurnStart readonly text: string @@ -80,92 +61,114 @@ type PreparedPrompt = { readonly command?: CommandInfo } +export type Failure = ACPError.Error | RequestError | ACPCatalog.Error + export interface Interface { - initialize(input: InitializeRequest): Promise - authenticate(input: AuthenticateRequest): Promise - newSession(input: NewSessionRequest): Promise - loadSession(input: LoadSessionRequest): Promise - listSessions(input: ListSessionsRequest): Promise - deleteSession(input: DeleteSessionRequest): Promise - resumeSession(input: ResumeSessionRequest): Promise + readonly initialize: (input: InitializeRequest) => Effect.Effect + readonly authenticate: (input: AuthenticateRequest) => Effect.Effect + readonly newSession: (input: NewSessionRequest) => Effect.Effect + readonly loadSession: (input: LoadSessionRequest) => Effect.Effect + readonly listSessions: (input: ListSessionsRequest) => Effect.Effect + readonly deleteSession: (input: DeleteSessionRequest) => Effect.Effect + readonly resumeSession: (input: ResumeSessionRequest) => Effect.Effect + readonly forkSession: (input: ForkSessionRequest) => Effect.Effect + readonly setSessionConfigOption: ( + input: SetSessionConfigOptionRequest, + ) => Effect.Effect + readonly setSessionMode: (input: SetSessionModeRequest) => Effect.Effect closeSession(input: CloseSessionRequest): Promise - forkSession(input: ForkSessionRequest): Promise - setSessionConfigOption(input: SetSessionConfigOptionRequest): Promise - setSessionMode(input: SetSessionModeRequest): Promise prompt(input: PromptRequest, signal?: AbortSignal): Promise cancel(input: CancelNotification): Promise } -export function make(input: { +export const make = Effect.fnUntraced(function* (input: { readonly client: OpenCodeClient - readonly connection: ACPConnection.Connection + readonly connection: AgentConnection readonly catalog: ACPCatalog.Interface readonly run: (effect: Effect.Effect) => Promise -}): Interface { - const sessions = new Map() - const registeredMcp = new Map>() +}) { + 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 capabilities = { childSessionUpdates: false } - const catalogs = ACPCatalog.promise({ - catalog: input.catalog, - run: input.run, - changed: (live, previous) => - Promise.all( - Array.from(sessions.values()) - .filter((state) => state.catalog === live) - .map(async (state) => { - const options = configOptions(state) - if (!isDeepStrictEqual(options, configOptions(state, previous))) { - await input.connection.sessionUpdate({ - sessionId: state.id, - update: { sessionUpdate: "config_option_update", configOptions: options }, - }) - } - if (!isDeepStrictEqual(live.current.commands, previous.commands)) await sendCommands(state) - }), - ), + const configOptions = Effect.fnUntraced(function* (attached: Attached) { + const catalog = yield* input.catalog.get(attached.cwd) + return ACPSessions.configOptions(catalog, yield* Ref.get(attached.selection)) }) - const sendCommands = (state: Attached) => - input.connection.sessionUpdate({ - sessionId: state.id, + // A selection the catalog has not seen may be new on the server, so reload once before rejecting it. + const withReload = (attached: Attached, select: Effect.Effect) => { + const retry = () => input.catalog.reload(attached.cwd).pipe(Effect.andThen(select)) + return select.pipe( + Effect.catchTags({ ACPInvalidModelError: retry, ACPInvalidModeError: retry, ACPInvalidEffortError: retry }), + ) + } + + 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) + yield* Ref.update(attached.selection, (selection) => ({ ...selection, model })) + yield* ACPError.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 })) + }) + + const getSession = Effect.fnUntraced(function* (sessionID: string, cwd: string) { + const session = yield* ACPError.promise(() => input.client.session.get({ sessionID }).catch(notFound(sessionID))) + 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 }) + : 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( + Stream.runCollect, + Effect.flatMap((messages) => + ACPError.promise(() => replayMessages(turnConnection, 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: "available_commands_update", - availableCommands: [ - ...state.catalog.current.commands - .filter((command) => !builtinCommands.has(command.name)) - .map((command) => ({ name: command.name, description: command.description ?? "" })), - ...Array.from(builtinCommands, ([name, command]) => ({ name, description: command.description })), - ], + sessionUpdate: "usage_update", + used, + size: model.limit.context, + cost: { amount: info.cost, currency: "USD" }, }, }) - - const withReload = (state: Attached, select: () => Promise) => - select().catch(async (error: unknown) => { - if ( - !( - error instanceof ACPError.InvalidModelError || - error instanceof ACPError.InvalidModeError || - error instanceof ACPError.InvalidEffortError - ) - ) - throw error - await catalogs.reload(state.catalog) - return select() - }) - - const requireSession = async (sessionID: string) => { - const current = sessions.get(sessionID) - if (current) return current - throw new ACPError.SessionNotFoundError({ sessionId: sessionID }) - } - - const detach = (sessionID: string) => { - sessions.get(sessionID)?.abort.abort() - sessions.delete(sessionID) - registeredMcp.delete(sessionID) - } + }) const cancelTurn = (sessionID: string) => { const turn = active.get(sessionID) @@ -176,41 +179,11 @@ export function make(input: { return input.client.session.interrupt({ sessionID }) } - const attach = async (session: SessionInfo, cwd: string, mcpServers: readonly McpServer[]) => { - const catalog = await catalogs.get(cwd) - sessions.get(session.id)?.abort.abort() - const state: Attached = { - id: session.id, - cwd, - abort: new AbortController(), - catalog, - model: session.model, - modeID: session.agent, - } - sessions.set(session.id, state) - await registerMcpServers(input.client, registeredMcp, state, mcpServers) - await sendCommands(state) - 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, catalog = state.catalog.current) => { - const model = currentModel(state, catalog) - return buildConfigOptions({ - providers: catalog.providers, - currentModel: { providerID: model.providerID, modelID: model.id }, - currentVariant: model.variant, - modes: catalog.modes, - currentModeId: state.modeID ?? catalog.defaultModeID, - }) - } - return { - initialize: async (params) => { - capabilities.childSessionUpdates = params.clientCapabilities?._meta?.[ChildSessionUpdatesCapability] === true + initialize: Effect.fnUntraced(function* (params) { + yield* Ref.set(capabilities, { + childSessionUpdates: params.clientCapabilities?._meta?.[ChildSessionUpdatesCapability] === true, + }) const authMethod: AuthMethod = { description: "Run `opencode auth login` in the terminal", name: "Login with opencode", @@ -233,32 +206,37 @@ export function make(input: { authMethods: [authMethod], agentInfo: { name: "OpenCode", version: OPENCODE_VERSION }, } - }, - authenticate: async (params) => { - if (params.methodId !== AuthMethodID) throw new ACPError.UnknownAuthMethodError({ methodId: params.methodId }) + }), + authenticate: Effect.fnUntraced(function* (params) { + if (params.methodId !== AuthMethodID) + return yield* new ACPError.UnknownAuthMethodError({ methodId: params.methodId }) return {} - }, - newSession: async (params) => { + }), + newSession: Effect.fnUntraced(function* (params) { // 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. - await catalogs.get(params.cwd) - const created = await input.client.session.create({ location: { directory: params.cwd } }) - const state = await attach(created, params.cwd, params.mcpServers) - return { sessionId: state.id, configOptions: configOptions(state) } - }, - loadSession: async (params) => { - const session = await getSession(input.client, params.sessionId, params.cwd) - const state = await attach(session, session.location.directory, params.mcpServers) - await replay(state) - return { configOptions: configOptions(state) } - }, - listSessions: async (params) => { - const page = await input.client.session.list({ - ...(params.cwd ? { directory: params.cwd } : {}), - order: "desc", - limit: 100, - ...(params.cursor ? { cursor: params.cursor } : {}), - }) + yield* input.catalog.get(params.cwd) + const created = yield* ACPError.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) } + }), + loadSession: Effect.fnUntraced(function* (params) { + const session = yield* getSession(params.sessionId, params.cwd) + const attached = yield* sessions.attach(session, session.location.directory, params.mcpServers) + yield* replay(attached) + return { configOptions: yield* configOptions(attached) } + }), + listSessions: Effect.fnUntraced(function* (params) { + const page = yield* ACPError.promise(() => + input.client.session.list({ + ...(params.cwd ? { directory: params.cwd } : {}), + order: "desc", + limit: 100, + ...(params.cursor ? { cursor: params.cursor } : {}), + }), + ) return { sessions: page.data.map((session) => ({ sessionId: session.id, @@ -268,74 +246,59 @@ export function make(input: { })), ...(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) + }), + 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) return {} - }, - resumeSession: async (params) => { - const session = await getSession(input.client, params.sessionId, params.cwd) - const state = await attach(session, session.location.directory, params.mcpServers ?? []) - return { configOptions: configOptions(state) } - }, + }), + 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) } + }), + 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 ?? []) + yield* replay(attached) + return { sessionId: attached.id, configOptions: yield* configOptions(attached) } + }), + setSessionConfigOption: Effect.fnUntraced(function* (params) { + const attached = yield* 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) } + }), + setSessionMode: Effect.fnUntraced(function* (params) { + const attached = yield* 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(() => {}) - detach(params.sessionId) - return {} - }, - forkSession: async (params) => { - const forked = await input.client.session.fork({ - sessionID: params.sessionId, - }) - 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) - const value = params.value - if (typeof value !== "string") throw new ACPError.InvalidConfigOptionError({ configId: params.configId }) - await withReload(state, async () => { - switch (params.configId) { - case "model": { - const selected = requireModel(state.catalog.current, value, currentModel(state)) - state.model = selected - await input.client.session.switchModel({ sessionID: state.id, model: selected }) - return - } - case "effort": { - const current = currentModel(state) - const model = state.catalog.current.models.find( - (item) => item.providerID === current.providerID && item.id === current.id, - ) - if (!model || (value !== DEFAULT_VARIANT_VALUE && !model.variants.some((variant) => variant.id === value))) - throw new ACPError.InvalidEffortError({ effort: value }) - state.model = { ...current, variant: value } - await input.client.session.switchModel({ sessionID: state.id, model: state.model }) - return - } - case "mode": - return selectMode(input.client, state, value) - default: - throw new ACPError.InvalidConfigOptionError({ configId: params.configId }) - } - }) - return { configOptions: configOptions(state) } - }, - setSessionMode: async (params) => { - const state = await requireSession(params.sessionId) - await withReload(state, () => selectMode(input.client, state, params.modeId)) + await input.run(sessions.detach(params.sessionId)) return {} }, prompt: async (params, signal) => { - const state = await requireSession(params.sessionId) + // 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}`, @@ -343,30 +306,30 @@ export function make(input: { }) } const messageID = SessionMessage.ID.create() - const prepared = preparePrompt(state.catalog.current, params.prompt, messageID) + const prepared = preparePrompt(resolved.catalog, params.prompt, messageID) const control: TurnControl = { cancelled: false, admission: new AbortController() } - const extNotification = input.connection.extNotification + const extNotification = turnConnection.extNotification const childSessionUpdate = - capabilities.childSessionUpdates && extNotification + 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, + connection: turnConnection, sessionID: state.id, cwd: state.cwd, start: prepared.start, action: prepared.command !== undefined, control, - connectionSignal: input.connection.signal, - sessionSignal: state.abort.signal, + connectionSignal: turnConnection.signal, + sessionSignal: state.signal, submit: (signal) => submitPrompt(input.client, state, prepared, signal), ...(childSessionUpdate ? { childSessionUpdate } : {}), }) .then(async (result) => { - await sendUsageUpdate(input.client, input.connection, state, result.contextTokens).catch(() => {}) + await input.run(sendUsageUpdate(state, result.contextTokens)).catch(() => {}) return result.response }) .finally(() => { @@ -382,6 +345,12 @@ export function make(input: { 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 } } @@ -432,14 +401,14 @@ function turnStart(messageID: string, slash: PreparedPrompt["slash"]): TurnStart return { type: "input", id: messageID } } -function requireModel(catalog: Catalog, modelID: string, current: ModelRef): ModelRef { +const requireModel = Effect.fnUntraced(function* (catalog: Catalog, modelID: string, current: ModelRef) { const selected = parseModelSelection(modelID, catalog.providers) const model = catalog.models.find( (item) => item.providerID === selected.model.providerID && item.id === selected.model.modelID, ) - if (!model) throw new ACPError.InvalidModelError({ providerId: selected.model.providerID, modelId: modelID }) + if (!model) return yield* new ACPError.InvalidModelError({ providerId: selected.model.providerID, modelId: modelID }) if (selected.variant && !model.variants.some((variant) => variant.id === selected.variant)) - throw new ACPError.InvalidEffortError({ effort: selected.variant }) + return yield* new ACPError.InvalidEffortError({ effort: selected.variant }) const variant = selected.variant ?? (current.providerID === model.providerID && @@ -447,117 +416,15 @@ function requireModel(catalog: Catalog, modelID: string, current: ModelRef): Mod (current.variant === DEFAULT_VARIANT_VALUE || model.variants.some((variant) => variant.id === current.variant)) ? current.variant : undefined) - return { providerID: model.providerID, id: model.id, variant } -} + return { providerID: model.providerID, id: model.id, variant } satisfies ModelRef +}) -function currentModel(state: Attached, catalog = state.catalog.current) { - return state.model ?? catalog.defaultModel -} - -async function selectMode(client: OpenCodeClient, state: Attached, modeID: string) { - if (!state.catalog.current.modes.some((mode) => mode.id === modeID)) - throw new ACPError.InvalidModeError({ mode: modeID }) - state.modeID = modeID - await client.session.switchAgent({ sessionID: state.id, agent: modeID }) -} - -async function getSession(client: OpenCodeClient, sessionID: string, cwd: string) { - const session = await client.session.get({ sessionID }).catch((error) => { - if (isSessionNotFoundError(error)) throw new ACPError.SessionNotFoundError({ sessionId: sessionID }) - throw error - }) - if (FSUtil.resolve(cwd) !== FSUtil.resolve(session.location.directory)) { - throw new ACPError.SessionDirectoryMismatchError({ sessionId: sessionID, cwd }) - } - return session -} - -async function messages(client: OpenCodeClient, sessionID: string) { - const result: SessionMessageInfo[] = [] - let cursor: string | undefined - do { - const page = cursor - ? await client.message.list({ sessionID, limit: 200, cursor }) - : await client.message.list({ sessionID, limit: 200, order: "asc" }) - result.push(...page.data) - cursor = page.cursor.next ?? undefined - } while (cursor) - return result -} - -async function registerMcpServers( - client: OpenCodeClient, - registered: Map>, - session: Attached, - servers: readonly McpServer[], -) { - const current = registered.get(session.id) ?? new Set() - registered.set(session.id, current) - await Promise.all( - servers.flatMap((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 - }), - ] - }), - ) -} - -function mcpConfig(server: McpServer) { - if ("type" in server) { - if (server.type === "acp") throw new Error("MCP-over-ACP is not supported") - return { - type: "remote" as const, - url: server.url, - headers: Object.fromEntries(server.headers.map((header) => [header.name, header.value])), - oauth: false as const, - } - } - return { - type: "local" as const, - command: [server.command, ...server.args], - environment: Object.fromEntries(server.env.map((entry) => [entry.name, entry.value])), - } -} - -function stableStringify(value: unknown): string { - if (Array.isArray(value)) return `[${value.map(stableStringify).join(",")}]` - if (!value || typeof value !== "object") return JSON.stringify(value) - return `{${Object.entries(value) - .toSorted(([a], [b]) => a.localeCompare(b)) - .map(([key, item]) => `${JSON.stringify(key)}:${stableStringify(item)}`) - .join(",")}}` -} - -async function sendUsageUpdate( - client: OpenCodeClient, - connection: ACPConnection.Connection, - session: Attached, - used?: number, -) { - if (!used) return - const current = currentModel(session) - const model = session.catalog.current.models.find( - (item) => item.providerID === current.providerID && item.id === current.id, - ) - if (!model?.limit.context) return - const info = await client.session.get({ sessionID: session.id }) - await connection.sessionUpdate({ - sessionId: session.id, - update: { - sessionUpdate: "usage_update", - used, - size: model.limit.context, - cost: { amount: info.cost, currency: "USD" }, - }, - }) -} +const requireEffort = Effect.fnUntraced(function* (catalog: Catalog, effort: string, current: ModelRef) { + const model = catalog.models.find((item) => item.providerID === current.providerID && item.id === current.id) + if (!model || (effort !== DEFAULT_VARIANT_VALUE && !model.variants.some((variant) => variant.id === effort))) + return yield* new ACPError.InvalidEffortError({ effort }) + return { ...current, variant: effort } satisfies ModelRef +}) function detectSlashCommand(text: string): { readonly name: string; readonly args: string } | undefined { const value = text.trim() diff --git a/packages/cli/src/acp/sessions.ts b/packages/cli/src/acp/sessions.ts new file mode 100644 index 00000000000..5f50aadca2b --- /dev/null +++ b/packages/cli/src/acp/sessions.ts @@ -0,0 +1,196 @@ +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 { ACPCatalog, Catalog } from "./catalog" +import { buildConfigOptions } 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 +} + +export type Attached = { + readonly id: string + readonly cwd: string + readonly selection: Ref.Ref + /** Aborted when the session detaches, for the promise-based turn. */ + readonly signal: AbortSignal +} + +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. + */ + readonly attach: ( + session: SessionInfo, + cwd: string, + mcpServers: readonly McpServer[], + ) => Effect.Effect + /** Closes the session scope. No-op when the session is not attached. */ + readonly detach: (sessionID: string) => Effect.Effect + readonly require: (sessionID: string) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/cli/acp/Sessions") {} + +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() + // 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 })), + ], + }, + }) + + const changed = Effect.fnUntraced(function* (attached: Attached, previous: Catalog, next: Catalog) { + const selection = yield* Ref.get(attached.selection) + const options = configOptions(next, selection) + if (!isDeepStrictEqual(options, configOptions(previous, selection))) { + yield* input.connection.sessionUpdate({ + sessionId: attached.id, + update: { sessionUpdate: "config_option_update", configOptions: options }, + }) + } + if (!isDeepStrictEqual(next.commands, previous.commands)) yield* sendCommands(attached.id, next) + }) + + const registerMcp = (attached: Attached, servers: readonly McpServer[]) => + Effect.suspend(() => { + const registered = registeredMcp.get(attached.id) ?? new Set() + registeredMcp.set(attached.id, registered) + return Effect.forEach( + servers, + (server) => + Effect.suspend(() => { + const config = mcpConfig(server) + const key = `${server.name}:${stableStringify(config)}` + if (registered.has(key)) return Effect.void + registered.add(key) + return ACPError.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)), + ), + ) + }), + { concurrency: "unbounded", discard: true }, + ) + }) + + 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, + } + // 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 }) + if (replaced) yield* Scope.close(replaced.scope, Exit.void) + yield* Scope.addFinalizer( + sessionScope, + Effect.sync(() => abort.abort()), + ) + 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)), + ), + Effect.ignore, + Effect.forkIn(sessionScope), + ) + yield* registerMcp(attached, mcpServers) + yield* sendCommands(attached.id, yield* input.catalog.get(cwd)) + return 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) + }), + require: Effect.fn("cli.acp.sessions.require")(function* (sessionID) { + const entry = sessions.get(sessionID) + if (!entry) return yield* new ACPError.SessionNotFoundError({ sessionId: sessionID }) + return entry.attached + }), + }) +}) + +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") + return { + type: "remote" as const, + url: server.url, + headers: Object.fromEntries(server.headers.map((header) => [header.name, header.value])), + oauth: false as const, + } + } + return { + type: "local" as const, + command: [server.command, ...server.args], + environment: Object.fromEntries(server.env.map((entry) => [entry.name, entry.value])), + } +} + +function stableStringify(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(stableStringify).join(",")}]` + if (!value || typeof value !== "object") return JSON.stringify(value) + return `{${Object.entries(value) + .toSorted(([a], [b]) => a.localeCompare(b)) + .map(([key, item]) => `${JSON.stringify(key)}:${stableStringify(item)}`) + .join(",")}}` +} + +export * as ACPSessions from "./sessions" diff --git a/packages/cli/test/acp/catalog.test.ts b/packages/cli/test/acp/catalog.test.ts index b6e237bcb35..822c85a0672 100644 --- a/packages/cli/test/acp/catalog.test.ts +++ b/packages/cli/test/acp/catalog.test.ts @@ -136,6 +136,86 @@ describe("acp catalog and config options over the wire", () => { expect(agentReads(acp)).toBeGreaterThan(reads) }) + test("reloads the catalog once for an unseen model and selects it", async () => { + 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 since = acp.updates.length + + await change(acp, open.sessionId, "config_option_update", () => { + acp.server.catalog.models = [testModel, secondModel] + acp.server.send(ephemeralEvent("model.updated", {})) + }) + await change(acp, open.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"], + ]) + }) + + 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) => { @@ -231,6 +311,21 @@ describe("acp catalog and config options over the wire", () => { }) }) +// Each change waits on a catalog reload over HTTP, so a stray update for an earlier change lands before the next one. +async function change(acp: Wire, sessionId: string, kind: string, trigger: () => void) { + const seen = acp.updates.filter((item) => item.sessionId === sessionId && item.update.sessionUpdate === kind).length + trigger() + await acp.until( + () => + acp.updates.filter((item) => item.sessionId === sessionId && item.update.sessionUpdate === kind).length > seen, + kind, + ) +} + +function updateKinds(acp: Wire, since: number) { + return acp.updates.slice(since).map((item) => [item.sessionId, item.update.sessionUpdate]) +} + function commandNames(item: SessionNotification) { if (item.update.sessionUpdate !== "available_commands_update") return undefined return item.update.availableCommands.map((command) => command.name)