From 21d7c19288beb844de59cbf4b76c0196a61b9d74 Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Fri, 6 Nov 2026 22:11:29 -0600 Subject: [PATCH 1/8] fix(acp): settle Copilot foreground work while background tasks run --- .../Adapters/AcpAdapterV2.test.ts | 158 ++++++++++++++++++ .../provider/acp/AcpJsonRpcConnection.test.ts | 58 +++++++ .../src/server/adapter.ts | 1 + .../src/server/copilotAcp.ts | 11 ++ .../src/server/AcpSessionRuntime.ts | 131 ++++++++------- packages/provider-acp/src/server/adapter.ts | 51 +++++- 6 files changed, 348 insertions(+), 62 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index f63a220fb5191..ea09e9954743a 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -102,6 +102,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 +3126,154 @@ 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 = []; + 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/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; + 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* () { + 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(); + yield* runtime.events.pipe( + Stream.runForEach((item) => 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) => + nativeRuntime!.request("_test/emit", { type, ...(agentId ? { agentId } : {}) }); + 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"); + assert.isTrue(Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4))))); + yield* event("assistant.idle"); + assert.equal((yield* terminal())[0]?.status, "completed"); + 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("routes interleaved Copilot subagent output into subagent threads", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -5250,6 +5399,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 +5459,7 @@ describe("AcpAdapterV2", () => { driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, enablePostSettleContinuation: true, + followAgentWorkState: true, onAvailableCommandsUpdate: (commands) => Effect.sync(() => { availableCommandUpdates.push(commands); @@ -5315,6 +5468,7 @@ describe("AcpAdapterV2", () => { Effect.sync(() => { backgroundMutationHandlers[registeredExtensionOrdinal] = context.applyBackgroundTaskMutation; + foregroundWorkHandlers[registeredExtensionOrdinal] = context.updateForegroundWork; registeredExtensionOrdinal += 1; }), makeRuntime, @@ -5384,6 +5538,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..1faa7e887b43d 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -205,6 +205,64 @@ describe("AcpSessionRuntime", () => { }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))), ); + it.effect("keeps a newer prompt cancellable after an older overlapping prompt returns", () => + Effect.gen(function* () { + const firstStarted = yield* Deferred.make(); + const secondStarted = yield* Deferred.make(); + const script = ` + const readline = require("node:readline"); + let first, 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; + send({ jsonrpc: "2.0", method: "session/update", params: { + sessionId: "test-session", update: { sessionUpdate: "agent_thought_chunk", + content: { type: "text", text: "prompt-" + count + "-started" } } + }}); + } else if (frame.method === "_test/release-first") { + send({ jsonrpc: "2.0", id: first, result: { stopReason: "end_turn" } }); + respond({}); + } + }); + `; + const runtime = yield* AcpSessionRuntime.make({ + cwd: process.cwd(), + clientInfo: { name: "t3-test", version: "0.0.0" }, + allowOverlappingPrompts: true, + spawn: { command: process.execPath, args: ["-e", script] }, + }); + yield* runtime.getEvents().pipe( + Stream.runForEach((event) => { + 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); + yield* runtime.request("_test/release-first", {}); + expect(yield* Fiber.join(first)).toMatchObject({ stopReason: "end_turn" }); + 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..59f6e2596581b 100644 --- a/packages/provider-acp-registry/src/server/copilotAcp.ts +++ b/packages/provider-acp-registry/src/server/copilotAcp.ts @@ -16,6 +16,9 @@ const COPILOT_SESSION_EVENT = "github.com/copilot/sessionEvent"; export const copilotClientCapabilitiesMeta = { [COPILOT_META_KEY]: { events: [ + "assistant.turn_start", + "assistant.idle", + "session.idle", "subagent.started", "subagent.completed", "subagent.failed", @@ -86,6 +89,14 @@ export function makeCopilotSubagentRouting() { const data = event.data ?? {}; const task = typeof data.parentToolCallId === "string" ? data.parentToolCallId : undefined; switch (event.type) { + case "assistant.turn_start": + case "assistant.idle": + return event.agentId === undefined + ? context.updateForegroundWork({ + sessionId: event.sessionId, + state: event.type === "assistant.turn_start" ? "running" : "idle", + }) + : 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..f350c7bb9f368 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,10 @@ export interface AcpAdapterV2UserInputRequest { export interface AcpAdapterV2ExtensionContext { readonly runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]; + readonly updateForegroundWork: (input: { + readonly sessionId: string; + readonly state: "running" | "idle"; + }) => Effect.Effect; readonly reportProviderRetry: (input: { readonly sessionId: string; readonly failure: OrchestrationV2ProviderFailure; @@ -547,6 +552,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 @@ -1890,6 +1896,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }; let pendingStopCloseTurn: ActiveAcpTurn | null = null; let followsAgentWorkState = false; + let pendingForegroundPrompts = 0; let agentWork: Extract< EffectAcpSchema.SessionUpdate, { sessionUpdate: "state_update" } @@ -1922,9 +1929,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( setup: AcpSessionRuntime.AcpSessionRuntimeStartResult, ) { followsAgentWorkState = - flavor.followAgentWorkState === true && continuationRequests !== undefined && - setup.initializeResult.protocolVersion === 2; + ((flavor.followAgentWorkState === true && + setup.initializeResult.protocolVersion === 2) || + (flavor.followForegroundEvents === true && + setup.initializeResult.protocolVersion === 1)); agentWork = null; agentRequests.clear(); nativeChildren.clear(); @@ -2365,6 +2374,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, @@ -6790,6 +6800,30 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (flavor.registerExtensions !== undefined) { yield* flavor.registerExtensions({ runtime: targetRuntime, + updateForegroundWork: (notice) => + runRuntimeCallbackAtGeneration( + handlerGeneration, + Effect.gen(function* () { + if (!followsAgentWorkState) return; + if ((yield* Ref.get(activeSessionId)) !== notice.sessionId) return; + if (yield* Ref.get(stoppedRunQuarantine)) return; + const context = yield* Ref.get(activeTurn); + if (notice.state === "running" && context !== null) context.workAccepted = true; + yield* handleSessionUpdate({ + sessionId: notice.sessionId, + update: { sessionUpdate: "state_update", state: notice.state }, + }); + }), + ).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, @@ -8367,6 +8401,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 +8453,11 @@ 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; // ACP v2 reports a failed prompt as an idle `error`, not an RPC error. const idleError = result.stopReason === "error" && !context.interrupted; const hiddenFailure = @@ -8512,6 +8551,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }), ).pipe(Effect.asVoid), ), + Effect.ensuring( + Effect.sync(() => { + if (flavor.followForegroundEvents === true) pendingForegroundPrompts -= 1; + }), + ), Effect.forkIn(sessionScope), ); }, @@ -8608,6 +8652,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ...(postSettleContinuationEnabled || followsAgentWorkState ? { hasPendingBackgroundWork: Effect.gen(function* () { + if (pendingForegroundPrompts > 0) return true; const rootSessionId = yield* Ref.get(activeSessionId); if ( followsAgentWorkState && From e18d6f24cf767d41636093aa3ff55dceaf046e29 Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Sat, 7 Nov 2026 08:06:25 -0600 Subject: [PATCH 2/8] fix(acp): preserve Copilot work across foreground changes --- .../Adapters/AcpAdapterV2.test.ts | 367 +++++++++++++++++- .../provider/acp/AcpJsonRpcConnection.test.ts | 26 +- .../src/server/copilotAcp.ts | 16 + packages/provider-acp/src/server/adapter.ts | 133 ++++++- 4 files changed, 518 insertions(+), 24 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index ea09e9954743a..6bdb7d38aeb72 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -3139,6 +3139,7 @@ describe("AcpAdapterV2", () => { 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 }); @@ -3149,6 +3150,15 @@ describe("AcpAdapterV2", () => { 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({}); @@ -3214,8 +3224,11 @@ describe("AcpAdapterV2", () => { runtimePolicy, }); const events = yield* Queue.unbounded(); + const projected: ProviderAdapter.ProviderAdapterV2Event[] = []; yield* runtime.events.pipe( - Stream.runForEach((item) => Queue.offer(events, item)), + Stream.runForEach((item) => + Effect.sync(() => projected.push(item)).pipe(Effect.andThen(Queue.offer(events, item))), + ), Effect.forkScoped, ); const terminal = () => @@ -3235,8 +3248,18 @@ describe("AcpAdapterV2", () => { yield* runtime.startTurn(turn(1)); yield* Deferred.await(promptDispatched); assert.isDefined(nativeRuntime); - const event = (type: string, agentId?: string) => - nativeRuntime!.request("_test/emit", { type, ...(agentId ? { agentId } : {}) }); + 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: "test prompt", delivery: "idle" }); yield* event("assistant.turn_start"); yield* event("assistant.idle", "child-agent"); assert.isTrue(Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(2))))); @@ -3266,14 +3289,350 @@ describe("AcpAdapterV2", () => { yield* nativeRuntime!.request("_test/release", { index: 0 }); yield* Deferred.await(firstPromptReturned); yield* event("assistant.turn_start"); - assert.isTrue(Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4))))); + 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* Queue.take(continuations); + yield* event("user.message", undefined, { content: "test prompt", 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: "test prompt", + 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: "test prompt" }); + 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: "test prompt", delivery: "idle" }); + 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 }, + { close: false, settled: false }, + { close: true, settled: true }, + { close: false, settled: true }, + ])("stops Copilot work with close=$close and settled=$settled", ({ close, settled }) => + 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 response = yield* Deferred.make<{ readonly stopReason: "cancelled" }>(); + const stopped = yield* Deferred.make(); + 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: {} } : {}, + }, + }, + })), + ), + 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* () { + 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(Deferred.succeed(stopped, "close")), Effect.as({})) + : Effect.fail( + new EffectAcpErrors.AcpRequestError({ + code: -32601, + errorMessage: "unsupported", + }), + ), + cancel: Deferred.succeed(stopped, "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(dispatched); + yield* event!({ + sessionId: "mock-session-1", + type: "user.message", + data: { content: "test prompt" }, + }); + 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 (settled) { + 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"); + } + yield* runtime.interruptTurn({ + providerThread, + providerTurnId: running.providerTurn.id, + requestRuntimeRestart: true, + }); + assert.equal(yield* Deferred.await(stopped), close ? "close" : "cancel"); + 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("restarts after a late lost-session error from a settled foreground prompt", () => + 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; + const instanceId = ProviderInstanceId.make("acp-late-lost-session"); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + selfInvocation: yield* resolveSelfInvocation(), + continuationRequests: { offer: () => Effect.void }, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + followForegroundEvents: true, + restartRuntimeAfterPromptError: () => { + Deferred.doneUnsafe(failed, Effect.void); + return true; + }, + 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 }, + })), + ), + prompt: (_payload, options) => + Effect.gen(function* () { + 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); + yield* update!({ sessionId: "mock-session-1", state: "accepted", content: "test prompt" }); + yield* update!({ sessionId: "mock-session-1", state: "idle" }); + yield* Deferred.fail( + response, + new EffectAcpErrors.AcpRequestError({ code: -32603, errorMessage: "lost session" }), + ); + yield* Deferred.await(failed); + yield* update!({ sessionId: "mock-session-1", state: "idle" }); + yield* runtime.startTurn( + makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal: 2 }), + ); + assert.equal(runtimeCount, 2); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("routes interleaved Copilot subagent output into subagent threads", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts index 1faa7e887b43d..0cba2a4c2e2cf 100644 --- a/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts +++ b/apps/server/src/provider/acp/AcpJsonRpcConnection.test.ts @@ -205,13 +205,15 @@ describe("AcpSessionRuntime", () => { }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))), ); - it.effect("keeps a newer prompt cancellable after an older overlapping prompt returns", () => + 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, count = 0; + 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); @@ -221,13 +223,20 @@ describe("AcpSessionRuntime", () => { 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" } }); } }); `; @@ -235,10 +244,20 @@ describe("AcpSessionRuntime", () => { 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") @@ -256,8 +275,11 @@ describe("AcpSessionRuntime", () => { .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))), diff --git a/packages/provider-acp-registry/src/server/copilotAcp.ts b/packages/provider-acp-registry/src/server/copilotAcp.ts index 59f6e2596581b..f4b248c8017ff 100644 --- a/packages/provider-acp-registry/src/server/copilotAcp.ts +++ b/packages/provider-acp-registry/src/server/copilotAcp.ts @@ -16,6 +16,7 @@ const COPILOT_SESSION_EVENT = "github.com/copilot/sessionEvent"; export const copilotClientCapabilitiesMeta = { [COPILOT_META_KEY]: { events: [ + "user.message", "assistant.turn_start", "assistant.idle", "session.idle", @@ -89,12 +90,27 @@ 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": diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index f350c7bb9f368..467871711d607 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -205,7 +205,9 @@ export interface AcpAdapterV2ExtensionContext { readonly runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]; readonly updateForegroundWork: (input: { readonly sessionId: string; - readonly state: "running" | "idle"; + readonly state: "running" | "idle" | "accepted"; + readonly content?: string; + readonly stopReason?: "cancelled"; }) => Effect.Effect; readonly reportProviderRetry: (input: { readonly sessionId: string; @@ -1896,6 +1898,7 @@ 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, @@ -1928,12 +1931,14 @@ 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 = continuationRequests !== undefined && - ((flavor.followAgentWorkState === true && - setup.initializeResult.protocolVersion === 2) || - (flavor.followForegroundEvents === true && - setup.initializeResult.protocolVersion === 1)); + flavor.followAgentWorkState === true && + setup.initializeResult.protocolVersion === 2; agentWork = null; agentRequests.clear(); nativeChildren.clear(); @@ -5048,6 +5053,29 @@ 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]); + if (update.sessionUpdate === "state_update" && update.state === "idle") { + yield* offerContinuationRun(notification.sessionId); + } + return; + } if ( followsAgentWorkState && !(yield* Ref.get(stoppedRunQuarantine)) && @@ -6804,14 +6832,44 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( runRuntimeCallbackAtGeneration( handlerGeneration, Effect.gen(function* () { - if (!followsAgentWorkState) return; 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 === "running" && context !== null) context.workAccepted = true; + if (notice.state === "accepted") { + const text = context?.input.message.text; + if ( + context !== null && + text && + notice.content?.includes(text) && + (yield* Deferred.isDone(context.promptDispatched)) + ) { + context.workAccepted = true; + agentWork = { sessionUpdate: "state_update", state: "running" }; + const continuationAlreadyRequested = yield* Ref.get(continuationRequested); + yield* Ref.update(wakeBuffer, (buffer) => + !continuationAlreadyRequested && + buffer.every( + (item) => + item.update.sessionUpdate === "state_update" && + item.update.state === "running", + ) + ? [] + : buffer, + ); + } + return; + } yield* handleSessionUpdate({ sessionId: notice.sessionId, - update: { sessionUpdate: "state_update", state: notice.state }, + update: { + sessionUpdate: "state_update", + state: notice.state, + ...(notice.state === "idle" && notice.stopReason !== undefined + ? { stopReason: notice.stopReason } + : {}), + }, }); }), ).pipe( @@ -8342,12 +8400,20 @@ 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 buffered = yield* takeWakeBuffer; + yield* Effect.forEach(buffered, handleSessionUpdate, { discard: true }); + const bufferedIdle = canFollowForegroundEvents + ? buffered.findLast( + (item) => + item.update.sessionUpdate === "state_update" && + item.update.state === "idle", + )?.update + : undefined; + agentWork = + bufferedIdle?.sessionUpdate === "state_update" ? bufferedIdle : liveWork; context.workAccepted = true; yield* settleIdleWork(context); + agentWork = liveWork; }), ); return; @@ -8458,6 +8524,14 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( (followsAgentWorkState && flavor.followForegroundEvents !== true) ) return; + if (canFollowForegroundEvents && !context.workAccepted) { + context.workAccepted = true; + const buffered = yield* Ref.getAndSet(wakeBuffer, []); + for (const notification of buffered) { + yield* handleSessionUpdate(notification); + } + 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 = @@ -8516,6 +8590,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. @@ -8525,10 +8606,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", @@ -8649,7 +8726,7 @@ 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; @@ -8902,8 +8979,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) @@ -8955,6 +9036,18 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); return undefined; } + if ( + flavor.followForegroundEvents === true && + !supportsClose && + interruptContext === null && + pendingForegroundPrompts > 0 && + turnInput.requestRuntimeRestart === true + ) { + yield* quarantineStoppedRun(); + yield* runtime.cancel; + yield* Ref.set(runtimeRestartRequired, true); + return undefined; + } const softSteerCandidate = flavor.preserveRuntimeOnSettledInterrupt === true && turnInput.requestRuntimeRestart !== true && @@ -9156,7 +9249,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); } } From 1c84119fb210884545e15d608816c6eec33d56df Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Sat, 7 Nov 2026 08:34:49 -0600 Subject: [PATCH 3/8] fix(acp): match Copilot acceptance to the dispatched prompt --- .../Adapters/AcpAdapterV2.test.ts | 221 ++++++++++++++++-- packages/provider-acp/src/server/adapter.ts | 17 +- 2 files changed, 219 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 6bdb7d38aeb72..bce14e8d9662c 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 { @@ -3171,6 +3172,7 @@ describe("AcpAdapterV2", () => { 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({ @@ -3190,6 +3192,9 @@ describe("AcpAdapterV2", () => { ...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); @@ -3259,7 +3264,7 @@ describe("AcpAdapterV2", () => { Option.isNone(yield* Queue.poll(continuations)), "idle without a wake must not create a continuation", ); - yield* event("user.message", undefined, { content: "test prompt", delivery: "idle" }); + 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))))); @@ -3298,8 +3303,10 @@ describe("AcpAdapterV2", () => { Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), "background wake must not accept the pending user prompt", ); - yield* Queue.take(continuations); - yield* event("user.message", undefined, { content: "test prompt", source: "agent-child" }); + 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)))), @@ -3319,7 +3326,7 @@ describe("AcpAdapterV2", () => { "a different message must not accept the pending prompt", ); yield* event("user.message", undefined, { - content: "test prompt", + content: promptTexts.at(-1)!, isAutopilotContinuation: true, }); yield* event("assistant.idle"); @@ -3327,13 +3334,14 @@ describe("AcpAdapterV2", () => { Exit.isFailure(yield* Effect.exit(runtime.startTurn(turn(4)))), "autopilot messages must not accept the pending prompt", ); - yield* event("user.message", "child-agent", { content: "test 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: "test prompt", delivery: "idle" }); + yield* event("user.message", undefined, { content: promptTexts.at(-1)!, delivery: "idle" }); + yield* Queue.take(continuations); yield* event("assistant.turn_start"); yield* event("assistant.idle", undefined, { aborted: true }); const cancelled = (yield* terminal())[0]; @@ -3396,6 +3404,9 @@ describe("AcpAdapterV2", () => { 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" }>(); const stopped = yield* Deferred.make(); let runtimeCount = 0; @@ -3432,8 +3443,21 @@ describe("AcpAdapterV2", () => { Effect.sync(() => { if (method === "github.com/copilot/sessionEvent") event = handler as typeof event; }).pipe(Effect.andThen(runtime.handleExtNotification(method, schema, handler))), - prompt: (_payload, options) => + 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); @@ -3495,11 +3519,19 @@ describe("AcpAdapterV2", () => { 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: "test prompt" }, + data: { content: sentText }, }); yield* event!({ sessionId: "mock-session-1", type: "assistant.turn_start" }); const take = (type: "provider_turn.updated" | "turn.terminal") => @@ -3541,6 +3573,163 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect.each([ + { name: "wrapped text", imageOnly: false, matches: true }, + { name: "image-only input", imageOnly: true, matches: true }, + { name: "an unmatched message", imageOnly: false, matches: false }, + ])( + "handles Copilot acceptance for $name without a spurious continuation", + ({ imageOnly, matches }) => + 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* Deferred.make(); + 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* Deferred.succeed(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 input = makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + }); + yield* runtime.startTurn({ + ...input, + message: { + ...input.message, + text: imageOnly ? "" : input.message.text, + attachments: imageOnly ? [image] : [], + }, + }); + yield* Deferred.await(dispatched); + 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, [], "own-turn idle must not offer a 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, [], "RPC fallback must not leave an empty continuation"); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("restarts after a late lost-session error from a settled foreground prompt", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; @@ -3553,19 +3742,20 @@ describe("AcpAdapterV2", () => { const response = yield* Deferred.make(); let update: AcpAdapterV2ExtensionContext["updateForegroundWork"] | undefined; let runtimeCount = 0; + let sentText = ""; const instanceId = ProviderInstanceId.make("acp-late-lost-session"); const adapter = yield* makeAcpAdapterV2({ instanceId, selfInvocation: yield* resolveSelfInvocation(), continuationRequests: { offer: () => Effect.void }, + testHooks: { + afterPromptFailureHandled: () => Deferred.succeed(failed, undefined).pipe(Effect.asVoid), + }, flavor: { driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, followForegroundEvents: true, - restartRuntimeAfterPromptError: () => { - Deferred.doneUnsafe(failed, Effect.void); - return true; - }, + restartRuntimeAfterPromptError: () => true, registerExtensions: (context) => Effect.sync(() => { update = context.updateForegroundWork; @@ -3584,8 +3774,11 @@ describe("AcpAdapterV2", () => { initializeResult: { ...setup.initializeResult, protocolVersion: 1 }, })), ), - prompt: (_payload, options) => + 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); @@ -3618,7 +3811,7 @@ describe("AcpAdapterV2", () => { makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now }), ); yield* Deferred.await(dispatched); - yield* update!({ sessionId: "mock-session-1", state: "accepted", content: "test prompt" }); + yield* update!({ sessionId: "mock-session-1", state: "accepted", content: sentText }); yield* update!({ sessionId: "mock-session-1", state: "idle" }); yield* Deferred.fail( response, diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 467871711d607..bfdcd57aff4ed 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -644,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; @@ -1406,6 +1407,7 @@ interface ActiveAcpTurn { promptSettled: boolean; promptSettledStatus: "completed" | "interrupted" | "failed" | "cancelled" | null; workAccepted: boolean; + readonly promptAcceptanceText: string | null; readonly promptWireSettled: Deferred.Deferred; /** Completed once `session/prompt` is the runtime's active prompt. */ readonly promptDispatched: Deferred.Deferred; @@ -5071,9 +5073,6 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) return; yield* Ref.update(wakeBuffer, (buffer) => [...buffer, notification]); - if (update.sessionUpdate === "state_update" && update.state === "idle") { - yield* offerContinuationRun(notification.sessionId); - } return; } if ( @@ -6838,7 +6837,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (canFollowForegroundEvents) followsAgentWorkState = true; const context = yield* Ref.get(activeTurn); if (notice.state === "accepted") { - const text = context?.input.message.text; + const text = context?.promptAcceptanceText; if ( context !== null && text && @@ -6858,6 +6857,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ? [] : buffer, ); + if ((yield* Ref.get(wakeBuffer)).length > 0) { + yield* offerContinuationRun(notice.sessionId); + } } return; } @@ -8299,6 +8301,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( promptSettled: false, promptSettledStatus: null, workAccepted: false, + promptAcceptanceText: + promptParts?.prompt.find((block) => block.type === "text")?.text ?? null, promptWireSettled, promptDispatched, promptSentAt: null, @@ -8626,7 +8630,10 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ), ); }), - ).pipe(Effect.asVoid), + ).pipe( + Effect.asVoid, + Effect.ensuring(options.testHooks?.afterPromptFailureHandled?.() ?? Effect.void), + ), ), Effect.ensuring( Effect.sync(() => { From 1ea91233d3513550fc46a2c23eabdf56b131424c Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Sat, 7 Nov 2026 08:53:09 -0600 Subject: [PATCH 4/8] fix(acp): retain Copilot wake output after prompt rejection --- .../Adapters/AcpAdapterV2.test.ts | 95 +++++++++++++++++-- packages/provider-acp/src/server/adapter.ts | 9 ++ 2 files changed, 94 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index bce14e8d9662c..543583c6bc5a4 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -3341,7 +3341,10 @@ describe("AcpAdapterV2", () => { "child messages must not accept the pending prompt", ); yield* event("user.message", undefined, { content: promptTexts.at(-1)!, delivery: "idle" }); - yield* Queue.take(continuations); + 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]; @@ -3730,7 +3733,17 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); - it.effect("restarts after a late lost-session error from a settled foreground prompt", () => + it.effect.each([ + { scenario: "a late lost-session error", accepted: true, wake: false, restart: true }, + { + scenario: "a rejected prompt with a buffered wake", + accepted: false, + wake: true, + restart: false, + }, + { scenario: "a rejected prompt without a wake", accepted: false, wake: false, restart: false }, + { scenario: "a lost runtime with a buffered wake", accepted: false, wake: true, restart: true }, + ])("handles a Copilot prompt failure with $scenario", ({ accepted, wake, restart }) => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -3742,12 +3755,28 @@ describe("AcpAdapterV2", () => { 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: () => Effect.void }, + continuationRequests: { + offer: (request) => + Effect.sync( + () => + void offered.push({ + threadId: request.threadId, + providerThreadId: request.providerThreadId, + }), + ), + }, testHooks: { afterPromptFailureHandled: () => Deferred.succeed(failed, undefined).pipe(Effect.asVoid), }, @@ -3755,7 +3784,7 @@ describe("AcpAdapterV2", () => { driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, followForegroundEvents: true, - restartRuntimeAfterPromptError: () => true, + restartRuntimeAfterPromptError: () => restart, registerExtensions: (context) => Effect.sync(() => { update = context.updateForegroundWork; @@ -3774,6 +3803,10 @@ describe("AcpAdapterV2", () => { initializeResult: { ...setup.initializeResult, protocolVersion: 1 }, })), ), + 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"); @@ -3811,18 +3844,60 @@ describe("AcpAdapterV2", () => { makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now }), ); yield* Deferred.await(dispatched); - yield* update!({ sessionId: "mock-session-1", state: "accepted", content: sentText }); - yield* update!({ sessionId: "mock-session-1", state: "idle" }); + 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" }); + 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"); yield* Deferred.fail( response, new EffectAcpErrors.AcpRequestError({ code: -32603, errorMessage: "lost session" }), ); yield* Deferred.await(failed); - yield* update!({ sessionId: "mock-session-1", state: "idle" }); - yield* runtime.startTurn( - makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal: 2 }), + const shouldContinue = wake && !restart; + assert.deepEqual( + offered, + shouldContinue ? [{ threadId, providerThreadId: providerThread.id }] : [], + "only a surviving runtime's buffered wake can continue after rejection", ); - assert.equal(runtimeCount, 2); + if (!restart) { + 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 ? 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), ); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index bfdcd57aff4ed..2b3d366a03783 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -8629,6 +8629,15 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }), ), ); + if ( + canFollowForegroundEvents && + !context.workAccepted && + !interrupted && + !(yield* Ref.get(runtimeRestartRequired)) && + (yield* Ref.get(wakeBuffer)).length > 0 + ) { + yield* offerContinuationRun(requestedSessionId); + } }), ).pipe( Effect.asVoid, From 38907e5ddd91dfbf1a4dc25350688033edcd68e3 Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Sat, 7 Nov 2026 09:11:13 -0600 Subject: [PATCH 5/8] fix(acp): skip continuations for Copilot state-only wakes --- .../Adapters/AcpAdapterV2.test.ts | 120 ++++++++++++++---- packages/provider-acp/src/server/adapter.ts | 8 +- 2 files changed, 98 insertions(+), 30 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 543583c6bc5a4..3bcbf9c342437 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -3577,12 +3577,18 @@ describe("AcpAdapterV2", () => { ); it.effect.each([ - { name: "wrapped text", imageOnly: false, matches: true }, - { name: "image-only input", imageOnly: true, matches: true }, - { name: "an unmatched message", imageOnly: false, matches: false }, + { name: "wrapped text", imageOnly: false, matches: true, stateOnlyWake: false }, + { name: "image-only input", imageOnly: true, matches: true, stateOnlyWake: false }, + { name: "an unmatched message", imageOnly: false, matches: false, stateOnlyWake: false }, + { + name: "a state-only pre-acceptance wake", + imageOnly: false, + matches: true, + stateOnlyWake: true, + }, ])( "handles Copilot acceptance for $name without a spurious continuation", - ({ imageOnly, matches }) => + ({ imageOnly, matches, stateOnlyWake }) => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -3696,6 +3702,10 @@ describe("AcpAdapterV2", () => { }, }); yield* Deferred.await(dispatched); + 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", @@ -3734,16 +3744,49 @@ describe("AcpAdapterV2", () => { ); it.effect.each([ - { scenario: "a late lost-session error", accepted: true, wake: false, restart: true }, + { + 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: true, + wake: "output", restart: false, + stop: false, }, - { scenario: "a rejected prompt without a wake", accepted: false, wake: false, restart: false }, - { scenario: "a lost runtime with a buffered wake", accepted: false, wake: true, restart: true }, - ])("handles a Copilot prompt failure with $scenario", ({ accepted, wake, restart }) => + { + 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; @@ -3800,9 +3843,26 @@ describe("AcpAdapterV2", () => { runtime.start().pipe( Effect.map((setup) => ({ ...setup, - initializeResult: { ...setup.initializeResult, protocolVersion: 1 }, + 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; @@ -3850,28 +3910,40 @@ describe("AcpAdapterV2", () => { } if (wake) { yield* update!({ sessionId: "mock-session-1", state: "running" }); - yield* sessionUpdate!({ - sessionId: "mock-session-1", - update: { - sessionUpdate: "agent_message_chunk", - content: { type: "text", text: "buffered wake output" }, - }, - }); + 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"); - yield* Deferred.fail( - response, - new EffectAcpErrors.AcpRequestError({ code: -32603, errorMessage: "lost session" }), - ); + 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* Deferred.await(failed); - const shouldContinue = wake && !restart; + 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", ); - if (!restart) { + if (!restart && !stop) { const failedSnapshot = yield* runtime.readThreadSnapshot({ providerThread }); assert.equal(failedSnapshot.providerTurns.at(-1)?.status, "failed"); } @@ -3887,7 +3959,7 @@ describe("AcpAdapterV2", () => { : {}), }); yield* runtime.startTurn(next); - assert.equal(runtimeCount, restart ? 2 : 1); + assert.equal(runtimeCount, restart || stop ? 2 : 1); if (shouldContinue) { const snapshot = yield* runtime.readThreadSnapshot({ providerThread }); assert.equal(snapshot.providerTurns.at(-1)?.status, "completed"); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 2b3d366a03783..c1321be1988e4 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -6849,11 +6849,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const continuationAlreadyRequested = yield* Ref.get(continuationRequested); yield* Ref.update(wakeBuffer, (buffer) => !continuationAlreadyRequested && - buffer.every( - (item) => - item.update.sessionUpdate === "state_update" && - item.update.state === "running", - ) + !buffer.some(acpRootSessionUpdateIngestsOutput) ? [] : buffer, ); @@ -8634,7 +8630,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( !context.workAccepted && !interrupted && !(yield* Ref.get(runtimeRestartRequired)) && - (yield* Ref.get(wakeBuffer)).length > 0 + (yield* Ref.get(wakeBuffer)).some(acpRootSessionUpdateIngestsOutput) ) { yield* offerContinuationRun(requestedSessionId); } From 8ce7bbac8114b04289ef54ea02d055441822f4ac Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Sat, 7 Nov 2026 17:37:58 -0600 Subject: [PATCH 6/8] fix(acp): preserve Copilot wake ownership and stop background work --- .../Adapters/AcpAdapterV2.test.ts | 556 ++++++++++++------ packages/provider-acp/src/server/adapter.ts | 53 +- 2 files changed, 395 insertions(+), 214 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 3bcbf9c342437..b6edb24d02556 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -3395,200 +3395,273 @@ describe("AcpAdapterV2", () => { ); it.effect.each([ - { close: true, settled: false }, - { close: false, settled: false }, - { close: true, settled: true }, - { close: false, settled: true }, - ])("stops Copilot work with close=$close and settled=$settled", ({ close, settled }) => - 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" }>(); - const stopped = yield* Deferred.make(); - 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: {} } : {}, + { 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, ), - 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(Deferred.succeed(stopped, "close")), Effect.as({})) - : Effect.fail( - new EffectAcpErrors.AcpRequestError({ - code: -32601, - errorMessage: "unsupported", - }), - ), - cancel: Deferred.succeed(stopped, "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({ + }; + }, + }), + }).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, - providerThread, - instanceId, + providerSessionId: ProviderSessionId.make("copilot-stop-lifecycle-session"), + modelSelection, 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 events = yield* Queue.unbounded(); + yield* runtime.events.pipe( + Stream.runForEach((value) => Queue.offer(events, value)), + Effect.forkScoped, ); - const running = (yield* take("provider_turn.updated"))[0]; - if (running?.type !== "provider_turn.updated") - return yield* Effect.die("missing running turn"); - if (settled) { - 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"); - } - yield* runtime.interruptTurn({ - providerThread, - providerTurnId: running.providerTurn.id, - requestRuntimeRestart: true, - }); - assert.equal(yield* Deferred.await(stopped), close ? "close" : "cancel"); - 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({ + const providerThread = yield* runtime.ensureThread({ threadId, - providerThread, - instanceId, + modelSelection, runtimePolicy, - now: yield* DateTime.now, - ordinal: 2, - }), - ); - assert.equal(runtimeCount, 2); - }).pipe(Effect.provide(layerTest), Effect.scoped), + }); + 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 }, - { name: "image-only input", imageOnly: true, matches: true, stateOnlyWake: false }, - { name: "an unmatched message", imageOnly: false, matches: false, stateOnlyWake: false }, + { + name: "wrapped text", + imageOnly: false, + matches: true, + stateOnlyWake: false, + queuedWake: false, + stillRunning: false, + }, + { + name: "image-only input", + imageOnly: true, + matches: true, + stateOnlyWake: false, + queuedWake: false, + stillRunning: false, + }, + { + name: "an unmatched message", + imageOnly: false, + matches: false, + stateOnlyWake: false, + queuedWake: false, + stillRunning: false, + }, { name: "a state-only pre-acceptance wake", imageOnly: false, matches: true, stateOnlyWake: true, + queuedWake: false, + stillRunning: false, + }, + { + name: "an unmatched echo overtaking a queued wake", + imageOnly: false, + matches: false, + stateOnlyWake: false, + queuedWake: true, + stillRunning: false, + }, + { + name: "a queued wake followed by newer running work", + imageOnly: false, + matches: true, + stateOnlyWake: false, + queuedWake: true, + stillRunning: true, }, ])( "handles Copilot acceptance for $name without a spurious continuation", - ({ imageOnly, matches, stateOnlyWake }) => + ({ imageOnly, matches, stateOnlyWake, queuedWake, stillRunning }) => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -3605,7 +3678,7 @@ describe("AcpAdapterV2", () => { sizeBytes: 5, }; if (imageOnly) yield* fs.writeFileString(host.resolveAttachmentPath(image)!, "image"); - const dispatched = yield* Deferred.make(); + const dispatched = yield* Queue.unbounded(); let sentText = ""; let update: | Parameters[0] @@ -3653,7 +3726,7 @@ describe("AcpAdapterV2", () => { if (firstText?.type !== "text") return yield* Effect.die("missing prompt text"); sentText = firstText.text; if (options?.dispatched) yield* Deferred.succeed(options.dispatched, undefined); - yield* Deferred.succeed(dispatched, undefined); + yield* Queue.offer(dispatched, undefined); return yield* Deferred.await(response); }), }), @@ -3686,12 +3759,40 @@ describe("AcpAdapterV2", () => { 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: yield* DateTime.now, + now, + ordinal: queuedWake ? 2 : 1, }); yield* runtime.startTurn({ ...input, @@ -3701,7 +3802,8 @@ describe("AcpAdapterV2", () => { attachments: imageOnly ? [image] : [], }, }); - yield* Deferred.await(dispatched); + yield* Queue.take(dispatched); + if (queuedWake) 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" }); @@ -3720,7 +3822,11 @@ describe("AcpAdapterV2", () => { }, }); yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); - assert.deepEqual(offered, [], "own-turn idle must not offer a continuation"); + 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 }); @@ -3730,16 +3836,74 @@ describe("AcpAdapterV2", () => { Stream.runCollect, ), ); - assert.isTrue( - observed.some( - (value) => - value.type === "message.updated" && - value.message.role === "assistant" && - value.message.runId === input.runId && - value.message.text === "answer", - ), + if (!queuedWake) + 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", ); - assert.deepEqual(offered, [], "RPC fallback must not leave an empty 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"), + ), + ); + 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), ); @@ -3794,7 +3958,7 @@ describe("AcpAdapterV2", () => { new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), ); const dispatched = yield* Deferred.make(); - const failed = yield* Deferred.make(); + const failed = yield* Deferred.make>(); const response = yield* Deferred.make(); let update: AcpAdapterV2ExtensionContext["updateForegroundWork"] | undefined; let runtimeCount = 0; @@ -3821,7 +3985,8 @@ describe("AcpAdapterV2", () => { ), }, testHooks: { - afterPromptFailureHandled: () => Deferred.succeed(failed, undefined).pipe(Effect.asVoid), + afterPromptFailureHandled: () => + Effect.withFiber((fiber) => Deferred.succeed(failed, fiber).pipe(Effect.asVoid)), }, flavor: { driver: ACP_TEST_DRIVER, @@ -3936,13 +4101,18 @@ describe("AcpAdapterV2", () => { new EffectAcpErrors.AcpRequestError({ code: -32603, errorMessage: "lost session" }), ); } - yield* Deferred.await(failed); + 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"); diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index c1321be1988e4..026d10f3b8f96 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -8402,15 +8402,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( const liveWork = agentWork; const buffered = yield* takeWakeBuffer; yield* Effect.forEach(buffered, handleSessionUpdate, { discard: true }); - const bufferedIdle = canFollowForegroundEvents - ? buffered.findLast( - (item) => - item.update.sessionUpdate === "state_update" && - item.update.state === "idle", - )?.update + const bufferedWork = canFollowForegroundEvents + ? buffered.findLast((item) => item.update.sessionUpdate === "state_update") + ?.update : undefined; agentWork = - bufferedIdle?.sessionUpdate === "state_update" ? bufferedIdle : liveWork; + bufferedWork?.sessionUpdate === "state_update" ? bufferedWork : liveWork; context.workAccepted = true; yield* settleIdleWork(context); agentWork = liveWork; @@ -8526,9 +8523,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return; if (canFollowForegroundEvents && !context.workAccepted) { context.workAccepted = true; - const buffered = yield* Ref.getAndSet(wakeBuffer, []); - for (const notification of buffered) { - yield* handleSessionUpdate(notification); + if (!(yield* Ref.get(continuationRequested))) { + const buffered = yield* Ref.getAndSet(wakeBuffer, []); + for (const notification of buffered) { + yield* handleSessionUpdate(notification); + } } if (context.finalized) return; } @@ -8625,14 +8624,16 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( }), ), ); - if ( - canFollowForegroundEvents && - !context.workAccepted && - !interrupted && - !(yield* Ref.get(runtimeRestartRequired)) && - (yield* Ref.get(wakeBuffer)).some(acpRootSessionUpdateIngestsOutput) - ) { - yield* offerContinuationRun(requestedSessionId); + 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( @@ -9052,12 +9053,22 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( flavor.followForegroundEvents === true && !supportsClose && interruptContext === null && - pendingForegroundPrompts > 0 && + (yield* sessionRuntime.hasPendingBackgroundWork ?? Effect.succeed(false)) && turnInput.requestRuntimeRestart === true ) { - yield* quarantineStoppedRun(); + 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; - yield* Ref.set(runtimeRestartRequired, true); return undefined; } const softSteerCandidate = From 4dcea59268af69c5fb0eaa5b854df173db80f72d Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Sat, 7 Nov 2026 17:55:48 -0600 Subject: [PATCH 7/8] fix(acp): keep Copilot replies and wake state in their own turns --- .../Adapters/AcpAdapterV2.test.ts | 48 ++++++++++++----- packages/provider-acp/src/server/adapter.ts | 52 ++++++++++++++----- 2 files changed, 76 insertions(+), 24 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index b6edb24d02556..dc58382238693 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -3618,6 +3618,7 @@ describe("AcpAdapterV2", () => { stateOnlyWake: false, queuedWake: false, stillRunning: false, + steered: false, }, { name: "image-only input", @@ -3626,6 +3627,7 @@ describe("AcpAdapterV2", () => { stateOnlyWake: false, queuedWake: false, stillRunning: false, + steered: false, }, { name: "an unmatched message", @@ -3634,6 +3636,7 @@ describe("AcpAdapterV2", () => { stateOnlyWake: false, queuedWake: false, stillRunning: false, + steered: false, }, { name: "a state-only pre-acceptance wake", @@ -3642,6 +3645,7 @@ describe("AcpAdapterV2", () => { stateOnlyWake: true, queuedWake: false, stillRunning: false, + steered: false, }, { name: "an unmatched echo overtaking a queued wake", @@ -3650,6 +3654,7 @@ describe("AcpAdapterV2", () => { stateOnlyWake: false, queuedWake: true, stillRunning: false, + steered: false, }, { name: "a queued wake followed by newer running work", @@ -3658,10 +3663,20 @@ describe("AcpAdapterV2", () => { 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 }) => + ({ imageOnly, matches, stateOnlyWake, queuedWake, stillRunning, steered }) => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; const path = yield* Path.Path; @@ -3803,7 +3818,8 @@ describe("AcpAdapterV2", () => { }, }); yield* Queue.take(dispatched); - if (queuedWake) yield* event!({ sessionId: "mock-session-1", type: "assistant.idle" }); + 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" }); @@ -3836,16 +3852,15 @@ describe("AcpAdapterV2", () => { Stream.runCollect, ), ); - if (!queuedWake) - assert.isTrue( - observed.some( - (value) => - value.type === "message.updated" && - value.message.role === "assistant" && - value.message.runId === input.runId && - value.message.text === "answer", - ), - ); + 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)] : [], @@ -3894,6 +3909,15 @@ describe("AcpAdapterV2", () => { 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( diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 026d10f3b8f96..128a1280cb6bd 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -1408,6 +1408,7 @@ interface ActiveAcpTurn { 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; @@ -1906,6 +1907,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( 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) => @@ -1942,6 +1948,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( flavor.followAgentWorkState === true && setup.initializeResult.protocolVersion === 2; agentWork = null; + foregroundWork = null; agentRequests.clear(); nativeChildren.clear(); yield* Ref.set(activeSessionSetup, setup); @@ -6846,6 +6853,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ) { context.workAccepted = true; agentWork = { sessionUpdate: "state_update", state: "running" }; + foregroundWork = agentWork; const continuationAlreadyRequested = yield* Ref.get(continuationRequested); yield* Ref.update(wakeBuffer, (buffer) => !continuationAlreadyRequested && @@ -6859,15 +6867,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( } 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: { - sessionUpdate: "state_update", - state: notice.state, - ...(notice.state === "idle" && notice.stopReason !== undefined - ? { stopReason: notice.stopReason } - : {}), - }, + update, }); }), ).pipe( @@ -8297,6 +8307,9 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( promptSettled: false, promptSettledStatus: null, workAccepted: false, + continuationBufferLength: continuationWasRequested + ? (yield* Ref.get(wakeBuffer)).length + : 0, promptAcceptanceText: promptParts?.prompt.find((block) => block.type === "text")?.text ?? null, promptWireSettled, @@ -8400,14 +8413,20 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( yield* runtimeCallbackPermit.withPermit( Effect.gen(function* () { const liveWork = agentWork; + 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 = - bufferedWork?.sessionUpdate === "state_update" ? bufferedWork : liveWork; + latestWork?.state === "idle" && + bufferedWork?.sessionUpdate === "state_update" && + bufferedWork.state === "idle" + ? bufferedWork + : latestWork; context.workAccepted = true; yield* settleIdleWork(context); agentWork = liveWork; @@ -8523,12 +8542,21 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( return; if (canFollowForegroundEvents && !context.workAccepted) { context.workAccepted = true; - if (!(yield* Ref.get(continuationRequested))) { - const buffered = yield* Ref.getAndSet(wakeBuffer, []); - for (const notification of buffered) { + 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. From 7dd739db92ec9a8877a72bab9c8165522d64540a Mon Sep 17 00:00:00 2001 From: Nicolas Layne Date: Sat, 7 Nov 2026 18:05:03 -0600 Subject: [PATCH 8/8] fix(acp): refresh queued wake state before attaching a turn --- packages/provider-acp/src/server/adapter.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 128a1280cb6bd..4a24bb43a90d0 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -8307,7 +8307,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( promptSettled: false, promptSettledStatus: null, workAccepted: false, - continuationBufferLength: continuationWasRequested + continuationBufferLength: (yield* Ref.get(continuationRequested)) ? (yield* Ref.get(wakeBuffer)).length : 0, promptAcceptanceText: