diff --git a/packages/opencode/src/acp/event.ts b/packages/opencode/src/acp/event.ts index 7d05fa6ee58..dec34b8b662 100644 --- a/packages/opencode/src/acp/event.ts +++ b/packages/opencode/src/acp/event.ts @@ -40,7 +40,10 @@ export class Subscription { private readonly abort = new AbortController() private readonly shellSnapshots = new Map() private readonly toolStarts = new Set() + private readonly connectionWaiters = new Set<() => void>() + private readonly idleWaiters = new Map>>() private readonly permission: ACPPermission.Handler + private connected = false private started = false constructor( @@ -63,10 +66,35 @@ export class Subscription { stop() { this.abort.abort() + this.disconnected() + for (const resolve of this.connectionWaiters) resolve() + this.connectionWaiters.clear() + } + + async runUntilIdle(sessionId: string, request: () => Promise) { + await this.waitUntilConnected() + const waiter = signal() + const waiters = this.idleWaiters.get(sessionId) ?? new Set() + waiters.add(waiter) + this.idleWaiters.set(sessionId, waiters) + + try { + // Idle is queued after the turn's events, and this subscription awaits each update in order. + void waiter.promise.catch(() => {}) + const response = await request() + await waiter.promise + return response + } finally { + waiters.delete(waiter) + if (waiters.size === 0) this.idleWaiters.delete(sessionId) + } } async handle(event: Event) { switch (event.type) { + case "session.status": + if (event.properties.status.type === "idle") this.idle(event.properties.sessionID) + return case "permission.asked": this.permission.handle(event) return @@ -115,19 +143,51 @@ export class Subscription { private async run() { while (!this.abort.signal.aborted) { - const events = (await this.input.sdk.global.event({ - signal: this.abort.signal, - })) as GlobalEventStream - - for await (const event of events.stream) { - if (this.abort.signal.aborted) return - if (!event.payload) continue - await this.handle(event.payload).catch(() => {}) - } + await this.consume().catch(() => {}) + this.disconnected() if (!this.abort.signal.aborted) await new Promise((resolve) => setTimeout(resolve, 1000)) } } + private async consume() { + const events = (await this.input.sdk.global.event({ + signal: this.abort.signal, + })) as GlobalEventStream + this.connected = true + for (const resolve of this.connectionWaiters) resolve() + this.connectionWaiters.clear() + + for await (const event of events.stream) { + if (this.abort.signal.aborted) return + if (!event.payload) continue + await this.handle(event.payload).catch(() => {}) + } + } + + private async waitUntilConnected() { + while (!this.connected) { + if (this.abort.signal.aborted) throw new Error("ACP event subscription stopped") + await new Promise((resolve) => this.connectionWaiters.add(resolve)) + } + } + + private disconnected() { + if (!this.connected) return + this.connected = false + const error = new Error("ACP event stream disconnected") + for (const waiters of this.idleWaiters.values()) { + for (const waiter of waiters) waiter.reject(error) + } + this.idleWaiters.clear() + } + + private idle(sessionId: string) { + const waiters = this.idleWaiters.get(sessionId) + if (!waiters) return + this.idleWaiters.delete(sessionId) + for (const waiter of waiters) waiter.resolve() + } + private async handlePartUpdated(event: EventMessagePartUpdated) { const part = event.properties.part const sessionId = part.sessionID || event.properties.sessionID @@ -339,4 +399,23 @@ export class Subscription { } } +function signal() { + const state: { + resolve: () => void + reject: (reason?: unknown) => void + } = { + resolve: () => {}, + reject: () => {}, + } + const promise = new Promise((resolve, reject) => { + state.resolve = resolve + state.reject = reject + }) + return { + promise, + resolve: () => state.resolve(), + reject: (reason?: unknown) => state.reject(reason), + } +} + export * as ACPEvent from "./event" diff --git a/packages/opencode/src/acp/service.ts b/packages/opencode/src/acp/service.ts index 7257efa7658..fe295918978 100644 --- a/packages/opencode/src/acp/service.ts +++ b/packages/opencode/src/acp/service.ts @@ -88,6 +88,8 @@ export function make(input: { ? ACPEvent.start({ sdk: input.sdk, connection: input.connection, session }) : undefined if (events) input.eventSubscription?.(events) + const runUntilIdle = (sessionId: string, fn: () => Promise) => + events ? events.runUntilIdle(sessionId, fn) : fn() const initialize = Effect.fn("ACP.initialize")(function* (params: InitializeRequest) { const started = performance.now() @@ -504,19 +506,21 @@ export function make(input: { if (!command) { const response = yield* request( () => - input.sdk.session.prompt( - { - sessionID: current.id, - model: { - providerID: selected.providerID, - modelID: selected.modelID, + runUntilIdle(current.id, () => + input.sdk.session.prompt( + { + sessionID: current.id, + model: { + providerID: selected.providerID, + modelID: selected.modelID, + }, + ...(variant ? { variant } : {}), + parts, + ...(modeId ? { agent: modeId } : {}), + directory: current.cwd, }, - ...(variant ? { variant } : {}), - parts, - ...(modeId ? { agent: modeId } : {}), - directory: current.cwd, - }, - { throwOnError: true }, + { throwOnError: true }, + ), ), "session", ) @@ -528,17 +532,19 @@ export function make(input: { if (known) { const response = yield* request( () => - input.sdk.session.command( - { - sessionID: current.id, - command: known.name, - arguments: command.args, - model: `${selected.providerID}/${selected.modelID}`, - ...(variant ? { variant } : {}), - ...(modeId ? { agent: modeId } : {}), - directory: current.cwd, - }, - { throwOnError: true }, + runUntilIdle(current.id, () => + input.sdk.session.command( + { + sessionID: current.id, + command: known.name, + arguments: command.args, + model: `${selected.providerID}/${selected.modelID}`, + ...(variant ? { variant } : {}), + ...(modeId ? { agent: modeId } : {}), + directory: current.cwd, + }, + { throwOnError: true }, + ), ), "session", ) @@ -549,14 +555,16 @@ export function make(input: { if (command.name === "compact") { yield* request( () => - input.sdk.session.summarize( - { - sessionID: current.id, - directory: current.cwd, - providerID: selected.providerID, - modelID: selected.modelID, - }, - { throwOnError: true }, + runUntilIdle(current.id, () => + input.sdk.session.summarize( + { + sessionID: current.id, + directory: current.cwd, + providerID: selected.providerID, + modelID: selected.modelID, + }, + { throwOnError: true }, + ), ), "session", ) diff --git a/packages/opencode/test/acp/service-session.test.ts b/packages/opencode/test/acp/service-session.test.ts index 8dd25492c09..d784709dbb2 100644 --- a/packages/opencode/test/acp/service-session.test.ts +++ b/packages/opencode/test/acp/service-session.test.ts @@ -10,7 +10,7 @@ import type { SessionConfigSelectOption, SetSessionConfigOptionResponse, } from "@agentclientprotocol/sdk" -import type { AssistantMessage, OpencodeClient } from "@opencode-ai/sdk/v2" +import type { AssistantMessage, Event, OpencodeClient } from "@opencode-ai/sdk/v2" import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" import { Effect } from "effect" @@ -24,6 +24,54 @@ const modelID = ModelV2.ID.make("test-model") const configuredModelID = ModelV2.ID.make("configured-model") const secondModelID = ModelV2.ID.make("second-model") +function createEventStream() { + const queue: Event[] = [] + const waiters: Array<(event: Event | undefined) => void> = [] + const push = (event: Event) => { + const waiter = waiters.shift() + if (waiter) return waiter(event) + queue.push(event) + } + const stream = async function* (signal?: AbortSignal) { + while (!signal?.aborted) { + const event = queue.shift() + if (event) { + yield { payload: event } + continue + } + const next = await new Promise((resolve) => { + waiters.push(resolve) + signal?.addEventListener("abort", () => resolve(undefined), { once: true }) + }) + if (!next) return + yield { payload: next } + } + } + return { push, stream } +} + +function idleEvent(sessionID: string): Event { + return { + id: `evt_idle_${sessionID}`, + type: "session.status", + properties: { + sessionID, + status: { type: "idle" }, + }, + } +} + +function deferred() { + const state: { resolve?: (value: A) => void } = {} + const promise = new Promise((resolve) => { + state.resolve = resolve + }) + return { + promise, + resolve: (value: A) => state.resolve?.(value), + } +} + const provider: Provider.Info = { id: providerID, name: "Test", @@ -147,6 +195,7 @@ describe("ACP service sessions", () => { options?: { abort?: (input: { sessionID: string }) => Promise<{ data: boolean }> prompt?: (input: unknown) => Promise<{ data: { info: ReturnType } }> + sessionUpdate?: (update: SessionNotification) => Promise }, ) => { const updates: SessionNotification[] = [] @@ -157,6 +206,7 @@ describe("ACP service sessions", () => { const commands: unknown[] = [] const summarizes: unknown[] = [] const usageUpdates: string[] = [] + const events = createEventStream() const sessions = Array.from({ length: 102 }, (_, index) => ({ id: `ses_${index + 1}`, directory: index % 2 === 0 ? "/workspace" : "/other", @@ -164,6 +214,9 @@ describe("ACP service sessions", () => { time: { created: index + 1, updated: index + 1 }, })) const sdk = { + global: { + event: (input?: { signal?: AbortSignal }) => Promise.resolve({ stream: events.stream(input?.signal) }), + }, config: { providers: () => Promise.resolve({ data: { providers: [provider], default: { test: modelID } } }), get: () => Promise.resolve({ data: {} }), @@ -196,11 +249,9 @@ describe("ACP service sessions", () => { data: input.directory ? sessions.filter((session) => session.directory === input.directory) : sessions, }), messages: () => Promise.resolve({ data: messages }), - prompt: - options?.prompt ?? - ((input: unknown) => { - prompts.push(input) - return Promise.resolve({ + prompt: async (input: { sessionID: string }) => { + const response = await (options?.prompt?.(input) ?? + Promise.resolve({ data: { info: assistantInfo({ input: 100, @@ -209,10 +260,14 @@ describe("ACP service sessions", () => { cache: { read: 11, write: 13 }, }), }, - }) - }), - command: (input: unknown) => { + })) + prompts.push(input) + events.push(idleEvent(input.sessionID)) + return response + }, + command: (input: { sessionID: string }) => { commands.push(input) + events.push(idleEvent(input.sessionID)) return Promise.resolve({ data: { info: assistantInfo({ @@ -224,8 +279,9 @@ describe("ACP service sessions", () => { }, }) }, - summarize: (input: unknown) => { + summarize: (input: { sessionID: string }) => { summarizes.push(input) + events.push(idleEvent(input.sessionID)) return Promise.resolve({ data: true }) }, abort: @@ -249,7 +305,7 @@ describe("ACP service sessions", () => { const connection = { sessionUpdate: (update: SessionNotification) => { updates.push(update) - return Promise.resolve() + return options?.sessionUpdate?.(update) ?? Promise.resolve() }, } as Pick const usage = UsageService.Service.of({ @@ -273,6 +329,7 @@ describe("ACP service sessions", () => { commands, summarizes, usageUpdates, + events, } } @@ -1018,6 +1075,75 @@ describe("ACP service sessions", () => { expect(usageUpdates).toEqual([session.sessionId]) }) + it("waits for queued session updates before returning end_turn", async () => { + const called = deferred() + const response = deferred<{ data: { info: ReturnType } }>() + const update = deferred() + const release = deferred() + const order: string[] = [] + const fixture = makeService([], { + prompt: () => { + called.resolve(undefined) + return response.promise + }, + sessionUpdate: (notification) => { + if (notification.update.sessionUpdate !== "agent_thought_chunk") return Promise.resolve() + update.resolve(undefined) + return release.promise.then(() => { + order.push("update") + }) + }, + }) + const session = await Effect.runPromise(fixture.service.newSession({ cwd: "/workspace", mcpServers: [] })) + const result = Effect.runPromise( + fixture.service.prompt({ sessionId: session.sessionId, prompt: [{ type: "text", text: "hello" }] }), + ).then((value) => { + order.push("response") + return value + }) + + await called.promise + fixture.events.push({ + id: "evt_part", + type: "message.part.updated", + properties: { + sessionID: session.sessionId, + time: Date.now(), + part: { + id: "part_reasoning", + sessionID: session.sessionId, + messageID: "msg_assistant", + type: "reasoning", + text: "", + time: { start: Date.now() }, + }, + }, + }) + fixture.events.push({ + id: "evt_delta", + type: "message.part.delta", + properties: { + sessionID: session.sessionId, + messageID: "msg_assistant", + partID: "part_reasoning", + field: "text", + delta: "thinking", + }, + }) + response.resolve({ + data: { + info: assistantInfo({ input: 1, output: 1, reasoning: 1, cache: { read: 0, write: 0 } }), + }, + }) + + await update.promise + expect(order).toEqual([]) + + release.resolve(undefined) + expect((await result).stopReason).toBe("end_turn") + expect(order).toEqual(["update", "response"]) + }) + it("maps assistant prompt errors to request errors instead of end turn", async () => { const { service } = makeService([], { prompt: () =>