From 3ecdaedfdece31348a7e48bf0762f89d6fc7adc5 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Tue, 23 Jun 2026 12:47:06 +0200 Subject: [PATCH 1/2] Handle ACP request cancellation ACP SDK v0.29.0 added support for the draft of request cancellation. --- package-lock.json | 8 +- package.json | 2 +- src/CodexAcpClient.ts | 2 +- src/CodexAcpServer.ts | 226 ++++++++++++------ src/CodexAppServerClient.ts | 7 +- src/CodexApprovalHandler.ts | 27 ++- src/CodexCommands.ts | 30 ++- src/CodexElicitationHandler.ts | 14 +- .../CodexACPAgent/CodexAcpClient.test.ts | 216 ++++++++++++++++- src/index.ts | 2 +- 10 files changed, 441 insertions(+), 93 deletions(-) diff --git a/package-lock.json b/package-lock.json index 0b291df7..2fffe2d7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,7 +9,7 @@ "version": "1.0.0", "license": "Apache-2.0", "dependencies": { - "@agentclientprotocol/sdk": "^0.28.1", + "@agentclientprotocol/sdk": "^0.29.0", "@openai/codex": "^0.142.2", "diff": "^8.0.3", "open": "^11.0.0", @@ -29,9 +29,9 @@ } }, "node_modules/@agentclientprotocol/sdk": { - "version": "0.28.1", - "resolved": "https://registry.npmjs.org/@agentclientprotocol/sdk/-/sdk-0.28.1.tgz", - "integrity": "sha512-Z2Frs6YtPhnZZ+XwFXyQkRDXY0fn8FjCalEs0W4yUhQnY4TztmNq0/RnfzWdFN3vqT3h0jTz5klzYbZHGxCDyQ==", + "version": "0.29.0", + "resolved": "https://registry.npmjs.org/@agentclientprotocol/sdk/-/sdk-0.29.0.tgz", + "integrity": "sha512-Sb8xJ/cIhqlvxeJiB0+1scBk0LAkVAwJ9WAFG9XCe3LEI5JWAyOxbqaCa8O1HmNnrk4+qHN11eaKH6zL2iHIYA==", "license": "Apache-2.0", "peerDependencies": { "zod": "^3.25.0 || ^4.0.0" diff --git a/package.json b/package.json index dbf33517..fd3db924 100644 --- a/package.json +++ b/package.json @@ -61,7 +61,7 @@ "vitest": "^4.0.10" }, "dependencies": { - "@agentclientprotocol/sdk": "^0.28.1", + "@agentclientprotocol/sdk": "^0.29.0", "@openai/codex": "^0.142.2", "diff": "^8.0.3", "open": "^11.0.0", diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index f8be1d59..042c24f1 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -309,7 +309,7 @@ export class CodexAcpClient { async runReview( sessionId: string, target: ReviewTarget, - onTurnStarted?: (turnId: string) => void, + onTurnStarted?: (turnId: string, threadId: string) => void, ): Promise { return await this.codexClient.runReview({ threadId: sessionId, diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 11c33a7e..fd0a6286 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -100,6 +100,10 @@ interface PendingTurnStart { interface ActivePrompt { completion: Promise; closeSignal: Promise; + cancelSignal: Promise; + signal: AbortSignal; + currentTurn: { threadId: string, turnId: string } | null; + requestCancel: () => void; requestClose: () => void; complete: () => void; } @@ -1116,17 +1120,33 @@ export class CodexAcpServer { const closeSignal = new Promise((resolve) => { resolveCloseSignal = resolve; }); + let resolveCancelSignal: (value: null) => void = () => {}; + const cancelSignal = new Promise((resolve) => { + resolveCancelSignal = resolve; + }); + const abortController = new AbortController(); let completed = false; let closeRequested = false; const activePrompt: ActivePrompt = { completion, closeSignal, + cancelSignal, + signal: abortController.signal, + currentTurn: null, + requestCancel: () => { + if (abortController.signal.aborted) { + return; + } + abortController.abort(); + resolveCancelSignal(null); + }, requestClose: () => { if (closeRequested) { return; } closeRequested = true; + activePrompt.requestCancel(); resolveCloseSignal(null); }, complete: () => { @@ -1145,6 +1165,48 @@ export class CodexAcpServer { return activePrompt; } + private cancelBeforeTurnStarted(activePrompt: ActivePrompt): Promise { + return activePrompt.cancelSignal.then(() => { + if (activePrompt.currentTurn === null) { + return null; + } + return new Promise(() => {}); + }); + } + + private observePromptRequestCancellation( + signal: AbortSignal | undefined, + sessionState: SessionState, + activePrompt: ActivePrompt, + ): () => void { + if (!signal) { + return () => {}; + } + + const onAbort = () => { + if (this.activePrompts.get(sessionState.sessionId) !== activePrompt) { + return; + } + logger.log("Prompt request cancelled", {sessionId: sessionState.sessionId}); + activePrompt.requestCancel(); + const turn = activePrompt.currentTurn; + if (!turn) { + return; + } + void this.interruptPromptTurn(turn, "Cancel").catch((err) => { + logger.error("Prompt request cancellation failed to interrupt turn", err); + }); + }; + + if (signal.aborted) { + onAbort(); + return () => {}; + } + + signal.addEventListener("abort", onAbort, {once: true}); + return () => signal.removeEventListener("abort", onAbort); + } + private createPendingTurnStart(): PendingTurnStart { let resolve: (turnId: string | null) => void = () => {}; const promise = new Promise((innerResolve) => { @@ -1153,26 +1215,39 @@ export class CodexAcpServer { return {promise, resolve}; } - private interruptLateStartedTurn(sessionId: string, turnId: string): void { + private async interruptPromptTurn( + turn: { threadId: string, turnId: string }, + requestName: "Cancel" | "Close", + ): Promise { this.codexAcpClient.markTurnStale({ - threadId: sessionId, - turnId, + threadId: turn.threadId, + turnId: turn.turnId, }); - void this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({ - threadId: sessionId, - turnId, - })).catch((err) => { - logger.error(`Close - late turnInterrupt failed`, err); - }).finally(() => { + try { + await this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({ + threadId: turn.threadId, + turnId: turn.turnId, + })); + logger.log(`${requestName} - turnInterrupt succeeded`, { + sessionId: turn.threadId, + currentTurnId: turn.turnId, + }); + } catch (err) { + logger.error(`${requestName} - turnInterrupt failed`, err); + } finally { this.codexAcpClient.resolveTurnInterrupted({ - threadId: sessionId, - turnId, + threadId: turn.threadId, + turnId: turn.turnId, }); - }); + } + } + + private interruptLateStartedTurn(turn: { threadId: string, turnId: string }): void { + void this.interruptPromptTurn(turn, "Close"); } - private promptIsClosedOrStale(sessionId: string, activePrompt: ActivePrompt): boolean { - return this.activePrompts.get(sessionId) !== activePrompt || this.sessionIsClosing(sessionId); + private promptShouldStop(sessionId: string, activePrompt: ActivePrompt): boolean { + return activePrompt.signal.aborted || this.activePrompts.get(sessionId) !== activePrompt || this.sessionIsClosing(sessionId); } private async interruptSessionTurn( @@ -1242,7 +1317,7 @@ export class CodexAcpServer { return turnId; } - async prompt(params: acp.PromptRequest): Promise { + async prompt(params: acp.PromptRequest, signal?: AbortSignal): Promise { logger.log("Prompt received", { sessionId: params.sessionId, prompt: params.prompt, @@ -1252,11 +1327,12 @@ export class CodexAcpServer { sessionState.lastTokenUsage = null; const activePrompt = this.trackActivePrompt(params.sessionId); let pendingTurnStart: PendingTurnStart | null = null; + const disposePromptRequestCancellation = this.observePromptRequestCancellation(signal, sessionState, activePrompt); try { const eventHandler = new CodexEventHandler(this.connection, sessionState); - const approvalHandler = new CodexApprovalHandler(this.connection, sessionState); - const elicitationHandler = new CodexElicitationHandler(this.connection, sessionState); + const approvalHandler = new CodexApprovalHandler(this.connection, sessionState, activePrompt.signal); + const elicitationHandler = new CodexElicitationHandler(this.connection, sessionState, activePrompt.signal); await this.codexAcpClient.subscribeToSessionEvents(params.sessionId, (event) => { elicitationHandler.handleNotification(event); @@ -1265,28 +1341,40 @@ export class CodexAcpServer { approvalHandler, elicitationHandler); - const commandResult = await this.availableCommands.tryHandleCommand(params.prompt, sessionState); + if (activePrompt.signal.aborted) { + return this.cancelledPromptResponse(sessionState); + } + + const commandPromise = this.availableCommands.tryHandleCommand(params.prompt, sessionState, { + onTurnStarted: (turnId, threadId) => { + const turn = {threadId, turnId}; + activePrompt.currentTurn = turn; + if (this.promptShouldStop(params.sessionId, activePrompt)) { + this.interruptLateStartedTurn(turn); + return; + } + sessionState.currentTurnId = turnId; + }, + }); + void commandPromise.catch((err) => { + if (this.activePrompts.get(params.sessionId) !== activePrompt) { + logger.error(`Command for cancelled prompt ${params.sessionId} failed after prompt returned`, err); + } + }); + const commandResult = await Promise.race([ + commandPromise, + activePrompt.closeSignal, + this.cancelBeforeTurnStarted(activePrompt), + ]); + if (commandResult === null) { + return this.cancelledPromptResponse(sessionState); + } if (commandResult.handled) { logger.log("Prompt handled by a command"); await this.codexAcpClient.waitForSessionNotifications(params.sessionId); if (commandResult.turnCompleted?.turn.status === "interrupted") { - if (!this.sessionIsClosing(params.sessionId) && this.sessions.has(params.sessionId)) { - await this.connection.notify(acp.methods.client.session.update, { - sessionId: params.sessionId, - update: { - sessionUpdate: "agent_message_chunk", - content: { - type: "text", - text: "*Conversation interrupted*" - } - } - }); - } - return { - stopReason: "cancelled", - usage: this.buildPromptUsage(sessionState.lastTokenUsage), - _meta: this.buildQuotaMeta(sessionState), - }; + await this.notifyConversationInterrupted(params.sessionId); + return this.cancelledPromptResponse(sessionState); } const error = eventHandler.getFailure(); if (error) { @@ -1301,11 +1389,7 @@ export class CodexAcpServer { } if (this.sessionIsClosing(params.sessionId)) { - return { - stopReason: "cancelled", - usage: this.buildPromptUsage(sessionState.lastTokenUsage), - _meta: this.buildQuotaMeta(sessionState), - }; + return this.cancelledPromptResponse(sessionState); } const modelId = ModelId.fromString(sessionState.currentModelId); @@ -1340,54 +1424,37 @@ export class CodexAcpServer { sessionState.cwd, sessionState.additionalDirectories, (turnId) => { - if (this.promptIsClosedOrStale(params.sessionId, activePrompt)) { - this.interruptLateStartedTurn(params.sessionId, turnId); + const turn = {threadId: params.sessionId, turnId}; + activePrompt.currentTurn = turn; + if (this.promptShouldStop(params.sessionId, activePrompt)) { + this.interruptLateStartedTurn(turn); return; } sessionState.currentTurnId = turnId; pendingTurnStart?.resolve(turnId); }, - () => this.promptIsClosedOrStale(params.sessionId, activePrompt), + () => this.promptShouldStop(params.sessionId, activePrompt), )); void sendPromptPromise.catch((err) => { if (this.activePrompts.get(params.sessionId) !== activePrompt) { - logger.error(`Prompt for closed session ${params.sessionId} failed after close`, err); + logger.error(`Prompt for cancelled session ${params.sessionId} failed after prompt returned`, err); } }); const turnCompleted = await Promise.race([ sendPromptPromise, activePrompt.closeSignal, + this.cancelBeforeTurnStarted(activePrompt), ]); if (turnCompleted === null) { - return { - stopReason: "cancelled", - usage: this.buildPromptUsage(sessionState.lastTokenUsage), - _meta: this.buildQuotaMeta(sessionState), - }; + return this.cancelledPromptResponse(sessionState); } await this.codexAcpClient.waitForSessionNotifications(params.sessionId); - // Check if turn was interrupted (cancelled) if (turnCompleted.turn.status === "interrupted") { - if (!this.sessionIsClosing(params.sessionId) && this.sessions.has(params.sessionId)) { - await this.connection.notify(acp.methods.client.session.update, { - sessionId: params.sessionId, - update: { - sessionUpdate: "agent_message_chunk", - content: { - type: "text", - text: "*Conversation interrupted*" - } - } - }); - } - return { - stopReason: "cancelled", - usage: this.buildPromptUsage(sessionState.lastTokenUsage), - _meta: this.buildQuotaMeta(sessionState), - }; + await this.notifyConversationInterrupted(params.sessionId); + return this.cancelledPromptResponse(sessionState); } const error = eventHandler.getFailure(); @@ -1406,6 +1473,7 @@ export class CodexAcpServer { throw err; } finally { logger.log("Prompt completed", {sessionId: params.sessionId}); + disposePromptRequestCancellation(); sessionState.currentTurnId = null; if (pendingTurnStart !== null && this.pendingTurnStarts.get(params.sessionId) === pendingTurnStart) { this.pendingTurnStarts.delete(params.sessionId); @@ -1415,6 +1483,30 @@ export class CodexAcpServer { } } + private cancelledPromptResponse(sessionState: SessionState): acp.PromptResponse { + return { + stopReason: "cancelled", + usage: this.buildPromptUsage(sessionState.lastTokenUsage), + _meta: this.buildQuotaMeta(sessionState), + }; + } + + private async notifyConversationInterrupted(sessionId: string): Promise { + if (this.sessionIsClosing(sessionId) || !this.sessions.has(sessionId)) { + return; + } + await this.connection.notify(acp.methods.client.session.update, { + sessionId, + update: { + sessionUpdate: "agent_message_chunk", + content: { + type: "text", + text: "*Conversation interrupted*" + } + } + }); + } + private buildQuotaMeta(sessionState: SessionState): { quota: QuotaMeta } { const lastTokenUsage = sessionState.lastTokenUsage; diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index 969e161d..6a5a0b51 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -242,7 +242,10 @@ export class CodexAppServerClient { } } - async runReview(params: ReviewStartParams, onTurnStarted?: (turnId: string) => void): Promise { + async runReview( + params: ReviewStartParams, + onTurnStarted?: (turnId: string, threadId: string) => void, + ): Promise { const capturedCompletions: Array = []; const releaseCapture = this.captureTurnCompletions(params.threadId, (event) => { capturedCompletions.push(event); @@ -250,7 +253,7 @@ export class CodexAppServerClient { try { const reviewStarted = await this.reviewStart(params); - onTurnStarted?.(reviewStarted.turn.id); + onTurnStarted?.(reviewStarted.turn.id, reviewStarted.reviewThreadId); const earlyCompletion = capturedCompletions.find(event => event.turn.id === reviewStarted.turn.id); releaseCapture(); if (earlyCompletion) { diff --git a/src/CodexApprovalHandler.ts b/src/CodexApprovalHandler.ts index 868bbe73..dd73b8ae 100644 --- a/src/CodexApprovalHandler.ts +++ b/src/CodexApprovalHandler.ts @@ -46,13 +46,16 @@ function permissionOption( export class CodexApprovalHandler implements ApprovalHandler { private readonly connection: AcpClientConnection; private readonly sessionState: SessionState; + private readonly cancellationSignal: AbortSignal | undefined; constructor( connection: AcpClientConnection, - sessionState: SessionState + sessionState: SessionState, + cancellationSignal?: AbortSignal, ) { this.connection = connection; this.sessionState = sessionState; + this.cancellationSignal = cancellationSignal; } async handleCommandExecution( @@ -61,7 +64,11 @@ export class CodexApprovalHandler implements ApprovalHandler { try { const sessionId = this.sessionState.sessionId; const acpRequest = this.buildCommandPermissionRequest(sessionId, params); - const response = await this.connection.request(acp.methods.client.session.requestPermission, acpRequest); + const response = await this.connection.request( + acp.methods.client.session.requestPermission, + acpRequest, + this.requestOptions(), + ); return this.convertCommandResponse(params, response); } catch (error) { logger.error("Error requesting command execution permission", error); @@ -75,7 +82,11 @@ export class CodexApprovalHandler implements ApprovalHandler { try { const sessionId = this.sessionState.sessionId; const acpRequest = this.buildFileChangePermissionRequest(sessionId, params); - const response = await this.connection.request(acp.methods.client.session.requestPermission, acpRequest); + const response = await this.connection.request( + acp.methods.client.session.requestPermission, + acpRequest, + this.requestOptions(), + ); return this.convertFileChangeResponse(params, response); } catch (error) { logger.error("Error requesting file change permission", error); @@ -89,7 +100,11 @@ export class CodexApprovalHandler implements ApprovalHandler { try { const sessionId = this.sessionState.sessionId; const acpRequest = this.buildPermissionsRequest(sessionId, params); - const response = await this.connection.request(acp.methods.client.session.requestPermission, acpRequest); + const response = await this.connection.request( + acp.methods.client.session.requestPermission, + acpRequest, + this.requestOptions(), + ); return this.convertPermissionsResponse(params, response); } catch (error) { logger.error("Error requesting permissions", error); @@ -97,6 +112,10 @@ export class CodexApprovalHandler implements ApprovalHandler { } } + private requestOptions(): acp.SendRequestOptions | undefined { + return this.cancellationSignal ? {cancellationSignal: this.cancellationSignal} : undefined; + } + private buildCommandPermissionRequest( sessionId: string, params: CommandExecutionRequestApprovalParams diff --git a/src/CodexCommands.ts b/src/CodexCommands.ts index e4c6f8c8..e483762b 100644 --- a/src/CodexCommands.ts +++ b/src/CodexCommands.ts @@ -17,6 +17,10 @@ export type CommandHandleResult = | { handled: false } | { handled: true, turnCompleted?: TurnCompletedNotification }; +export type CommandHandleOptions = { + onTurnStarted?: (turnId: string, threadId: string) => void; +}; + export class CodexCommands { private readonly connection: AcpClientConnection; private readonly codexAcpClient: CodexAcpClient; @@ -139,7 +143,11 @@ export class CodexCommands { }; } - async tryHandleCommand(prompt: acp.ContentBlock[], sessionState: SessionState): Promise { + async tryHandleCommand( + prompt: acp.ContentBlock[], + sessionState: SessionState, + options: CommandHandleOptions = {}, + ): Promise { const command = this.parseCommand(prompt); if (command === null) return { handled: false }; const commandName = command.name; @@ -153,7 +161,7 @@ export class CodexCommands { } case "review": { const target = this.buildReviewTarget(command.rest); - const turnCompleted = await this.runReviewCommand(sessionState, target); + const turnCompleted = await this.runReviewCommand(sessionState, target, options); return { handled: true, turnCompleted }; } case "review-branch": { @@ -164,7 +172,7 @@ export class CodexCommands { const turnCompleted = await this.runReviewCommand(sessionState, { type: "baseBranch", branch: command.rest, - }); + }, options); return { handled: true, turnCompleted }; } case "review-commit": { @@ -176,7 +184,7 @@ export class CodexCommands { type: "commit", sha: command.rest, title: null, - }); + }, options); return { handled: true, turnCompleted }; } case "status": { @@ -241,12 +249,20 @@ export class CodexCommands { } } - private async runReviewCommand(sessionState: SessionState, target: ReviewTarget): Promise { + private async runReviewCommand( + sessionState: SessionState, + target: ReviewTarget, + options: CommandHandleOptions, + ): Promise { return await this.runWithProcessCheck(() => this.codexAcpClient.runReview( sessionState.sessionId, target, - (turnId) => { - sessionState.currentTurnId = turnId; + (turnId, threadId) => { + if (options.onTurnStarted) { + options.onTurnStarted(turnId, threadId); + } else { + sessionState.currentTurnId = turnId; + } }, )); } diff --git a/src/CodexElicitationHandler.ts b/src/CodexElicitationHandler.ts index 21515e56..2a15df04 100644 --- a/src/CodexElicitationHandler.ts +++ b/src/CodexElicitationHandler.ts @@ -69,6 +69,7 @@ function buildToolApprovalOptions(persistOptions: Set): acp.Permis export class CodexElicitationHandler implements ElicitationHandler { private readonly connection: AcpClientConnection; private readonly sessionState: SessionState; + private readonly cancellationSignal: AbortSignal | undefined; // In Rust, the MCP elicitation handler receives ElicitationRequestEvent directly from the MCP // protocol layer, where id is set to "mcp_tool_call_approval_" — the call ID is extracted // by stripping that prefix. @@ -86,9 +87,10 @@ export class CodexElicitationHandler implements ElicitationHandler { // (threadId, serverName). private readonly pendingMcpApprovals = new Map(); - constructor(connection: AcpClientConnection, sessionState: SessionState) { + constructor(connection: AcpClientConnection, sessionState: SessionState, cancellationSignal?: AbortSignal) { this.connection = connection; this.sessionState = sessionState; + this.cancellationSignal = cancellationSignal; } handleNotification(notification: ServerNotification): void { @@ -112,7 +114,11 @@ export class CodexElicitationHandler implements ElicitationHandler { ): Promise { try { const { request, correlatedCallId } = this.buildPermissionRequest(params); - const response = await this.connection.request(acp.methods.client.session.requestPermission, request); + const response = await this.connection.request( + acp.methods.client.session.requestPermission, + request, + this.requestOptions(), + ); if (correlatedCallId !== undefined && response.outcome.outcome !== "cancelled") { const optionId = response.outcome.optionId; if (optionId !== McpApprovalOptionId.Decline) { @@ -129,6 +135,10 @@ export class CodexElicitationHandler implements ElicitationHandler { } } + private requestOptions(): acp.SendRequestOptions | undefined { + return this.cancellationSignal ? {cancellationSignal: this.cancellationSignal} : undefined; + } + private buildPermissionRequest( params: McpServerElicitationRequestParams ): { request: acp.RequestPermissionRequest; correlatedCallId: string | undefined } { diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 70582711..3e070631 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -693,7 +693,7 @@ describe('ACP server test', { timeout: 40_000 }, () => { return onServerNotification; } - function createTurn(id: string, status: "inProgress" | "completed") { + function createTurn(id: string, status: "inProgress" | "completed" | "interrupted") { return { id, items: [], @@ -720,6 +720,14 @@ describe('ACP server test', { timeout: 40_000 }, () => { await new Promise((resolve) => setTimeout(resolve, 0)); } + function deferred(): {promise: Promise, resolve: (value: T) => void} { + let resolve: (value: T) => void = () => {}; + const promise = new Promise((innerResolve) => { + resolve = innerResolve; + }); + return {promise, resolve}; + } + it('should map events from dump', async () => { fixture.getCodexAppServerClient().onServerNotification = loadNotifications(); @@ -924,6 +932,138 @@ describe('ACP server test', { timeout: 40_000 }, () => { })).resolves.toMatchObject({stopReason: "end_turn"}); }); + it('cancels an active prompt when the ACP prompt request is cancelled', async () => { + const { mockFixture, sessionState } = setupPromptFixture(); + const turnCompleted = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValue(turnCompleted.promise); + const turnInterruptSpy = vi.spyOn(mockFixture.getCodexAcpClient(), "turnInterrupt") + .mockImplementation(async ({threadId, turnId}) => { + turnCompleted.resolve({ + threadId, + turn: createTurn(turnId, "interrupted"), + }); + }); + const controller = new AbortController(); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{ type: "text", text: "long running prompt" }], + }, controller.signal); + + await vi.waitFor(() => { + expect(sessionState.currentTurnId).toBe("turn-id"); + }); + + controller.abort(); + + await vi.waitFor(() => { + expect(turnInterruptSpy).toHaveBeenCalledWith({ + threadId: "session-id", + turnId: "turn-id", + }); + }); + await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"}); + }); + + it('returns success when a cancelled ACP prompt request completes before interruption wins', async () => { + const { mockFixture, sessionState } = setupPromptFixture(); + const turnCompleted = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValue(turnCompleted.promise); + const turnInterruptSpy = vi.spyOn(mockFixture.getCodexAcpClient(), "turnInterrupt") + .mockResolvedValue(); + const controller = new AbortController(); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{ type: "text", text: "long running prompt" }], + }, controller.signal); + + await vi.waitFor(() => { + expect(sessionState.currentTurnId).toBe("turn-id"); + }); + + controller.abort(); + await vi.waitFor(() => { + expect(turnInterruptSpy).toHaveBeenCalledWith({ + threadId: "session-id", + turnId: "turn-id", + }); + }); + + turnCompleted.resolve({ + threadId: "session-id", + turn: createTurn("turn-id", "completed"), + }); + await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"}); + }); + + it('interrupts a late-started turn after the ACP prompt request is cancelled', async () => { + const { mockFixture } = setupPromptFixture(); + const turnStart = deferred<{turn: ReturnType}>(); + const turnStartCalled = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart") + .mockImplementation(async () => { + turnStartCalled.resolve(); + return await turnStart.promise; + }); + const turnCompleted = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValue(turnCompleted.promise); + const turnInterruptSpy = vi.spyOn(mockFixture.getCodexAcpClient(), "turnInterrupt") + .mockImplementation(async ({threadId, turnId}) => { + turnCompleted.resolve({ + threadId, + turn: createTurn(turnId, "interrupted"), + }); + }); + const controller = new AbortController(); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{ type: "text", text: "long running prompt" }], + }, controller.signal); + + await turnStartCalled.promise; + controller.abort(); + + await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"}); + expect(turnInterruptSpy).not.toHaveBeenCalled(); + + turnStart.resolve({turn: createTurn("late-turn-id", "inProgress")}); + await vi.waitFor(() => { + expect(turnInterruptSpy).toHaveBeenCalledWith({ + threadId: "session-id", + turnId: "late-turn-id", + }); + }); + }); + + it('returns cancelled when the ACP prompt request is cancelled during startup work', async () => { + const { mockFixture, turnStartSpy } = setupPromptFixture(); + const skillsRefresh = deferred<{data: []}>(); + const listSkillsSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "listSkills") + .mockReturnValue(skillsRefresh.promise); + const controller = new AbortController(); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{ type: "text", text: "long running prompt" }], + }, controller.signal); + + await vi.waitFor(() => { + expect(listSkillsSpy).toHaveBeenCalled(); + }); + + controller.abort(); + await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"}); + + skillsRefresh.resolve({data: []}); + await flushAsyncWork(); + expect(turnStartSpy).not.toHaveBeenCalled(); + }); + it('should send attachments as prompt items', async () => { const mockFixture = createCodexMockTestFixture(); const codexAcpAgent = mockFixture.getCodexAcpAgent(); @@ -1124,6 +1264,46 @@ describe('ACP server test', { timeout: 40_000 }, () => { expect(promptResolved).toBe(true); }); + it('interrupts a late-started review slash command after the ACP prompt request is cancelled', async () => { + const { mockFixture } = setupPromptFixture(); + const reviewStart = deferred(); + const reviewStartSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "reviewStart") + .mockReturnValue(reviewStart.promise); + const reviewCompleted = deferred(); + const awaitTurnCompletedSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValue(reviewCompleted.promise); + const turnInterruptSpy = vi.spyOn(mockFixture.getCodexAcpClient(), "turnInterrupt") + .mockImplementation(async ({threadId, turnId}) => { + reviewCompleted.resolve({ + threadId, + turn: createTurn(turnId, "interrupted"), + }); + }); + const controller = new AbortController(); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{ type: "text", text: "/review" }], + }, controller.signal); + + await vi.waitFor(() => { + expect(reviewStartSpy).toHaveBeenCalled(); + }); + + controller.abort(); + await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"}); + + reviewStart.resolve(createReviewStartResponse("review-thread-id", "review-turn-id")); + + await vi.waitFor(() => { + expect(turnInterruptSpy).toHaveBeenCalledWith({ + threadId: "review-thread-id", + turnId: "review-turn-id", + }); + }); + expect(awaitTurnCompletedSpy).toHaveBeenCalledWith("review-thread-id", "review-turn-id"); + }); + it('returns cancelled when review slash command is interrupted', async () => { const { mockFixture } = setupPromptFixture(); vi.spyOn(mockFixture.getCodexAppServerClient(), "reviewStart") @@ -1172,6 +1352,31 @@ describe('ACP server test', { timeout: 40_000 }, () => { expect(mockFixture.getAcpConnectionDump([])).toContain("Context compacted"); }); + it('returns cancelled promptly when non-interruptible slash command startup is cancelled', async () => { + const { mockFixture } = setupPromptFixture(); + const compactStartSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "threadCompactStart") + .mockResolvedValue({}); + const controller = new AbortController(); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{ type: "text", text: "/compact" }], + }, controller.signal); + + await vi.waitFor(() => { + expect(compactStartSpy).toHaveBeenCalledWith({ threadId: "session-id" }); + }); + + controller.abort(); + await expect(promptPromise).resolves.toMatchObject({stopReason: "cancelled"}); + + mockFixture.sendServerNotification({ + method: "thread/compacted", + params: { threadId: "session-id", turnId: "compact-turn-id" }, + }); + await flushAsyncWork(); + }); + it('reports missing review slash command input', async () => { const { mockFixture } = setupPromptFixture(); const reviewStartSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "reviewStart") @@ -1375,11 +1580,14 @@ describe('ACP server test', { timeout: 40_000 }, () => { return { mockFixture, sessionState, turnStartSpy }; } - function createReviewStartResponse(): ReviewStartResponse { + function createReviewStartResponse( + reviewThreadId: string = "session-id", + turnId: string = "review-turn-id", + ): ReviewStartResponse { return { - reviewThreadId: "session-id", + reviewThreadId, turn: { - id: "review-turn-id", + id: turnId, items: [], itemsView: "notLoaded", status: "inProgress", diff --git a/src/index.ts b/src/index.ts index 64a5caa1..5f8a83dd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -124,7 +124,7 @@ function startAcpServer() { .onRequest(acp.methods.agent.session.setConfigOption, (ctx) => getAgent().setSessionConfigOption(ctx.params)) .onRequest(acp.methods.agent.authenticate, (ctx) => getAgent().authenticate(ctx.params)) .onRequest(acp.methods.agent.logout, (ctx) => getAgent().logout(ctx.params)) - .onRequest(acp.methods.agent.session.prompt, (ctx) => getAgent().prompt(ctx.params)) + .onRequest(acp.methods.agent.session.prompt, (ctx) => getAgent().prompt(ctx.params, ctx.signal)) .onNotification(acp.methods.agent.session.cancel, (ctx) => getAgent().cancel(ctx.params)) .onRequest("authentication/status", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/status", ctx.params)) .onRequest("authentication/logout", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/logout", ctx.params)) From 03cc362372cfe54ffc8290516182cdf753c658c4 Mon Sep 17 00:00:00 2001 From: Ben Brandt Date: Tue, 23 Jun 2026 15:00:29 +0200 Subject: [PATCH 2/2] Resolve turn interruption only on close cancellation --- src/CodexAcpServer.ts | 23 ++++++++++++------- .../CodexACPAgent/CodexAcpClient.test.ts | 14 ++++++++++- 2 files changed, 28 insertions(+), 9 deletions(-) diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index fd0a6286..76b80af5 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -1193,9 +1193,7 @@ export class CodexAcpServer { if (!turn) { return; } - void this.interruptPromptTurn(turn, "Cancel").catch((err) => { - logger.error("Prompt request cancellation failed to interrupt turn", err); - }); + void this.requestTurnInterrupt(turn, "Cancel"); }; if (signal.aborted) { @@ -1223,6 +1221,20 @@ export class CodexAcpServer { threadId: turn.threadId, turnId: turn.turnId, }); + try { + await this.requestTurnInterrupt(turn, requestName); + } finally { + this.codexAcpClient.resolveTurnInterrupted({ + threadId: turn.threadId, + turnId: turn.turnId, + }); + } + } + + private async requestTurnInterrupt( + turn: { threadId: string, turnId: string }, + requestName: "Cancel" | "Close", + ): Promise { try { await this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({ threadId: turn.threadId, @@ -1234,11 +1246,6 @@ export class CodexAcpServer { }); } catch (err) { logger.error(`${requestName} - turnInterrupt failed`, err); - } finally { - this.codexAcpClient.resolveTurnInterrupted({ - threadId: turn.threadId, - turnId: turn.turnId, - }); } } diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 3e070631..9f822c97 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -971,8 +971,9 @@ describe('ACP server test', { timeout: 40_000 }, () => { const turnCompleted = deferred(); vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") .mockReturnValue(turnCompleted.promise); + const turnInterrupt = deferred(); const turnInterruptSpy = vi.spyOn(mockFixture.getCodexAcpClient(), "turnInterrupt") - .mockResolvedValue(); + .mockReturnValue(turnInterrupt.promise); const controller = new AbortController(); const promptPromise = mockFixture.getCodexAcpAgent().prompt({ @@ -992,11 +993,22 @@ describe('ACP server test', { timeout: 40_000 }, () => { }); }); + mockFixture.sendServerNotification({ + method: "item/agentMessage/delta", + params: { + threadId: "session-id", + turnId: "turn-id", + itemId: "tail-item", + delta: "tail output", + }, + }); turnCompleted.resolve({ threadId: "session-id", turn: createTurn("turn-id", "completed"), }); await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"}); + expect(mockFixture.getAcpConnectionDump([])).toContain("tail output"); + turnInterrupt.resolve(undefined); }); it('interrupts a late-started turn after the ACP prompt request is cancelled', async () => {