Skip to content

Handle ACP request cancellation #221

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
benbrandt merged 2 commits into from
Jun 25, 2026
Merged

Handle ACP request cancellation #221

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3ecdaed
Handle ACP request cancellation
benbrandt Jun 23, 2026
03cc362
Resolve turn interruption only on close cancellation
benbrandt Jun 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions package-lock.json
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
2 changes: 1 addition & 1 deletion src/CodexAcpClient.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -309,7 +309,7 @@ export class CodexAcpClient {
async runReview(
sessionId: string,
target: ReviewTarget,
onTurnStarted?: (turnId: string) => void,
onTurnStarted?: (turnId: string, threadId: string) => void,
): Promise<TurnCompletedNotification> {
return await this.codexClient.runReview({
threadId: sessionId,
Expand Down
233 changes: 166 additions & 67 deletions src/CodexAcpServer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,10 @@ interface PendingTurnStart {
interface ActivePrompt {
completion: Promise<void>;
closeSignal: Promise<null>;
cancelSignal: Promise<null>;
signal: AbortSignal;
currentTurn: { threadId: string, turnId: string } | null;
requestCancel: () => void;
requestClose: () => void;
complete: () => void;
}
Expand Down Expand Up @@ -1116,17 +1120,33 @@ export class CodexAcpServer {
const closeSignal = new Promise<null>((resolve) => {
resolveCloseSignal = resolve;
});
let resolveCancelSignal: (value: null) => void = () => {};
const cancelSignal = new Promise<null>((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: () => {
Expand All @@ -1145,6 +1165,46 @@ export class CodexAcpServer {
return activePrompt;
}

private cancelBeforeTurnStarted(activePrompt: ActivePrompt): Promise<null> {
return activePrompt.cancelSignal.then(() => {
if (activePrompt.currentTurn === null) {
return null;
}
return new Promise<null>(() => {});
});
}

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.requestTurnInterrupt(turn, "Cancel");
};

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<string | null>((innerResolve) => {
Expand All @@ -1153,26 +1213,48 @@ 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<void> {
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.requestTurnInterrupt(turn, requestName);
} finally {
this.codexAcpClient.resolveTurnInterrupted({
threadId: sessionId,
turnId,
threadId: turn.threadId,
turnId: turn.turnId,
});
});
}
}

private async requestTurnInterrupt(
turn: { threadId: string, turnId: string },
requestName: "Cancel" | "Close",
): Promise<void> {
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);
}
}

private promptIsClosedOrStale(sessionId: string, activePrompt: ActivePrompt): boolean {
return this.activePrompts.get(sessionId) !== activePrompt || this.sessionIsClosing(sessionId);
private interruptLateStartedTurn(turn: { threadId: string, turnId: string }): void {
void this.interruptPromptTurn(turn, "Close");
}

private promptShouldStop(sessionId: string, activePrompt: ActivePrompt): boolean {
return activePrompt.signal.aborted || this.activePrompts.get(sessionId) !== activePrompt || this.sessionIsClosing(sessionId);
}

private async interruptSessionTurn(
Expand Down Expand Up @@ -1242,7 +1324,7 @@ export class CodexAcpServer {
return turnId;
}

async prompt(params: acp.PromptRequest): Promise<acp.PromptResponse> {
async prompt(params: acp.PromptRequest, signal?: AbortSignal): Promise<acp.PromptResponse> {
logger.log("Prompt received", {
sessionId: params.sessionId,
prompt: params.prompt,
Expand All @@ -1252,11 +1334,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);
Expand All @@ -1265,28 +1348,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) {
Expand All @@ -1301,11 +1396,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);
Expand Down Expand Up @@ -1340,54 +1431,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();
Expand All @@ -1406,6 +1480,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);
Expand All @@ -1415,6 +1490,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<void> {
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;

Expand Down
Loading