diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index f63a220fb5191..dc58382238693 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -59,6 +59,7 @@ import type * as EffectAcpSchema from "effect-acp/compat"; import { writeFakeCli } from "@t3tools/provider-testing/fakeCli"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; +import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime"; import { @@ -102,6 +103,7 @@ import { const DEFAULT_GROK_SETTINGS = Schema.decodeSync(GrokSettings)({}); const DROID_SETTINGS = Schema.decodeSync(AcpRegistrySettings)({ agentId: "factory-droid" }); +const COPILOT_SETTINGS = Schema.decodeSync(AcpRegistrySettings)({ agentId: "github-copilot-cli" }); const layerHost = TestProviderHost.layer().pipe( Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), @@ -3125,6 +3127,1046 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect("settles Copilot foreground idle while the original prompt remains pending", () => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const fileSystem = yield* FileSystem.FileSystem; + const directory = yield* fileSystem.makeTempDirectoryScoped({ prefix: "copilot-lifecycle-" }); + const mockAgentPath = path.join(directory, "agent.cjs"); + yield* fileSystem.writeFileString( + mockAgentPath, + ` + const readline = require("node:readline"); + const send = value => process.stdout.write(JSON.stringify(value) + "\\n"); + const promptIds = []; + let cancellations = 0; + readline.createInterface({ input: process.stdin }).on("line", line => { + const frame = JSON.parse(line); + const respond = result => send({ jsonrpc: "2.0", id: frame.id, result }); + if (frame.method === "initialize") respond({ protocolVersion: 1, agentCapabilities: {} }); + else if (frame.method === "session/new") respond({ sessionId: "mock-session-1" }); + else if (frame.method === "session/prompt") promptIds.push(frame.id); + else if (frame.method === "_test/emit") { + send({ jsonrpc: "2.0", method: "github.com/copilot/sessionEvent", + params: { sessionId: "mock-session-1", ...frame.params } }); + respond({}); + } else if (frame.method === "_test/update") { + send({ jsonrpc: "2.0", method: "session/update", params: { sessionId: "mock-session-1", update: frame.params } }); + respond({}); + } else if (frame.method === "session/cancel") { + cancellations += 1; + } else if (frame.method === "session/close") { + send({ jsonrpc: "2.0", id: frame.id, error: { code: -32601, message: "unsupported" } }); + } else if (frame.method === "_test/cancellations") { + respond({ cancellations }); + } else if (frame.method === "_test/release") { + send({ jsonrpc: "2.0", id: promptIds[frame.params.index], result: { stopReason: "end_turn" } }); + respond({}); + } + }); + `, + ); + const promptDispatched = yield* Deferred.make(); + const nextPromptDispatched = yield* Deferred.make(); + const firstPromptReturned = yield* Deferred.make(); + const continuations = yield* Queue.unbounded(); + let prompts = 0; + const promptTexts: string[] = []; + let nativeRuntime: AcpSessionRuntime.AcpSessionRuntime["Service"] | undefined; + const instanceId = ProviderInstanceId.make("copilot-foreground-idle"); + const adapter = yield* makeAcpRegistryAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + settings: COPILOT_SETTINGS, + environment: {}, + continuationRequests: { + offer: () => Queue.offer(continuations, undefined).pipe(Effect.asVoid), + }, + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => { + nativeRuntime = runtime; + return { + ...runtime, + prompt: (payload, options) => + Effect.gen(function* () { + const text = payload.prompt.find((block) => block.type === "text"); + if (text?.type !== "text") return yield* Effect.die("missing prompt text"); + promptTexts.push(text.text); + prompts += 1; + const ordinal = prompts; + const fiber = yield* runtime.prompt(payload, options).pipe(Effect.forkChild); + if (options?.dispatched) yield* Deferred.await(options.dispatched); + yield* Deferred.succeed( + ordinal === 1 ? promptDispatched : nextPromptDispatched, + undefined, + ); + const result = yield* Fiber.join(fiber); + if (ordinal === 1) yield* Deferred.succeed(firstPromptReturned, undefined); + return result; + }), + }; + }, + }), + }).pipe( + Effect.provide( + Layer.mock(AcpRegistryCatalog)({ resolve: () => Effect.die("runtime is injected") }), + ), + ); + const threadId = ThreadId.make("copilot-foreground-idle-thread"); + const modelSelection = { instanceId, model: "default" } as const; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("copilot-foreground-idle-session"), + modelSelection, + runtimePolicy, + }); + const events = yield* Queue.unbounded(); + const projected: ProviderAdapter.ProviderAdapterV2Event[] = []; + yield* runtime.events.pipe( + Stream.runForEach((item) => + Effect.sync(() => projected.push(item)).pipe(Effect.andThen(Queue.offer(events, item))), + ), + Effect.forkScoped, + ); + const terminal = () => + Stream.fromQueue(events).pipe( + Stream.filter((item) => item.type === "turn.terminal"), + Stream.take(1), + Stream.runCollect, + ); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + const turn = (ordinal: number) => + makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal }); + yield* runtime.startTurn(turn(1)); + yield* Deferred.await(promptDispatched); + assert.isDefined(nativeRuntime); + const event = (type: string, agentId?: string, data?: object) => + nativeRuntime!.request("_test/emit", { + type, + ...(agentId ? { agentId } : {}), + ...(data ? { data } : {}), + }); + yield* event("session.idle"); + assert.isTrue( + Option.isNone(yield* Queue.poll(continuations)), + "idle without a wake must not create a continuation", + ); + yield* event("user.message", undefined, { content: promptTexts.at(-1)!, delivery: "idle" }); + yield* event("assistant.turn_start"); + yield* event("assistant.idle", "child-agent"); + assert.isTrue(Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(2))))); + yield* event("assistant.idle"); + assert.equal((yield* terminal())[0]?.status, "completed"); + yield* event("assistant.turn_start"); + yield* Queue.take(continuations); + const continuation = makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now, + ordinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }); + const next = yield* Effect.exit(runtime.startTurn(continuation)); + assert.isTrue(Exit.isSuccess(next), "root idle must settle before the prompt RPC returns"); + yield* event("assistant.idle", "child-agent"); + assert.isTrue(Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(3))))); + yield* event("assistant.idle"); + assert.equal((yield* terminal())[0]?.status, "completed"); + yield* event("session.idle"); + yield* runtime.startTurn(turn(3)); + yield* Deferred.await(nextPromptDispatched); + yield* nativeRuntime!.request("_test/release", { index: 0 }); + yield* Deferred.await(firstPromptReturned); + yield* event("assistant.turn_start"); + yield* nativeRuntime!.request("_test/update", { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "background wake" }, + }); + yield* event("assistant.idle"); + assert.isTrue( + Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), + "background wake must not accept the pending user prompt", + ); + yield* event("user.message", undefined, { + content: promptTexts.at(-1)!, + source: "agent-child", + }); + yield* event("assistant.idle"); + assert.isTrue( + Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), + "injected messages must not accept the user prompt", + ); + yield* event("user.message", undefined, { content: 42 }); + yield* event("user.message", undefined, {}); + yield* event("assistant.idle"); + assert.isTrue( + Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), + "invalid messages must not accept the pending prompt", + ); + yield* event("user.message", undefined, { content: "different request", delivery: "idle" }); + yield* event("assistant.idle"); + assert.isTrue( + Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), + "a different message must not accept the pending prompt", + ); + yield* event("user.message", undefined, { + content: promptTexts.at(-1)!, + isAutopilotContinuation: true, + }); + yield* event("assistant.idle"); + assert.isTrue( + Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), + "autopilot messages must not accept the pending prompt", + ); + yield* event("user.message", "child-agent", { content: promptTexts.at(-1)! }); + yield* event("assistant.idle"); + assert.isTrue( + Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), + "child messages must not accept the pending prompt", + ); + yield* event("user.message", undefined, { content: promptTexts.at(-1)!, delivery: "idle" }); + assert.isTrue( + Option.isSome(yield* Queue.poll(continuations)), + "acceptance must offer the buffered wake continuation", + ); + yield* event("assistant.turn_start"); + yield* event("assistant.idle", undefined, { aborted: true }); + const cancelled = (yield* terminal())[0]; + assert.equal(cancelled?.status, "cancelled"); + if (cancelled?.type !== "turn.terminal") return yield* Effect.die("missing terminal"); + assert.isFalse( + projected.some( + (item) => + item.type === "message.updated" && + item.message.role === "assistant" && + item.message.runId === turn(3).runId && + item.message.text.includes("background wake"), + ), + ); + const wake = makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now, + ordinal: 4, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }); + yield* runtime.startTurn(wake); + assert.equal((yield* terminal())[0]?.status, "completed"); + assert.isTrue( + projected.some( + (item) => + item.type === "message.updated" && + item.message.role === "assistant" && + item.message.runId === wake.runId && + item.message.text.includes("background wake"), + ), + JSON.stringify(projected.filter((item) => item.type === "message.updated")), + ); + yield* runtime.interruptTurn({ + providerThread, + providerTurnId: cancelled.providerTurnId, + requestRuntimeRestart: true, + }); + assert.deepEqual(yield* nativeRuntime!.request("_test/cancellations", {}), { + cancellations: 1, + }); + yield* nativeRuntime!.request("_test/release", { index: 1 }); + assert.equal(prompts, 2, "continuation must not send a prompt"); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.effect.each([ + { close: true, settled: false, returned: false }, + { close: false, settled: false, returned: false }, + { close: true, settled: true, returned: false }, + { close: false, settled: true, returned: false }, + { close: false, settled: true, returned: true }, + ])( + "stops Copilot work with close=$close and settled=$settled and returned=$returned", + ({ close, settled, returned }) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const dispatched = yield* Deferred.make(); + const beforeDispatch = yield* Deferred.make(); + const allowDispatch = yield* Deferred.make(); + let sentText = ""; + const response = yield* Deferred.make<{ readonly stopReason: "cancelled" | "end_turn" }>(); + const stopped: string[] = []; + let sessionUpdate: + | Parameters[0] + | undefined; + let runtimeCount = 0; + let event: ((value: unknown) => Effect.Effect) | undefined; + const instanceId = ProviderInstanceId.make("copilot-stop-lifecycle"); + const adapter = yield* makeAcpRegistryAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + settings: COPILOT_SETTINGS, + environment: {}, + continuationRequests: { offer: () => Effect.void }, + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => { + runtimeCount += 1; + return { + ...runtime, + start: () => + runtime.start().pipe( + Effect.map((setup) => ({ + ...setup, + initializeResult: { + ...setup.initializeResult, + protocolVersion: 1, + agentCapabilities: { + ...setup.initializeResult.agentCapabilities, + sessionCapabilities: close ? { close: {} } : {}, + }, + }, + })), + ), + handleSessionUpdate: (handler) => + Effect.sync(() => { + sessionUpdate = handler; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))), + handleExtNotification: (method, schema, handler) => + Effect.sync(() => { + if (method === "github.com/copilot/sessionEvent") + event = handler as typeof event; + }).pipe(Effect.andThen(runtime.handleExtNotification(method, schema, handler))), + prompt: (payload, options) => + Effect.gen(function* () { + const text = payload.prompt.find((block) => block.type === "text"); + if (text?.type !== "text") return yield* Effect.die("missing prompt text"); + sentText = text.text; + if (runtimeCount === 1) { + yield* event!({ + sessionId: "mock-session-1", + type: "user.message", + data: { content: sentText }, + }); + yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + yield* Deferred.succeed(beforeDispatch, undefined); + yield* Deferred.await(allowDispatch); + } + if (options?.dispatched) yield* Deferred.succeed(options.dispatched, undefined); + yield* Deferred.succeed(dispatched, undefined); + return yield* Deferred.await(response); + }), + closeSession: () => + close + ? event!({ + sessionId: "mock-session-1", + type: "assistant.idle", + data: { aborted: true }, + }).pipe( + Effect.andThen(Effect.sync(() => void stopped.push("close"))), + Effect.as({}), + ) + : Effect.fail( + new EffectAcpErrors.AcpRequestError({ + code: -32601, + errorMessage: "unsupported", + }), + ), + cancel: Effect.sync(() => void stopped.push("cancel")).pipe( + Effect.andThen(Deferred.succeed(response, { stopReason: "cancelled" as const })), + Effect.asVoid, + ), + }; + }, + }), + }).pipe( + Effect.provide( + Layer.mock(AcpRegistryCatalog)({ resolve: () => Effect.die("runtime is injected") }), + ), + ); + const threadId = ThreadId.make("copilot-stop-lifecycle-thread"); + const modelSelection = { instanceId, model: "default" } as const; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("copilot-stop-lifecycle-session"), + modelSelection, + runtimePolicy, + }); + const events = yield* Queue.unbounded(); + yield* runtime.events.pipe( + Stream.runForEach((value) => Queue.offer(events, value)), + Effect.forkScoped, + ); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + }), + ); + yield* Deferred.await(beforeDispatch); + const before = yield* runtime.readThreadSnapshot({ providerThread }); + assert.equal( + before.providerTurns.at(-1)?.status, + "running", + "an echo before dispatch must not accept the prompt", + ); + yield* Deferred.succeed(allowDispatch, undefined); + yield* Deferred.await(dispatched); + yield* event!({ + sessionId: "mock-session-1", + type: "user.message", + data: { content: sentText }, + }); + yield* event!({ sessionId: "mock-session-1", type: "assistant.turn_start" }); + const take = (type: "provider_turn.updated" | "turn.terminal") => + Stream.fromQueue(events).pipe( + Stream.filter((value) => value.type === type), + Stream.take(1), + Stream.runCollect, + ); + const running = (yield* take("provider_turn.updated"))[0]; + if (running?.type !== "provider_turn.updated") + return yield* Effect.die("missing running turn"); + if (returned) { + yield* sessionUpdate!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "tool_call", + toolCallId: "stop-bg", + kind: "other", + status: "completed", + title: "Background agent", + rawInput: { agent_type: "explore", prompt: "Do work", mode: "background" }, + }, + }); + yield* Deferred.succeed(response, { stopReason: "end_turn" as const }); + } + if (settled) { + if (!returned) yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + const terminal = (yield* take("turn.terminal"))[0]; + assert.equal(terminal?.type === "turn.terminal" ? terminal.status : null, "completed"); + } + if (returned) yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + yield* runtime.interruptTurn({ + providerThread, + providerTurnId: running.providerTurn.id, + requestRuntimeRestart: true, + }); + assert.deepEqual( + stopped, + [close ? "close" : "cancel"], + "Stop must cancel background work after its prompt RPC returns", + ); + if (returned) assert.isFalse(yield* runtime.hasPendingBackgroundWork!); + if (!settled) { + const terminal = (yield* take("turn.terminal"))[0]; + assert.equal(terminal?.type === "turn.terminal" ? terminal.status : null, "interrupted"); + } + yield* Deferred.succeed(response, { stopReason: "cancelled" as const }); + yield* runtime.startTurn( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + ordinal: 2, + }), + ); + assert.equal(runtimeCount, 2); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.effect.each([ + { + name: "wrapped text", + imageOnly: false, + matches: true, + stateOnlyWake: false, + queuedWake: false, + stillRunning: false, + steered: false, + }, + { + name: "image-only input", + imageOnly: true, + matches: true, + stateOnlyWake: false, + queuedWake: false, + stillRunning: false, + steered: false, + }, + { + name: "an unmatched message", + imageOnly: false, + matches: false, + stateOnlyWake: false, + queuedWake: false, + stillRunning: false, + steered: false, + }, + { + name: "a state-only pre-acceptance wake", + imageOnly: false, + matches: true, + stateOnlyWake: true, + queuedWake: false, + stillRunning: false, + steered: false, + }, + { + name: "an unmatched echo overtaking a queued wake", + imageOnly: false, + matches: false, + stateOnlyWake: false, + queuedWake: true, + stillRunning: false, + steered: false, + }, + { + name: "a queued wake followed by newer running work", + imageOnly: false, + matches: true, + stateOnlyWake: false, + queuedWake: true, + stillRunning: true, + steered: false, + }, + { + name: "a queued wake with steering acceptance", + imageOnly: false, + matches: true, + stateOnlyWake: false, + queuedWake: true, + stillRunning: false, + steered: true, + }, + ])( + "handles Copilot acceptance for $name without a spurious continuation", + ({ imageOnly, matches, stateOnlyWake, queuedWake, stillRunning, steered }) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const fs = yield* FileSystem.FileSystem; + const host = yield* ProviderHost.ProviderHost; + const image = { + type: "image" as const, + id: "copilot-input-image", + name: "image.png", + mimeType: "image/png", + sizeBytes: 5, + }; + if (imageOnly) yield* fs.writeFileString(host.resolveAttachmentPath(image)!, "image"); + const dispatched = yield* Queue.unbounded(); + let sentText = ""; + let update: + | Parameters[0] + | undefined; + const response = yield* Deferred.make<{ readonly stopReason: "end_turn" }>(); + const offered: string[] = []; + let event: ((value: unknown) => Effect.Effect) | undefined; + const instanceId = ProviderInstanceId.make("copilot-review-repro"); + const adapter = yield* makeAcpRegistryAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + settings: COPILOT_SETTINGS, + environment: {}, + continuationRequests: { + offer: (request) => Effect.sync(() => void offered.push(String(request.threadId))), + }, + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => ({ + ...runtime, + start: () => + runtime.start().pipe( + Effect.map((setup) => ({ + ...setup, + initializeResult: { + ...setup.initializeResult, + protocolVersion: 1, + agentCapabilities: { promptCapabilities: { image: true } }, + }, + })), + ), + handleSessionUpdate: (handler) => + Effect.sync(() => { + update = handler; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))), + handleExtNotification: (method, schema, handler) => + Effect.sync(() => { + if (method === "github.com/copilot/sessionEvent") event = handler as typeof event; + }).pipe(Effect.andThen(runtime.handleExtNotification(method, schema, handler))), + prompt: (payload, options) => + Effect.gen(function* () { + const firstText = payload.prompt.find((block) => block.type === "text"); + assert.equal(firstText?.type, "text"); + if (firstText?.type !== "text") return yield* Effect.die("missing prompt text"); + sentText = firstText.text; + if (options?.dispatched) yield* Deferred.succeed(options.dispatched, undefined); + yield* Queue.offer(dispatched, undefined); + return yield* Deferred.await(response); + }), + }), + }), + }).pipe( + Effect.provide( + Layer.mock(AcpRegistryCatalog)({ resolve: () => Effect.die("runtime is injected") }), + ), + ); + const threadId = ThreadId.make("copilot-review-repro-thread"); + const modelSelection = { instanceId, model: "default" } as const; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("copilot-review-repro-session"), + modelSelection, + runtimePolicy, + }); + const events = yield* Queue.unbounded(); + yield* runtime.events.pipe( + Stream.runForEach((value) => Queue.offer(events, value)), + Effect.forkScoped, + ); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + const text = (value: string) => + update!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: value }, + }, + }); + if (queuedWake) { + yield* runtime.startTurn( + makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now }), + ); + yield* Queue.take(dispatched); + yield* event!({ + sessionId: "mock-session-1", + type: "user.message", + data: { content: sentText }, + }); + yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + yield* Stream.fromQueue(events).pipe( + Stream.takeUntil((value) => value.type === "turn.terminal"), + Stream.runDrain, + ); + yield* event!({ sessionId: "mock-session-1", type: "assistant.turn_start" }); + yield* text("queued wake output"); + } + const input = makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now, + ordinal: queuedWake ? 2 : 1, + }); + yield* runtime.startTurn({ + ...input, + message: { + ...input.message, + text: imageOnly ? "" : input.message.text, + attachments: imageOnly ? [image] : [], + }, + }); + yield* Queue.take(dispatched); + if (queuedWake && !steered) + yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + if (stateOnlyWake) { + yield* event!({ sessionId: "mock-session-1", type: "assistant.turn_start" }); + yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + } + yield* event!({ + sessionId: "mock-session-1", + type: "user.message", + data: { content: matches ? sentText : "provider rewrote this request" }, + }); + yield* event!({ sessionId: "mock-session-1", type: "assistant.turn_start" }); + yield* update!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "answer" }, + }, + }); + yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + assert.deepEqual( + offered, + queuedWake ? [String(threadId)] : [], + "own-turn idle must not offer another continuation", + ); + const snapshot = yield* runtime.readThreadSnapshot({ providerThread }); + assert.equal(snapshot.providerTurns.at(-1)?.status, matches ? "completed" : "running"); + yield* Deferred.succeed(response, { stopReason: "end_turn" as const }); + const observed = Array.from( + yield* Stream.fromQueue(events).pipe( + Stream.takeUntil((value) => value.type === "turn.terminal"), + Stream.runCollect, + ), + ); + assert.isTrue( + observed.some( + (value) => + value.type === "message.updated" && + value.message.role === "assistant" && + value.message.runId === input.runId && + value.message.text === "answer", + ), + ); + assert.deepEqual( + offered, + queuedWake ? [String(threadId)] : [], + "RPC fallback must preserve the queued continuation", + ); + if (queuedWake) { + const userSnapshot = yield* runtime.readThreadSnapshot({ providerThread }); + assert.isFalse( + userSnapshot.messages.some( + (message) => + message.runId === input.runId && + message.role === "assistant" && + message.text.includes("queued wake output"), + ), + "RPC fallback must not consume the queued continuation's output", + ); + if (stillRunning) { + yield* event!({ sessionId: "mock-session-1", type: "assistant.turn_start" }); + yield* text("new wake output"); + } + const continuation = makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now, + ordinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }); + yield* runtime.startTurn(continuation); + const snapshot = yield* runtime.readThreadSnapshot({ providerThread }); + assert.equal( + snapshot.providerTurns.at(-1)?.status, + stillRunning ? "running" : "completed", + "the latest buffered state must determine continuation settlement", + ); + if (stillRunning) yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + const completed = yield* runtime.readThreadSnapshot({ providerThread }); + assert.equal(completed.providerTurns.at(-1)?.status, "completed"); + assert.isTrue( + completed.messages.some( + (message) => + message.runId === continuation.runId && + message.role === "assistant" && + message.text.includes("queued wake output"), + ), + ); + assert.isFalse( + completed.messages.some( + (message) => + message.runId === continuation.runId && + message.role === "assistant" && + message.text.includes("answer"), + ), + "the user's answer must not appear in the background continuation", + ); + if (stillRunning) + assert.isTrue( + completed.messages.some( + (message) => + message.runId === continuation.runId && + message.role === "assistant" && + message.text.includes("new wake output"), + ), + ); + } + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + + it.effect.each([ + { + scenario: "a late lost-session error", + accepted: true, + wake: false, + restart: true, + stop: false, + }, + { + scenario: "a rejected prompt with a buffered wake", + accepted: false, + wake: "output", + restart: false, + stop: false, + }, + { + scenario: "a rejected prompt without a wake", + accepted: false, + wake: false, + restart: false, + stop: false, + }, + { + scenario: "a lost runtime with a buffered wake", + accepted: false, + wake: "output", + restart: true, + stop: false, + }, + { + scenario: "a state-only buffered wake", + accepted: false, + wake: "state", + restart: false, + stop: false, + }, + { + scenario: "a prompt rejected during Stop close", + accepted: false, + wake: "output", + restart: false, + stop: true, + }, + ])("handles a Copilot prompt failure with $scenario", ({ accepted, wake, restart, stop }) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const dispatched = yield* Deferred.make(); + const failed = yield* Deferred.make>(); + const response = yield* Deferred.make(); + let update: AcpAdapterV2ExtensionContext["updateForegroundWork"] | undefined; + let runtimeCount = 0; + let sessionUpdate: + | Parameters[0] + | undefined; + const offered: Array<{ + readonly threadId: ThreadId; + readonly providerThreadId: OrchestrationV2ProviderThread["id"]; + }> = []; + let sentText = ""; + const instanceId = ProviderInstanceId.make("acp-late-lost-session"); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + continuationRequests: { + offer: (request) => + Effect.sync( + () => + void offered.push({ + threadId: request.threadId, + providerThreadId: request.providerThreadId, + }), + ), + }, + testHooks: { + afterPromptFailureHandled: () => + Effect.withFiber((fiber) => Deferred.succeed(failed, fiber).pipe(Effect.asVoid)), + }, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + followForegroundEvents: true, + restartRuntimeAfterPromptError: () => restart, + registerExtensions: (context) => + Effect.sync(() => { + update = context.updateForegroundWork; + }), + makeRuntime: makeMockRuntime({ + childProcessSpawner, + mockAgentPath, + wrapRuntime: (runtime) => { + runtimeCount += 1; + return { + ...runtime, + start: () => + runtime.start().pipe( + Effect.map((setup) => ({ + ...setup, + initializeResult: { + ...setup.initializeResult, + protocolVersion: 1, + agentCapabilities: { + ...setup.initializeResult.agentCapabilities, + sessionCapabilities: { close: {} }, + }, + }, + })), + ), + closeSession: () => + stop + ? Deferred.fail( + response, + new EffectAcpErrors.AcpRequestError({ + code: -32800, + errorMessage: "cancelled before insertion", + }), + ).pipe(Effect.andThen(Deferred.await(failed)), Effect.as({})) + : runtime.closeSession(), + handleSessionUpdate: (handler) => + Effect.sync(() => { + sessionUpdate = handler; + }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))), + prompt: (payload, options) => + Effect.gen(function* () { + const text = payload.prompt.find((block) => block.type === "text"); + if (text?.type !== "text") return yield* Effect.die("missing prompt text"); + sentText = text.text; + if (options?.dispatched) yield* Deferred.succeed(options.dispatched, undefined); + yield* Deferred.succeed(dispatched, undefined); + return yield* Deferred.await(response); + }), + }; + }, + }), + }, + }); + const threadId = ThreadId.make("acp-late-lost-session-thread"); + const modelSelection = { instanceId, model: "default" } as const; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("acp-late-lost-session-provider"), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + const now = yield* DateTime.now; + yield* runtime.startTurn( + makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now }), + ); + yield* Deferred.await(dispatched); + if (accepted) { + yield* update!({ sessionId: "mock-session-1", state: "accepted", content: sentText }); + yield* update!({ sessionId: "mock-session-1", state: "idle" }); + } + if (wake) { + yield* update!({ sessionId: "mock-session-1", state: "running" }); + if (wake === "output") + yield* sessionUpdate!({ + sessionId: "mock-session-1", + update: { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "buffered wake output" }, + }, + }); + yield* update!({ sessionId: "mock-session-1", state: "idle" }); + } + assert.deepEqual(offered, [], "a wake must remain buffered before acceptance"); + if (stop) { + const snapshot = yield* runtime.readThreadSnapshot({ providerThread }); + const running = snapshot.providerTurns.at(-1); + if (running === undefined) return yield* Effect.die("missing running turn"); + yield* runtime.interruptTurn({ + providerThread, + providerTurnId: running.id, + requestRuntimeRestart: true, + }); + } else { + yield* Deferred.fail( + response, + new EffectAcpErrors.AcpRequestError({ code: -32603, errorMessage: "lost session" }), + ); + } + yield* Fiber.await(yield* Deferred.await(failed)); + const shouldContinue = wake === "output" && !restart && !stop; + assert.deepEqual( + offered, + shouldContinue ? [{ threadId, providerThreadId: providerThread.id }] : [], + "only a surviving runtime's buffered wake can continue after rejection", + ); + assert.equal( + yield* runtime.hasPendingBackgroundWork!, + shouldContinue, + "a rejected prompt must retain only a continuation-owned wake", + ); + if (!restart && !stop) { + const failedSnapshot = yield* runtime.readThreadSnapshot({ providerThread }); + assert.equal(failedSnapshot.providerTurns.at(-1)?.status, "failed"); + } + const next = makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now, + ordinal: 2, + ...(shouldContinue + ? { messageCreatedBy: "agent" as const, messageCreationSource: "provider" as const } + : {}), + }); + yield* runtime.startTurn(next); + assert.equal(runtimeCount, restart || stop ? 2 : 1); + if (shouldContinue) { + const snapshot = yield* runtime.readThreadSnapshot({ providerThread }); + assert.equal(snapshot.providerTurns.at(-1)?.status, "completed"); + assert.include( + snapshot.messages + .filter((message) => message.runId === next.runId) + .map((message) => message.text), + "buffered wake output", + ); + } + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("routes interleaved Copilot subagent output into subagent threads", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -5250,6 +6292,9 @@ describe("AcpAdapterV2", () => { const backgroundMutationHandlers: Array< AcpAdapterV2ExtensionContext["applyBackgroundTaskMutation"] | undefined > = []; + const foregroundWorkHandlers: Array< + AcpAdapterV2ExtensionContext["updateForegroundWork"] | undefined + > = []; const availableCommandUpdates: Array> = []; let registeredExtensionOrdinal = 0; let runtimeOrdinalSeen = 0; @@ -5307,6 +6352,7 @@ describe("AcpAdapterV2", () => { driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, enablePostSettleContinuation: true, + followAgentWorkState: true, onAvailableCommandsUpdate: (commands) => Effect.sync(() => { availableCommandUpdates.push(commands); @@ -5315,6 +6361,7 @@ describe("AcpAdapterV2", () => { Effect.sync(() => { backgroundMutationHandlers[registeredExtensionOrdinal] = context.applyBackgroundTaskMutation; + foregroundWorkHandlers[registeredExtensionOrdinal] = context.updateForegroundWork; registeredExtensionOrdinal += 1; }), makeRuntime, @@ -5384,6 +6431,10 @@ describe("AcpAdapterV2", () => { status: "running", }); assert.isFalse(yield* hasPendingBackgroundWork); + const staleForegroundWork = foregroundWorkHandlers[1]; + assert.isDefined(staleForegroundWork); + yield* staleForegroundWork!({ sessionId: "mock-session-3", state: "running" }); + assert.isFalse(yield* hasPendingBackgroundWork); }).pipe(Effect.provide(layerTest), Effect.scoped), ); diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index f06c4f47570dd..0cba2a4c2e2cf 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -205,6 +205,86 @@ describe("AcpSessionRuntime", () => { }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))), ); + it.effect("delivers Copilot-style cancellation after an older overlapping prompt returns", () => + Effect.gen(function* () { + const firstStarted = yield* Deferred.make(); + const secondStarted = yield* Deferred.make(); + const secondItem = yield* Deferred.make(); + const completedItems: string[] = []; + const script = ` + const readline = require("node:readline"); + let first, second, count = 0; + const send = value => process.stdout.write(JSON.stringify(value) + "\\n"); + readline.createInterface({ input: process.stdin }).on("line", line => { + const frame = JSON.parse(line); + const respond = result => send({ jsonrpc: "2.0", id: frame.id, result }); + if (frame.method === "initialize") respond({ protocolVersion: 1, agentCapabilities: {} }); + else if (frame.method === "session/new") respond({ sessionId: "test-session" }); + else if (frame.method === "session/prompt") { + count += 1; + if (count === 1) first = frame.id; + else second = frame.id; + send({ jsonrpc: "2.0", method: "session/update", params: { + sessionId: "test-session", update: { sessionUpdate: "agent_thought_chunk", + content: { type: "text", text: "prompt-" + count + "-started" } } + }}); + if (count === 2) send({ jsonrpc: "2.0", method: "session/update", params: { + sessionId: "test-session", update: { sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "new prompt output" } } + }}); + } else if (frame.method === "_test/release-first") { + send({ jsonrpc: "2.0", id: first, result: { stopReason: "end_turn" } }); + respond({}); + } else if (frame.method === "session/cancel") { + send({ jsonrpc: "2.0", id: second, result: { stopReason: "cancelled" } }); + } + }); + `; + const runtime = yield* AcpSessionRuntime.make({ + cwd: process.cwd(), + clientInfo: { name: "t3-test", version: "0.0.0" }, + allowOverlappingPrompts: true, + interruptPromptOnCancel: false, + spawn: { command: process.execPath, args: ["-e", script] }, + }); + yield* runtime.getEvents().pipe( + Stream.runForEach((event) => { + if (event._tag === "EventStreamBarrier") + return Deferred.succeed(event.acknowledge, undefined).pipe(Effect.asVoid); + if (event._tag === "AssistantItemCompleted") completedItems.push(event.itemId); + if ( + event._tag === "ContentDelta" && + event.text === "new prompt output" && + event.itemId !== undefined + ) + return Deferred.succeed(secondItem, event.itemId).pipe(Effect.asVoid); + if (event._tag === "ThoughtDelta" && event.text === "prompt-1-started") + return Deferred.succeed(firstStarted, undefined).pipe(Effect.asVoid); + if (event._tag === "ThoughtDelta" && event.text === "prompt-2-started") + return Deferred.succeed(secondStarted, undefined).pipe(Effect.asVoid); + return Effect.void; + }), + Effect.forkScoped, + ); + yield* runtime.start(); + const first = yield* runtime + .prompt({ prompt: [{ type: "text", text: "first" }] }) + .pipe(Effect.forkScoped); + yield* Deferred.await(firstStarted); + const second = yield* runtime + .prompt({ prompt: [{ type: "text", text: "second" }] }) + .pipe(Effect.forkScoped); + yield* Deferred.await(secondStarted); + const itemId = yield* Deferred.await(secondItem); + yield* runtime.request("_test/release-first", {}); + expect(yield* Fiber.join(first)).toMatchObject({ stopReason: "end_turn" }); + yield* runtime.drainEvents; + expect(completedItems).not.toContain(itemId); + yield* runtime.cancel; + expect(yield* Fiber.join(second)).toMatchObject({ stopReason: "cancelled" }); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))), + ); + it.effect("reports an idle child exit and rejects later prompts", () => Effect.gen(function* () { const stopped = yield* Deferred.make(); diff --git a/packages/provider-acp-registry/src/server/adapter.ts b/packages/provider-acp-registry/src/server/adapter.ts index 1b8b4240fd00e..022bb58a0e837 100644 --- a/packages/provider-acp-registry/src/server/adapter.ts +++ b/packages/provider-acp-registry/src/server/adapter.ts @@ -250,6 +250,7 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu ...(copilotRouting === undefined ? {} : { + followForegroundEvents: true, clientCapabilitiesMeta: copilotClientCapabilitiesMeta, // Copilot CLI 1.0.105 rejects stdio MCP servers from the client. mcpOverHttp: true, diff --git a/packages/provider-acp-registry/src/server/copilotAcp.ts b/packages/provider-acp-registry/src/server/copilotAcp.ts index 5f29c930e9a91..f4b248c8017ff 100644 --- a/packages/provider-acp-registry/src/server/copilotAcp.ts +++ b/packages/provider-acp-registry/src/server/copilotAcp.ts @@ -16,6 +16,10 @@ const COPILOT_SESSION_EVENT = "github.com/copilot/sessionEvent"; export const copilotClientCapabilitiesMeta = { [COPILOT_META_KEY]: { events: [ + "user.message", + "assistant.turn_start", + "assistant.idle", + "session.idle", "subagent.started", "subagent.completed", "subagent.failed", @@ -86,6 +90,29 @@ export function makeCopilotSubagentRouting() { const data = event.data ?? {}; const task = typeof data.parentToolCallId === "string" ? data.parentToolCallId : undefined; switch (event.type) { + case "user.message": + return event.agentId === undefined && + (data.source === undefined || data.source === "user") && + data.isAutopilotContinuation !== true && + typeof data.content === "string" + ? context.updateForegroundWork({ + sessionId: event.sessionId, + state: "accepted", + content: data.content, + }) + : Effect.void; + case "assistant.turn_start": + case "assistant.idle": + case "session.idle": + return event.agentId === undefined + ? context.updateForegroundWork({ + sessionId: event.sessionId, + state: event.type === "assistant.turn_start" ? "running" : "idle", + ...(event.type === "assistant.idle" && data.aborted === true + ? { stopReason: "cancelled" } + : {}), + }) + : Effect.void; case "assistant.message_delta": case "assistant.reasoning_delta": { if (typeof data.deltaContent !== "string") return Effect.void; diff --git a/packages/provider-acp/src/server/AcpSessionRuntime.ts b/packages/provider-acp/src/server/AcpSessionRuntime.ts index 514ac91164153..7d818323109c5 100644 --- a/packages/provider-acp/src/server/AcpSessionRuntime.ts +++ b/packages/provider-acp/src/server/AcpSessionRuntime.ts @@ -114,6 +114,7 @@ export interface AcpSessionRuntimeOptions { readonly posixProcessTreeController?: AcpPosixProcessTreeController; readonly linuxCgroupController?: AcpLinuxCgroupController | null; readonly cancelBehavior?: "interrupt" | "wait-for-prompt"; + readonly allowOverlappingPrompts?: boolean; readonly cancelTimeout?: Duration.Input; readonly clientCapabilities?: EffectAcpSchema.InitializeRequest["clientCapabilities"]; readonly clientInfo: { @@ -2603,67 +2604,79 @@ export const make = Effect.fn("AcpSessionRuntime.make")(function* ( ), Effect.andThen(runLoggedRequest("logout", {}, acp.agent.logout({}))), ), - prompt: (payload, promptOptions?) => - promptSerializationSemaphore.withPermit( - Effect.acquireUseRelease( - promptDispatchSemaphore.withPermit( - Effect.gen(function* () { - const started = yield* getStartedState; - yield* closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef }); - yield* Ref.set(assistantUpdatesOpenRef, true); - const requestPayload = { - sessionId: started.sessionId, - ...payload, - } satisfies EffectAcpSchema.PromptRequest; - const completed = yield* Deferred.make(); - const fiber = yield* runLoggedRequest( - "session/prompt", - requestPayload, - acp.agent.prompt(requestPayload), - ).pipe(Effect.forkIn(runtimeScope)); - const active = { fiber, completed } satisfies AcpActivePrompt; - yield* Ref.set(activePromptRef, Option.some(active)); - if (promptOptions?.dispatched) { - yield* Deferred.succeed(promptOptions.dispatched, undefined); - } - return active; - }).pipe(notificationSemaphore.withPermit), - ), - (activePrompt) => - Fiber.join(activePrompt.fiber).pipe( - Effect.catchCauseIf( - (cause) => - options.cancelBehavior !== "wait-for-prompt" && Cause.hasInterruptsOnly(cause), - (_cause) => - Effect.succeed({ - stopReason: "cancelled", - } satisfies EffectAcpSchema.PromptResponse), - ), - Effect.tap(() => - closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef }), + prompt: (payload, promptOptions?) => { + const prompt = Effect.acquireUseRelease( + promptDispatchSemaphore.withPermit( + Effect.gen(function* () { + const started = yield* getStartedState; + yield* closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef }); + yield* Ref.set(assistantUpdatesOpenRef, true); + const requestPayload = { + sessionId: started.sessionId, + ...payload, + } satisfies EffectAcpSchema.PromptRequest; + const completed = yield* Deferred.make(); + const fiber = yield* runLoggedRequest( + "session/prompt", + requestPayload, + acp.agent.prompt(requestPayload), + ).pipe(Effect.forkIn(runtimeScope)); + const active = { fiber, completed } satisfies AcpActivePrompt; + yield* Ref.set(activePromptRef, Option.some(active)); + if (promptOptions?.dispatched) { + yield* Deferred.succeed(promptOptions.dispatched, undefined); + } + return active; + }).pipe(notificationSemaphore.withPermit), + ), + (activePrompt) => + Fiber.join(activePrompt.fiber).pipe( + Effect.catchCauseIf( + (cause) => + options.cancelBehavior !== "wait-for-prompt" && Cause.hasInterruptsOnly(cause), + (_cause) => + Effect.succeed({ + stopReason: "cancelled", + } satisfies EffectAcpSchema.PromptResponse), + ), + Effect.tap(() => + notificationSemaphore.withPermit( + Ref.get(activePromptRef).pipe( + Effect.flatMap((current) => + Option.isSome(current) && current.value === activePrompt + ? closeActiveAssistantSegment({ queue: eventQueue, assistantSegmentRef }) + : Effect.void, + ), + ), ), ), - (activePrompt, result) => - Effect.gen(function* () { - if ( - options.cancelBehavior === "wait-for-prompt" && - Exit.isFailure(result) && - Cause.hasInterrupts(result.cause) - ) { - yield* retireRuntime( - new EffectAcpErrors.AcpTransportError({ - method: "session/prompt", - detail: "The ACP prompt stopped before the agent confirmed completion.", - cause: undefined, - }), - ); - } - yield* Fiber.interrupt(activePrompt.fiber).pipe(Effect.ignore); - yield* Ref.set(activePromptRef, Option.none()); - yield* Deferred.succeed(activePrompt.completed, undefined); - }), - ), - ), + ), + (activePrompt, result) => + Effect.gen(function* () { + if ( + options.cancelBehavior === "wait-for-prompt" && + Exit.isFailure(result) && + Cause.hasInterrupts(result.cause) + ) { + yield* retireRuntime( + new EffectAcpErrors.AcpTransportError({ + method: "session/prompt", + detail: "The ACP prompt stopped before the agent confirmed completion.", + cause: undefined, + }), + ); + } + yield* Fiber.interrupt(activePrompt.fiber).pipe(Effect.ignore); + yield* Ref.update(activePromptRef, (current) => + Option.isSome(current) && current.value === activePrompt ? Option.none() : current, + ); + yield* Deferred.succeed(activePrompt.completed, undefined); + }), + ); + return options.allowOverlappingPrompts === true + ? prompt + : promptSerializationSemaphore.withPermit(prompt); + }, cancel: options.cancelBehavior === "wait-for-prompt" ? promptDispatchSemaphore.withPermit(cancel) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index c4fc2f092644a..4a24bb43a90d0 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -177,6 +177,7 @@ export interface AcpAdapterV2RuntimeInput { readonly processEnvironment?: NodeJS.ProcessEnv; readonly resumeSessionId?: string; readonly interruptPromptOnCancel?: boolean; + readonly allowOverlappingPrompts?: boolean; readonly clientCapabilities: EffectAcpSchema.InitializeRequest["clientCapabilities"]; readonly clientInfo: AcpSessionRuntime.AcpSessionRuntimeOptions["clientInfo"]; readonly requestLogger?: NonNullable; @@ -202,6 +203,12 @@ export interface AcpAdapterV2UserInputRequest { export interface AcpAdapterV2ExtensionContext { readonly runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]; + readonly updateForegroundWork: (input: { + readonly sessionId: string; + readonly state: "running" | "idle" | "accepted"; + readonly content?: string; + readonly stopReason?: "cancelled"; + }) => Effect.Effect; readonly reportProviderRetry: (input: { readonly sessionId: string; readonly failure: OrchestrationV2ProviderFailure; @@ -547,6 +554,7 @@ export interface AcpAdapterV2Flavor { readonly enablePostSettleContinuation?: boolean; /** Settle on native idle after the accepted user echo, only on negotiated ACP v2. */ readonly followAgentWorkState?: boolean; + readonly followForegroundEvents?: boolean; /** * Grok CLI currently accepts and vision-processes image blocks while still * reporting the capability as false; without this override, screenshot turns @@ -636,6 +644,7 @@ export interface AcpAdapterV2Options { /** `cancelAfterPromptStarts`: Stop is holding `session/cancel` until the prompt starts. */ readonly onCancelHeld?: () => Effect.Effect; readonly afterPromptWireSettled?: () => Effect.Effect; + readonly afterPromptFailureHandled?: () => Effect.Effect; readonly afterPromptSettledWithBackgroundWork?: () => Effect.Effect; readonly onNativeResponseLifecycle?: (event: { readonly generation?: number; @@ -1398,6 +1407,8 @@ interface ActiveAcpTurn { promptSettled: boolean; promptSettledStatus: "completed" | "interrupted" | "failed" | "cancelled" | null; workAccepted: boolean; + readonly promptAcceptanceText: string | null; + readonly continuationBufferLength: number; readonly promptWireSettled: Deferred.Deferred; /** Completed once `session/prompt` is the runtime's active prompt. */ readonly promptDispatched: Deferred.Deferred; @@ -1890,10 +1901,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }; let pendingStopCloseTurn: ActiveAcpTurn | null = null; let followsAgentWorkState = false; + let canFollowForegroundEvents = false; + let pendingForegroundPrompts = 0; let agentWork: Extract< EffectAcpSchema.SessionUpdate, { sessionUpdate: "state_update" } > | null = null; + // Replay mutates agentWork; retain the newest native foreground state separately. + let foregroundWork: Extract< + EffectAcpSchema.SessionUpdate, + { sessionUpdate: "state_update" } + > | null = null; const agentRequests = new Set(); const nativeChildren = new Map(); const extractSubagentUpdate = (tool: AcpToolCallState) => @@ -1921,11 +1939,16 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const setActiveSessionSetup = Effect.fnUntraced(function* ( setup: AcpSessionRuntime.AcpSessionRuntimeStartResult, ) { + canFollowForegroundEvents = + continuationRequests !== undefined && + flavor.followForegroundEvents === true && + setup.initializeResult.protocolVersion === 1; followsAgentWorkState = - flavor.followAgentWorkState === true && continuationRequests !== undefined && + flavor.followAgentWorkState === true && setup.initializeResult.protocolVersion === 2; agentWork = null; + foregroundWork = null; agentRequests.clear(); nativeChildren.clear(); yield* Ref.set(activeSessionSetup, setup); @@ -2365,6 +2388,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( : { processEnvironment: mcpContext.processEnvironment }), ...(resumeSessionId === undefined ? {} : { resumeSessionId }), interruptPromptOnCancel: flavor.interruptPromptOnCancel ?? false, + ...(flavor.followForegroundEvents === true ? { allowOverlappingPrompts: true } : {}), clientCapabilities: { fs: { readTextFile: flavor.clientFileSystem !== undefined, @@ -5038,6 +5062,26 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return; } if (yield* followNativeChild(notification, context)) return; + if ( + canFollowForegroundEvents && + followsAgentWorkState && + context !== null && + !context.workAccepted && + !context.finalized && + !acpIsProviderContinuationMessage(context.input.message) && + notification.sessionId === (yield* Ref.get(activeSessionId)) && + (update.sessionUpdate === "state_update" || + acpRootSessionUpdateIngestsOutput(notification)) + ) { + if ( + update.sessionUpdate === "state_update" && + update.state === "idle" && + (yield* Ref.get(wakeBuffer)).length === 0 + ) + return; + yield* Ref.update(wakeBuffer, (buffer) => [...buffer, notification]); + return; + } if ( followsAgentWorkState && !(yield* Ref.get(stoppedRunQuarantine)) && @@ -6790,6 +6834,62 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (flavor.registerExtensions !== undefined) { yield* flavor.registerExtensions({ runtime: targetRuntime, + updateForegroundWork: (notice) => + runRuntimeCallbackAtGeneration( + handlerGeneration, + Effect.gen(function* () { + if ((yield* Ref.get(activeSessionId)) !== notice.sessionId) return; + if (yield* Ref.get(stoppedRunQuarantine)) return; + if (!followsAgentWorkState && !canFollowForegroundEvents) return; + if (canFollowForegroundEvents) followsAgentWorkState = true; + const context = yield* Ref.get(activeTurn); + if (notice.state === "accepted") { + const text = context?.promptAcceptanceText; + if ( + context !== null && + text && + notice.content?.includes(text) && + (yield* Deferred.isDone(context.promptDispatched)) + ) { + context.workAccepted = true; + agentWork = { sessionUpdate: "state_update", state: "running" }; + foregroundWork = agentWork; + const continuationAlreadyRequested = yield* Ref.get(continuationRequested); + yield* Ref.update(wakeBuffer, (buffer) => + !continuationAlreadyRequested && + !buffer.some(acpRootSessionUpdateIngestsOutput) + ? [] + : buffer, + ); + if ((yield* Ref.get(wakeBuffer)).length > 0) { + yield* offerContinuationRun(notice.sessionId); + } + } + return; + } + const update = { + sessionUpdate: "state_update" as const, + state: notice.state, + ...(notice.state === "idle" && notice.stopReason !== undefined + ? { stopReason: notice.stopReason } + : {}), + }; + foregroundWork = update; + yield* handleSessionUpdate({ + sessionId: notice.sessionId, + update, + }); + }), + ).pipe( + Effect.asVoid, + Effect.mapError( + (cause) => + new EffectAcpErrors.AcpTransportError({ + detail: "Failed to project ACP foreground state", + cause, + }), + ), + ), reportProviderRetry: (notice) => runRuntimeCallbackAtGeneration(handlerGeneration, reportProviderRetry(notice)).pipe( Effect.asVoid, @@ -8207,6 +8307,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( promptSettled: false, promptSettledStatus: null, workAccepted: false, + continuationBufferLength: (yield* Ref.get(continuationRequested)) + ? (yield* Ref.get(wakeBuffer)).length + : 0, + promptAcceptanceText: + promptParts?.prompt.find((block) => block.type === "text")?.text ?? null, promptWireSettled, promptDispatched, promptSentAt: null, @@ -8308,12 +8413,23 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* runtimeCallbackPermit.withPermit( Effect.gen(function* () { const liveWork = agentWork; - yield* Effect.forEach(yield* takeWakeBuffer, handleSessionUpdate, { - discard: true, - }); - agentWork = liveWork; + const latestWork = canFollowForegroundEvents ? foregroundWork : liveWork; + const buffered = yield* takeWakeBuffer; + yield* Effect.forEach(buffered, handleSessionUpdate, { discard: true }); + const bufferedWork = canFollowForegroundEvents + ? buffered.findLast((item) => item.update.sessionUpdate === "state_update") + ?.update + : undefined; + // A later user's cancelled idle cannot cancel a wake that already ended. + agentWork = + latestWork?.state === "idle" && + bufferedWork?.sessionUpdate === "state_update" && + bufferedWork.state === "idle" + ? bufferedWork + : latestWork; context.workAccepted = true; yield* settleIdleWork(context); + agentWork = liveWork; }), ); return; @@ -8367,6 +8483,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const compactSession = isCompactionMessage(turnInput.message) ? flavor.compactSession : undefined; + if (flavor.followForegroundEvents === true) pendingForegroundPrompts += 1; yield* ( compactSession === undefined ? runtime.prompt({ prompt: promptParts!.prompt }, promptOptions) @@ -8418,7 +8535,30 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( runRuntimeCallbackAtGeneration( promptGeneration, Effect.gen(function* () { - if (context.finalized || followsAgentWorkState) return; + if ( + context.finalized || + (followsAgentWorkState && flavor.followForegroundEvents !== true) + ) + return; + if (canFollowForegroundEvents && !context.workAccepted) { + context.workAccepted = true; + const preservedLength = (yield* Ref.get(continuationRequested)) + ? context.continuationBufferLength + : 0; + const buffered = yield* Ref.modify(wakeBuffer, (current) => [ + current.slice(preservedLength), + current.slice(0, preservedLength), + ]); + for (const notification of buffered) { + if (notification.update.sessionUpdate !== "state_update") + yield* handleSessionUpdate(notification); + } + const latestState = buffered.findLast( + (notification) => notification.update.sessionUpdate === "state_update", + ); + if (latestState !== undefined) yield* handleSessionUpdate(latestState); + if (context.finalized) return; + } // ACP v2 reports a failed prompt as an idle `error`, not an RPC error. const idleError = result.stopReason === "error" && !context.interrupted; const hiddenFailure = @@ -8477,6 +8617,13 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( runRuntimeCallbackAtGeneration( promptGeneration, Effect.gen(function* () { + const failure = Cause.squash(cause); + if ( + !context.interrupted && + flavor.restartRuntimeAfterPromptError?.(failure) === true + ) { + yield* Ref.set(runtimeRestartRequired, true); + } if (context.finalized) return; const error = Cause.findError(cause); // Native close rejects prompts cancelled before insertion with -32800. @@ -8486,10 +8633,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( Result.isSuccess(error) && error.success._tag === "AcpRequestError" && error.success.code === -32800); - const failure = Cause.squash(cause); - if (!interrupted && flavor.restartRuntimeAfterPromptError?.(failure) === true) { - yield* Ref.set(runtimeRestartRequired, true); - } yield* finalizeTurn( context, interrupted ? "interrupted" : "failed", @@ -8509,8 +8652,27 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }), ), ); + if (canFollowForegroundEvents && !context.workAccepted) { + if ( + !interrupted && + !(yield* Ref.get(runtimeRestartRequired)) && + (yield* Ref.get(wakeBuffer)).some(acpRootSessionUpdateIngestsOutput) + ) { + yield* offerContinuationRun(requestedSessionId); + } else if (!(yield* Ref.get(continuationRequested))) { + yield* Ref.set(wakeBuffer, []); + } + } }), - ).pipe(Effect.asVoid), + ).pipe( + Effect.asVoid, + Effect.ensuring(options.testHooks?.afterPromptFailureHandled?.() ?? Effect.void), + ), + ), + Effect.ensuring( + Effect.sync(() => { + if (flavor.followForegroundEvents === true) pendingForegroundPrompts -= 1; + }), ), Effect.forkIn(sessionScope), ); @@ -8605,9 +8767,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( : flavor.getModelContextWindow?.(selection, session); }, }), - ...(postSettleContinuationEnabled || followsAgentWorkState + ...(postSettleContinuationEnabled || followsAgentWorkState || canFollowForegroundEvents ? { hasPendingBackgroundWork: Effect.gen(function* () { + if (pendingForegroundPrompts > 0) return true; const rootSessionId = yield* Ref.get(activeSessionId); if ( followsAgentWorkState && @@ -8857,8 +9020,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const transitionFiber = yield* Effect.gen(function* () { const transition = yield* Effect.gen(function* () { const interruptContext = yield* Ref.get(activeTurn); + const supportsClose = + (yield* Ref.get(activeSessionSetup))?.initializeResult.agentCapabilities + ?.sessionCapabilities?.close != null; if ( followsAgentWorkState && + (flavor.followForegroundEvents !== true || supportsClose) && turnInput.requestRuntimeRestart === true && (interruptContext === null || interruptContext.providerTurnId === turnInput.providerTurnId) @@ -8910,6 +9077,28 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); return undefined; } + if ( + flavor.followForegroundEvents === true && + !supportsClose && + interruptContext === null && + (yield* sessionRuntime.hasPendingBackgroundWork ?? Effect.succeed(false)) && + turnInput.requestRuntimeRestart === true + ) { + yield* runtimeCallbackPermit.withPermit( + Effect.gen(function* () { + const carryover = yield* Ref.get(carryoverSubagents); + yield* quarantineStoppedRun(); + agentWork = null; + agentRequests.clear(); + nativeChildren.clear(); + yield* cancelPendingRuntimeRequests(); + yield* terminalizeCarryoverSubagents(carryover); + yield* Ref.set(runtimeRestartRequired, true); + }), + ); + yield* runtime.cancel; + return undefined; + } const softSteerCandidate = flavor.preserveRuntimeOnSettledInterrupt === true && turnInput.requestRuntimeRestart !== true && @@ -9111,7 +9300,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (!settledSoftInterrupt && (yield* promptStillRunsAfterHold(context))) { yield* runtime.cancel; } - if (restartRuntime && flavor.restartRuntimeAfterInterrupt === true) { + if ( + restartRuntime && + (flavor.restartRuntimeAfterInterrupt === true || + flavor.followForegroundEvents === true) + ) { yield* Ref.set(runtimeRestartRequired, true); } }