chore(acp): trim comments (#52770)

This commit is contained in:
Shoubhit Dash
2026-10-02 20:50:09 +05:30
committed by GitHub
Unverified
parent f902a0fae7
commit 7440ff7844
15 changed files with 31 additions and 149 deletions
+1 -3
View File
@@ -23,7 +23,7 @@ import { ACPTurn } from "./turn"
export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stream: Stream) {
const run = Effect.runPromiseWith(yield* Effect.context<Scope.Scope>())
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.
// Requests can dispatch before the service below is built.
const ready = yield* Deferred.make<ACPService.Interface>()
const handle =
<Params, A>(
@@ -102,8 +102,6 @@ export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stre
"session/set_mode",
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((service, ctx) => service.prompt(ctx.params, ctx.signal)),
+3 -10
View File
@@ -6,7 +6,6 @@ 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 }],
])
@@ -17,7 +16,6 @@ export type Catalog = {
readonly defaultModel: Model.Ref
readonly modes: ReadonlyArray<{ id: Agent.ID; name: string; description?: string }>
readonly defaultModeID: Agent.ID
/** Server commands, without those shadowed by a built-in. */
readonly commands: ReadonlyArray<Command.Info>
}
@@ -36,11 +34,8 @@ export class LoadError extends Schema.TaggedError<LoadError>()("ACPCatalogLoadEr
export type Error = NotReadyError | LoadError
export interface Interface {
/** Loads a directory's catalog once. Concurrent callers share the load, and a failed load is not cached. */
readonly get: (cwd: string) => Effect.Effect<Catalog, Error>
/** Resolves after a reload that started after the call. A failed reload keeps the previous catalog. */
readonly reload: (cwd: string) => Effect.Effect<void, Error>
/** Emits the current catalog, then each reloaded one. */
readonly changes: (cwd: string) => Stream.Stream<Catalog, Error>
}
@@ -54,7 +49,6 @@ type Entry = {
loaded: number
}
// Provider, integration, and credential changes reach the catalog through model.updated.
const reloadOn = new Set<OpenCodeEvent["type"]>(["model.updated", "agent.updated", "command.updated"])
export const make = Effect.fnUntraced(function* (client: OpenCodeClient) {
@@ -62,8 +56,7 @@ export const make = Effect.fnUntraced(function* (client: OpenCodeClient) {
const entries = new Map<string, Deferred.Deferred<Entry, Error>>()
const connected = yield* Deferred.make<void>()
// A reload covers every request made before it starts, so requests queued behind a running reload share
// one more load. Typed load failures keep the previous catalog and still settle the requests they covered.
// Requests queued behind a running load share the next one.
const reload = (entry: Entry) =>
Effect.suspend(() => {
const target = ++entry.requested
@@ -144,7 +137,7 @@ export const make = Effect.fnUntraced(function* (client: OpenCodeClient) {
const load = (client: OpenCodeClient, cwd: string) =>
read(client, cwd).pipe(
// Some providers discover models in the background after plugin startup begins.
// Providers may still be discovering models after startup.
Effect.retry({
while: (error) => error._tag === "ACPCatalogNotReadyError",
schedule: Schedule.spaced("25 millis").pipe(Schedule.upTo({ duration: "5 seconds" })),
@@ -165,7 +158,7 @@ const read = Effect.fnUntraced(function* (client: OpenCodeClient, cwd: string) {
).pipe(Effect.mapError((cause) => new LoadError({ cause })))
const models = modelResult.data.filter((model) => model.enabled)
const preferred = defaultResult.data
// Parallel reads can straddle initialization; select only from this model list.
// The parallel default read can name a model missing from this list.
const defaultModel = preferred
? models.find((model) => model.providerID === preferred.providerID && model.id === preferred.id)
: models[0]
+1 -5
View File
@@ -5,7 +5,6 @@ import { Effect, Schema } from "effect"
import { HttpClientError } from "effect/unstable/http"
import { ACPError } from "./error"
/** Keeps the OpenCode client failures ACP reports. Any other failure is a defect. */
export function classify(error: unknown): Effect.Effect<never, ACPError.Error> {
if (
error instanceof ClientError &&
@@ -20,16 +19,13 @@ export function classify(error: unknown): Effect.Effect<never, ACPError.Error> {
return Effect.die(error)
}
// The client reads every body through the response stream or buffer, which report a failed read as a `DecodeError`
// carrying the read's cause. Its other `DecodeError`s, for an undeclared status or content type, have no cause, and
// a body that fails to parse or decode fails with a `SchemaError` instead.
// The client reports a failed body read as a `DecodeError` with a cause; its other `DecodeError`s have none.
function isConnectionFailure(error: HttpClientError.HttpClientError) {
return (
error.reason._tag === "TransportError" || (error.reason._tag === "DecodeError" && error.reason.cause !== undefined)
)
}
/** The client rejects a malformed ID before sending it, so it is rejected here as the server would. */
export function decodeSessionID(value: string) {
return Schema.decodeUnknownEffect(Session.ID)(value).pipe(
Effect.mapError(() => new ACPError.InvalidRequestError({ message: "Invalid session ID", field: "sessionID" })),
+5 -15
View File
@@ -5,7 +5,6 @@ import { SessionMessage } from "@opencode/schema/session-message"
const MarkerMeta = "opencode/compaction"
/** Message IDs follow core's compaction message projection, so live compactions match replayed ones. */
export type Started = { readonly status: "started"; readonly messageId: string; readonly reason: "auto" | "manual" }
type Compaction =
@@ -23,7 +22,6 @@ type Compaction =
readonly error: SessionError.Error
}
/** Each session's compaction that has started and not yet settled, by session ID. */
export type Tracked = ReadonlyMap<string, Started>
type LifecycleEvent = Extract<
@@ -35,14 +33,9 @@ type OpeningEvent = Extract<
{ readonly type: "session.compaction.started" | "session.compaction.failed" }
>
// Core reports a cancelled or interrupted compaction as a failure with one of these error types.
const Cancelled = new Set(["aborted", "compaction.interrupted"])
/**
* Whether compactions go out as the standard session updates rather than the `_meta` marker. A child compaction
* projected onto the parent session keeps the marker, since a standard update there would read as compaction of the
* parent's own context.
*/
// Without child updates, a child's standard update would read as the parent's own.
export function usesStandardUpdates(
ctx: { readonly compaction: boolean; readonly childUpdates: boolean },
child: boolean,
@@ -50,7 +43,6 @@ export function usesStandardUpdates(
return ctx.compaction && (!child || ctx.childUpdates)
}
/** Tracks a lifecycle event and returns the updates it sends. */
export function apply(event: LifecycleEvent, tracked: Tracked, standard: boolean) {
const sessionID = event.data.sessionID
if (event.type === "session.compaction.started") {
@@ -70,8 +62,7 @@ export function apply(event: LifecycleEvent, tracked: Tracked, standard: boolean
}
return { tracked: remaining, updates: [update(completed, standard)] }
}
// Automatic compaction can fail before it starts, for example when there is nothing to compact yet. A live
// standard compaction still opens before it settles.
// Auto compaction can fail without a started event; open it first for standard clients.
const started = current ?? open(event)
const failed: Compaction = { ...started, status: "failed", reason: event.data.reason, error: event.data.error }
return {
@@ -80,18 +71,16 @@ export function apply(event: LifecycleEvent, tracked: Tracked, standard: boolean
}
}
/** A summary chunk for the session's tracked compaction, if it has one and uses standard updates. */
export function chunk(started: Started | undefined, text: string, standard: boolean): SessionUpdate | undefined {
if (!started || !standard) return undefined
return { sessionUpdate: "compaction_summary_chunk", compactionId: started.messageId, content: { type: "text", text } }
}
/** Settles a compaction a cancelled turn left open the way core settles a cancelled one. */
export function abandon(started: Started, standard: boolean) {
return update({ ...started, status: "failed", error: { type: "aborted", message: "Compaction cancelled" } }, standard)
}
/** A settled compaction as one terminal update. A running one has no live turn on this connection to settle it. */
// Nothing on this connection would settle a replayed running compaction.
export function replay(message: Extract<SessionMessage.Info, { type: "compaction" }>, standard: boolean) {
if (message.status === "running") return undefined
const base = { messageId: message.id, reason: message.reason }
@@ -103,6 +92,7 @@ export function replay(message: Extract<SessionMessage.Info, { type: "compaction
)
}
// Matches core's compaction message ID, so live and replayed compactions line up.
function open(event: OpeningEvent): Started {
return {
status: "started",
@@ -139,7 +129,7 @@ function update(compaction: Compaction, standard: boolean): SessionUpdate {
}
}
// Core reports a defect as `compaction.failed` with the pretty-printed cause, whose stack trace spans several lines.
// Core reports a defect as `compaction.failed` with a multi-line pretty-printed cause.
function errorMessage(error: SessionError.Error) {
if (!error.message || (error.type === "compaction.failed" && error.message.includes("\n"))) return "Compaction failed"
return error.message
-1
View File
@@ -28,7 +28,6 @@ export type ModelSelection = {
variant?: string
}
/** A session's model and mode. Unset fields follow the catalog defaults. */
export type Selection = {
readonly model?: Model.Ref
readonly modeID?: Agent.ID
-9
View File
@@ -13,33 +13,25 @@ import {
} from "@agentclientprotocol/sdk"
import { Context, Deferred, Effect } from "effect"
/**
* Completes once the response to the request being handled is written, so messages sent afterwards follow it.
* Interrupts when the request fails. Outside a request it completes immediately.
*/
export const Responded = Context.Reference<Effect.Effect<void>>("@opencode/cli/acp/Connection/Responded", {
defaultValue: () => Effect.void,
})
export interface Interface {
readonly sessionUpdate: (params: SessionNotification) => Effect.Effect<void, RequestError>
/** Interruption cancels the client's request. */
readonly requestPermission: (
params: RequestPermissionRequest,
) => Effect.Effect<RequestPermissionResponse, RequestError>
readonly extNotification: (method: string, params: Record<string, unknown>) => Effect.Effect<void, RequestError>
/** Interruption cancels the client's request. */
readonly createElicitation: (
params: CreateElicitationRequest,
) => Effect.Effect<CreateElicitationResponse, RequestError>
/** Tracks an incoming request from now on and returns its `Responded`. */
readonly responded: (requestId: JsonRpcId) => Effect.Effect<void>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/cli/acp/Connection") {}
export function make(app: AgentApp, stream: Stream) {
// Settled as each response is written to the stream, which serializes every outgoing message.
const responses = new Map<JsonRpcId, Deferred.Deferred<void>>()
const writer = stream.writable.getWriter()
const agent = app.connect({
@@ -79,7 +71,6 @@ export function make(app: AgentApp, stream: Stream) {
}
}
// The client's rejections stay typed; any other rejection is a defect.
function promise<A>(evaluate: (signal: AbortSignal) => Promise<A>) {
return Effect.tryPromise({ try: evaluate, catch: (cause) => cause }).pipe(
Effect.catch((cause) => (cause instanceof RequestError ? Effect.fail(cause) : Effect.die(cause))),
+1 -15
View File
@@ -11,12 +11,7 @@ import { ACPError } from "./error"
const key = "opencode.acp.additionalDirectories"
const decodeStored = Schema.decodeUnknownOption(Schema.Array(Schema.String))
/**
* Normalizes additional workspace roots without following symlinks, keeping the spelling tools will see, and
* drops duplicates and roots that resolve to cwd. Glob characters are rejected because permission resources
* would treat them as wildcards. The ACP SDK has already dropped entries that are not strings and replaced a
* value that is not an array with `[]`.
*/
// Permission resources treat `*` and `?` as wildcards.
export const parse = Effect.fnUntraced(function* (cwd: string, directories: readonly string[] = []) {
const invalid = directories.find((directory) => !isAbsolute(directory) || /[*?]/.test(directory))
if (invalid !== undefined) return yield* new ACPError.InvalidAdditionalDirectoryError({ directory: invalid })
@@ -26,24 +21,15 @@ export const parse = Effect.fnUntraced(function* (cwd: string, directories: read
)
})
/** Session create fields that grant these directories. */
export function grant(directories: readonly string[]) {
if (directories.length === 0) return {}
return { permissions: rules(directories), metadata: { [key]: [...directories] } }
}
/** The additional directories ACP last activated for the session, in request order. */
export function list(session: Pick<Session.Info, "metadata">) {
return [...Option.getOrElse(decodeStored(session.metadata?.[key]), () => [])]
}
/**
* Replaces the rules granted for the previously activated directories with grants for these directories.
* Grants persist on the server session, so they also apply when it is used from other clients, and child
* sessions copy them when they are created; a child created earlier keeps roots its parent later dropped.
* Session rules are evaluated after agent and config rules, so a grant overrides config `external_directory`
* rules inside the root, while read, edit, and shell rules still apply.
*/
export const activate = Effect.fnUntraced(function* (
client: OpenCodeClient,
session: Session.Info,
+3 -25
View File
@@ -11,13 +11,12 @@ import { Cause, Effect, Option, Schema } from "effect"
import type { ACPConnection } from "./connection"
import type { ACPService } from "./service"
/** A form with an unbranded ID. */
export type AskedForm = Omit<Form.Info, "id"> & { readonly id: string }
type InputField = Exclude<Form.Field, Form.ExternalField>
type SelectField = Form.StringField | Form.MultiselectField
// Form mode must not collect secrets, so only forms from flows known not to ask for credentials are elicited.
const QuestionKind = "question"
// Form mode must not collect secrets; elicit only flows known to be credential-free.
const ElicitedKind = Schema.Struct({ kind: Schema.Literals([QuestionKind, "websearch.provider"]) })
const Credential = /password|passphrase|secret|token|api[_ -]?key|credential|private[_ -]?key/i
const ToolSource = Schema.Struct({ tool: Schema.Struct({ id: Schema.String }) })
@@ -29,19 +28,12 @@ type Input = {
readonly requestedSchema: ElicitationSchema
readonly clientSessionID: string
readonly child?: { readonly id: string; readonly title?: string }
/** Whether the asking tool call reached the client as a `session/update` tool call. */
readonly toolCallSent: boolean
/** Completes once the form is answered or cancelled elsewhere. */
readonly settled: Effect.Effect<void>
}
type Outcome = Form.Answer | "cancel" | "settled"
/**
* Asks the client, then resolves the form on the server. Once `cancelled` completes, the client's request is
* cancelled or never sent, and the form is cancelled. Once `settled` completes, the client's request is cancelled and
* the server is left alone. Resolving on the server is uninterruptible.
*/
export const reply = Effect.fn("cli.acp.elicitation.reply")(function* (input: Input, cancelled: Effect.Effect<void>) {
yield* Effect.uninterruptibleMask((restore) =>
// The race starts racers in order and stops once one is done, so an earlier cancel never starts the ask.
@@ -62,7 +54,6 @@ export const reply = Effect.fn("cli.acp.elicitation.reply")(function* (input: In
export const UnshownQuestionMessage =
"The question couldn't be shown to the user in this client. Continue without an answer: make reasonable assumptions and state them, or ask the user in your reply if you can't proceed."
/** Cancels a form, interrupting its session when the server can't cancel it. */
function cancel(client: OpenCodeClient, form: Form.Info, message?: string) {
return client.session.form.cancel({ sessionID: form.sessionID, formID: form.id, message }).pipe(
Effect.catchTag(["FormAlreadySettledError", "FormNotFoundError"], () => Effect.void),
@@ -75,14 +66,6 @@ function cancel(client: OpenCodeClient, form: Form.Info, message?: string) {
)
}
/**
* The form-mode schema for a form, or undefined when the form is cancelled instead: the client lacks form
* elicitation, the form is not from an allowed flow or has a field whose key, title, description, or option label
* looks like a credential, or ACP can't represent it faithfully. Unrepresentable forms have `external` fields, `when`
* conditions, a hidden required field without a default, a default outside a field's options, or a free-text answer
* alongside options that must also satisfy `required` or item bounds across both inputs. Hidden fields are not asked
* and answer with their default.
*/
export function requestedSchema(form: AskedForm, capabilities: ACPService.Capabilities): ElicitationSchema | undefined {
if (!capabilities.formElicitation) return undefined
if (Option.isNone(Schema.decodeUnknownOption(ElicitedKind)(form.metadata))) return undefined
@@ -100,17 +83,13 @@ export function requestedSchema(form: AskedForm, capabilities: ACPService.Capabi
}
}
/**
* Cancels a form that `requestedSchema` won't show. A question is cancelled with a message the question tool returns
* to the model, so the turn continues instead of ending as interrupted; other forms are cancelled silently.
*/
// The question tool returns the message to the model, so the turn continues instead of ending as interrupted.
export function cancelUnshown(client: OpenCodeClient, form: Form.Info) {
return cancel(client, form, form.metadata?.kind === QuestionKind ? UnshownQuestionMessage : undefined).pipe(
Effect.uninterruptible,
)
}
/** The answer for an accepted response, or undefined when the user declined, cancelled, or sent invalid content. */
export function answer(form: AskedForm, response: CreateElicitationResponse): Form.Answer | undefined {
if (response.action !== "accept") return undefined
const content = Schema.decodeUnknownOption(Form.Answer)(response.content ?? {})
@@ -206,7 +185,6 @@ function properties(field: InputField): Array<[string, ElicitationPropertySchema
}
}
// A free-text answer next to a field's options is a separate optional property that wins over the selection.
function other(field: SelectField, description: string): [string, ElicitationPropertySchema] {
return [
customKey(field),
@@ -219,7 +197,7 @@ function other(field: SelectField, description: string): [string, ElicitationPro
]
}
// Core rejects an empty string for a required field, so the client is told it needs at least one character.
// Core rejects an empty string for a required field.
function text(field: Form.StringField) {
return {
format: field.format,
+1 -11
View File
@@ -34,11 +34,6 @@ const options: PermissionOption[] = [
const decodeFiles = Schema.decodeUnknownOption(Schema.Array(FileDiff.Info))
/**
* Asks the client, then replies to the server. Once `cancelled` completes, the client's request is cancelled or never
* sent, and the server gets `reject`. The server reply is uninterruptible, so a server that is alive but stuck can
* hold a cancel past `CancelDrainTimeout`; a dead server fails fast.
*/
export const reply = Effect.fn("cli.acp.permission.reply")(function* (input: Input, cancelled: Effect.Effect<void>) {
yield* Effect.uninterruptibleMask((restore) =>
// The race starts racers in order and stops once one is done, so an earlier cancel never starts the ask.
@@ -93,8 +88,7 @@ function prefixedTitle(prefix: string | undefined, title: string | undefined) {
return `${prefix}: ${title}`
}
// Core sends the edit and write previews as file diffs; a diff that no longer applies to the file gets no preview.
// The patch tool's diffs have their common indentation trimmed for display, so its previews come from its own hunks.
// Core trims the patch tool's diffs for display, which breaks `applyPatch`, so its previews come from its own hunks.
const permissionPreviews = Effect.fnUntraced(function* (
toolName: string,
input: ToolInput,
@@ -118,8 +112,6 @@ const permissionPreviews = Effect.fnUntraced(function* (
return previews.flat()
})
// Patch.derive throws when a hunk does not match the current file, and a read can fail; the patch then gets no
// previews.
function patchPreviews(input: ToolInput, cwd: string) {
return Effect.forEach(
patchHunks(input),
@@ -166,8 +158,6 @@ function permissionTitle(toolName: string, input: ToolInput, previews: ReadonlyA
}
}
// Only these actions ask with path resources; an external directory asks for `<dir>/*`, located at the directory.
// Core's wildcard matching treats only `*` and `?` as special.
function permissionLocations(
toolName: string,
input: ToolInput,
+3 -6
View File
@@ -49,7 +49,6 @@ export const AuthMethodID = "opencode-login"
export type Failure = ACPError.Error | RequestError | ACPCatalog.Error
/** What the client advertised in `initialize`. */
export type Capabilities = {
readonly childSessionUpdates: boolean
readonly formElicitation: boolean
@@ -86,7 +85,6 @@ export function make(input: {
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.
const withReload = <A>(attached: Attached, select: Effect.Effect<A, Failure>) => {
const retry = () => input.catalog.reload(attached.cwd).pipe(Effect.andThen(select))
return select.pipe(
@@ -109,7 +107,7 @@ export function make(input: {
}
})
// Both selectors update the selection before switching on the server, so the echoed event diffs to no change.
// Update selection before switching so the echoed event is a no-op.
const selectModel = Effect.fnUntraced(function* (attached: Attached, model: Model.Ref) {
yield* Ref.update(attached.selection, (selection) => ({ ...selection, model }))
yield* input.client.session.switchModel({ sessionID: attached.id, model }).pipe(Effect.catch(ACPClient.classify))
@@ -193,8 +191,7 @@ export function make(input: {
}),
newSession: Effect.fnUntraced(function* (params) {
const directories = yield* ACPDirectories.parse(params.cwd, params.additionalDirectories)
// 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.
// Load first so a catalog failure leaves no session.
yield* input.catalog.get(params.cwd)
const created = yield* input.client.session
.create({ location: { directory: AbsolutePath.make(params.cwd) }, ...ACPDirectories.grant(directories) })
@@ -264,7 +261,7 @@ export function make(input: {
const directories = yield* ACPDirectories.parse(params.cwd, params.additionalDirectories)
const sessionID = yield* ACPClient.decodeSessionID(params.sessionId)
const forked = yield* input.client.session.fork({ sessionID }).pipe(Effect.catch(ACPClient.classify))
// Forks copy the source session's rules, so the request list replaces any inherited grants.
// Forks inherit the source's grants; replace them with this request's.
yield* ACPDirectories.activate(input.client, forked, directories)
const attachment = yield* input.sessions.attach(forked, forked.location.directory, params.mcpServers ?? [])
return { sessionId: attachment.attached.id, configOptions: attachment.configOptions }
+3 -12
View File
@@ -17,11 +17,6 @@ export type Attached = {
}
export interface Interface {
/**
* Attaches a session in its own scope, closing any previous attachment of the same ID. Once the attaching request
* has responded, the scope follows the cwd's catalog and pushes config option and command updates while it is
* open. A failed attach leaves the session detached. Returns the session's config options as of the attach.
*/
readonly attach: (
session: Session.Info,
cwd: string,
@@ -30,10 +25,8 @@ export interface Interface {
{ readonly attached: Attached; readonly configOptions: SessionConfigOption[] },
ACPError.Error | RequestError | ACPCatalog.Error
>
/** Closes the session scope. No-op when the session is not attached. */
readonly detach: (sessionID: string) => Effect.Effect<void>
readonly require: (sessionID: string) => Effect.Effect<Attached, ACPError.SessionNotFoundError>
/** Forks work into this attachment's scope, so it ends on detach or re-attach. Fails once the attachment is gone. */
readonly fork: (attached: Attached, effect: Effect.Effect<void>) => Effect.Effect<void, ACPError.SessionNotFoundError>
}
@@ -42,7 +35,6 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/cl
type Entry = {
readonly attached: Attached
readonly scope: Scope.Closeable
/** Selection changes from other clients, applied by the session's fold. */
readonly selected: Queue.Queue<Selection>
}
@@ -55,7 +47,7 @@ export const make = Effect.fnUntraced(function* (input: {
}) {
const scope = yield* Effect.scope
const sessions = new Map<string, Entry>()
// Kept across re-attachment so resuming with the same servers does not add them again.
// Outlives entries so re-attaching does not re-add servers.
const registeredMcp = new Map<string, Set<string>>()
const connected = yield* Deferred.make<void>()
@@ -147,9 +139,8 @@ export const make = Effect.fnUntraced(function* (input: {
if (replaced) yield* Scope.close(replaced.scope, Exit.void)
yield* registerMcp(entry.attached, mcpServers).pipe(Effect.onError(() => remove(session.id, entry)))
const responded = yield* ACPConnection.Responded
// Updates wait for the response that hands the client this session. `changes` emits the latest catalog
// first, so a reload since `current` is still pushed. One fold applies catalog and selection changes so
// pushes leave the client on the latest pair.
// `changes` replays the latest catalog, so a reload since `current` still pushes.
// One fold keeps catalog and selection consistent.
yield* Effect.gen(function* () {
yield* responded
yield* sendCommands(session.id, current)
-1
View File
@@ -109,7 +109,6 @@ export function completedToolUpdate(input: {
readonly cwd: string
}): ToolCallUpdate {
const normalized = toolContent(input.content)
// Read's model content is a JSON page envelope; show the clean text instead.
const firstText = input.content.find((part) => part.type === "text")
const read = input.toolName.toLocaleLowerCase() === "read" && firstText ? readDisplayText(firstText.text) : undefined
const images = normalized.filter((part) => part.type === "content" && part.content.type === "image")
+3 -12
View File
@@ -23,9 +23,8 @@ export type Context = {
readonly cwd: string
readonly start: TurnStart
readonly childUpdates: boolean
/** Whether the client advertised `session.compaction`, so compactions use the standard session updates. */
readonly compaction: boolean
/** A background consumer follows open children after the parent turn ends; it never writes `session/update`. */
/** Background mode follows open children after the turn ends and never writes `session/update`. */
readonly mode: "turn" | "background"
}
@@ -73,7 +72,6 @@ export type TurnState = {
readonly compactions: ACPCompaction.Tracked
readonly children: ReadonlyMap<string, ChildSession>
readonly openChildren: ReadonlySet<string>
/** Forms asked of the client that the server has not yet answered or cancelled. */
readonly forms: ReadonlySet<string>
readonly finish?: SessionMessage.Assistant["finish"]
readonly usage?: { readonly turn: TokenUsage.Info; readonly last: TokenUsage.Info }
@@ -97,7 +95,6 @@ export type Output =
readonly _tag: "FormAsk"
readonly form: FormEvent["data"]["form"]
readonly child?: ChildSession
/** Whether the form's session sends its tool calls to the client as `session/update` tool calls. */
readonly toolCallSent: boolean
}
| { readonly _tag: "FormSettled"; readonly formID: string }
@@ -340,7 +337,6 @@ export function step(state: TurnState, event: OpenCodeEvent, ctx: Context): Step
}
}
/** The ACP failure a settled turn reports instead of a response, if any. */
export function failure(state: TurnState) {
const error = state.stepError ?? state.executionError
if (error?.type === "provider.auth") return new ACPError.AuthRequiredError()
@@ -368,15 +364,12 @@ export function response(state: TurnState, sessionID: string, terminal: Terminal
: undefined
const error = (state.stepError ?? state.executionError)?.type
const stopReason = resolveStopReason({ terminal, finish: state.finish, error })
// Only an interrupt during backoff leaves a retry pending. Interruption clears the projected retry, so report it here.
// Interruption clears the projected retry, so a retry pending at interrupt is reported here.
const retry = state.retries.get(sessionID)
return { stopReason, ...(usage ? { usage } : {}), _meta: retry ? { [RetryMeta]: retry } : {} }
}
/**
* Fails the tools and cancels the session's compaction a cancelled turn left open, for when the server's wind-down
* never reports them. Child compactions are left to the consumer that follows children after the turn.
*/
// Child compactions are left to the background consumer.
export function abandon(state: TurnState, ctx: Context): Step {
const compaction = state.compactions.get(ctx.sessionID)
return {
@@ -403,7 +396,6 @@ export function abandon(state: TurnState, ctx: Context): Step {
}
}
/** The updates that replay a stored message. */
export function* replayMessage(
message: SessionMessage.Info,
cwd: string,
@@ -519,7 +511,6 @@ function childStatus(ctx: Context, child: ChildSession, value: ChildSessionEvent
]
}
// A background consumer ends once its last open child settles.
function childEnded(
state: TurnState,
ctx: Context,
+6 -21
View File
@@ -36,23 +36,15 @@ import { ACPTranslate } from "./translate"
type Failure = ACPError.Error | RequestError | ACPCatalog.Error
export interface Interface {
/**
* Runs the session's only turn. Cancelling it, including through the request's `$/cancel_request` signal,
* interrupts the turn and still resolves with `stopReason: "cancelled"`.
*/
readonly prompt: (input: PromptRequest, signal: AbortSignal) => Effect.Effect<PromptResponse, Failure>
/** Interrupts the session's active turn and waits for it to settle. No-op when the session is idle. */
readonly cancel: (input: CancelNotification) => Effect.Effect<void>
/** Like `cancel`, but an idle session is still interrupted, since server work can outlive its turn. */
/** Unlike `cancel`, interrupts an idle session too, since server work can outlive its turn. */
readonly close: (sessionID: string) => Effect.Effect<void, ACPError.Error | RequestError>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/cli/acp/Turn") {}
/**
* How long a cancelled turn keeps forwarding the server's wind-down. Core acknowledges an interrupt before its
* cleanup settles, and its shell tool waits 3 seconds before escalating to SIGKILL.
*/
/** Core acknowledges an interrupt before its cleanup settles, and its shell tool waits 3s before SIGKILL. */
export const CancelDrainTimeout = Context.Reference<Duration.Input>("@opencode/cli/acp/Turn/CancelDrainTimeout", {
defaultValue: () => "5 seconds",
})
@@ -68,15 +60,12 @@ type PreparedPrompt = {
type PermissionAsk = Extract<ACPTranslate.Output, { readonly _tag: "PermissionAsk" }>
/** A turn's event feed. It moves to the session scope when the turn ends with children still running. */
type Subscription = {
readonly scope: Scope.Closeable
readonly events: Queue.Dequeue<OpenCodeEvent, unknown>
/** Runs permission and form asks one at a time in ask order, without holding back the rest of the stream. */
/** Asks run serially off the event stream. */
readonly asks: Queue.Queue<Effect.Effect<void, ACPError.Error | RequestError>>
/** Completed when the turn is cancelled; pending and later asks then resolve without the client. */
readonly cancelled: Deferred.Deferred<void>
/** Completed per asked form once the server reports it answered or cancelled. */
readonly forms: Map<string, Deferred.Deferred<void>>
}
@@ -92,7 +81,7 @@ export const make = Effect.fnUntraced(function* (input: {
const turns = yield* FiberMap.make<string, PromptResponse, Failure>()
const subscribe = Effect.fnUntraced(function* () {
// Parented, so it still closes when the session scope it is handed to is already gone.
// Parented to the service scope; the session scope may already be closed.
const subscriptionScope = yield* Scope.fork(scope)
const subscription: Subscription = {
scope: subscriptionScope,
@@ -121,7 +110,6 @@ export const make = Effect.fnUntraced(function* (input: {
),
)
// A turn settles only after the asks it saw have been resolved.
const asksSettled = Effect.fnUntraced(function* (subscription: Subscription) {
const settled = yield* Deferred.make<void>()
yield* Queue.offer(subscription.asks, Deferred.succeed(settled, undefined).pipe(Effect.asVoid))
@@ -241,8 +229,6 @@ export const make = Effect.fnUntraced(function* (input: {
const interruptServer = (sessionID: Session.ID) =>
input.client.session.interrupt({ sessionID }).pipe(Effect.ignoreCause)
// Rejects pending asks, interrupts the server once, then forwards its wind-down until the terminal event or the
// timeout. Tools and a compaction still open at the timeout are settled so the client never shows them running.
const windDown = Effect.fnUntraced(function* (
subscription: Subscription,
ctx: ACPTranslate.Context,
@@ -295,7 +281,7 @@ export const make = Effect.fnUntraced(function* (input: {
const close = Scope.close(subscription.scope, Exit.void)
if (Exit.isFailure(exit) && !Cause.hasInterruptsOnly(exit.cause)) return yield* close
if ((yield* Ref.get(state)).openChildren.size === 0) return yield* close
// Children that outlive a cancelled turn were not cancelled, so their asks still go to the client.
// Children outlive a cancelled turn, so their asks still reach the client.
const cancelled = yield* Deferred.make<void>()
const background = consume({ ...subscription, cancelled }, { ...ctx, mode: "background" }, state).pipe(
Effect.ignore,
@@ -367,7 +353,7 @@ export const make = Effect.fnUntraced(function* (input: {
concurrency: "unbounded",
})
const prompt = preparePrompt(catalog, parts, SessionMessage.ID.create())
// Check and register in one synchronous step.
// Synchronous, so concurrent prompts for one session cannot both register.
const turn = yield* Effect.withFiber((fiber) => {
if (FiberMap.hasUnsafe(turns, attached.id)) {
return Effect.fail(
@@ -419,7 +405,6 @@ function preparePrompt(catalog: Catalog, parts: readonly PromptPart[], messageID
return { start, text, files, synthetic, slash, command }
}
// Covers only missing or permission-denied targets; the server still rejects oversized, non-regular, or unlistable ones.
function referenceUnreadableFile(part: PromptPart) {
if (part.type !== "file" || !part.url.startsWith("file://")) return Effect.succeed(part)
return Effect.tryPromise(() => access(fileURLToPath(part.url), constants.R_OK)).pipe(
+1 -3
View File
@@ -42,10 +42,8 @@ export default Runtime.handler(
Effect.map((reason) => `server exited unexpectedly (${reason})`),
),
)
// Closing the handler scope would wait for the private server's graceful shutdown; its lease pipe already
// ends the server once this process exits.
// Exit directly: closing the scope would wait on the server's graceful shutdown, and its lease pipe ends it anyway.
yield* Effect.sync(() => {
// stdout carries ACP, so the diagnostic goes to stderr.
if (failure) process.stderr.write(`opencode acp: ${failure}\n`)
process.exit(failure ? 1 : 0)
})