Skip to content

Commit 15a7a84

Browse files
xtmqxtmq
committed
Support turn cancellation
1 parent 7fec958 commit 15a7a84

4 files changed

Lines changed: 98 additions & 29 deletions

File tree

‎src/CodexAcpClient.ts‎

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import type {
99
SetDefaultModelParams,
1010
SetDefaultModelResponse
1111
} from "./app-server";
12+
import type { TurnCompletedNotification } from "./app-server/v2";
1213
import type {JsonValue} from "./app-server/serde_json/JsonValue";
1314
import type {Model} from "./app-server/v2";
1415
import {ModelId} from "./ModelId";
@@ -101,14 +102,18 @@ export class CodexAcpClient {
101102
};
102103
}
103104

104-
async sendPrompt(request: acp.PromptRequest, eventHandler: (result: ServerNotification) => void): Promise<void> {
105+
async sendPrompt(
106+
request: acp.PromptRequest,
107+
eventHandler: (result: ServerNotification) => void,
108+
onTurnStarted?: (turnId: string) => void
109+
): Promise<TurnCompletedNotification> {
105110
this.codexClient.onServerNotification(request.sessionId, eventHandler);
106111

107112
const input = request.prompt.filter(b => b.type === "text")
108113
.map(b => b.text)
109114
.join(" ");
110115

111-
await this.codexClient.turnStart({
116+
const turnStartResponse = await this.codexClient.turnStart({
112117
threadId: request.sessionId,
113118
input: [{type: "text", text: input}],
114119
approvalPolicy: null,
@@ -119,13 +124,27 @@ export class CodexAcpClient {
119124
model: null,
120125
});
121126

122-
await this.codexClient.awaitTurnCompleted();
127+
// Save turnId if callback provided
128+
if (onTurnStarted) {
129+
onTurnStarted(turnStartResponse.turn.id);
130+
}
131+
132+
// Wait for turn completion
133+
// If turnInterrupt() was called, Codex will send turn/completed event with status "interrupted"
134+
return await this.codexClient.awaitTurnCompleted();
123135
}
124136

125137
async setModel(params: SetDefaultModelParams): Promise<SetDefaultModelResponse> {
126138
return this.codexClient.setModelRequest(params);
127139
}
128140

141+
async turnInterrupt(params: { threadId: string, turnId: string }): Promise<void> {
142+
await this.codexClient.turnInterrupt({
143+
threadId: params.threadId,
144+
turnId: params.turnId
145+
});
146+
}
147+
129148
private async fetchAvailableModels(): Promise<Model[]> {
130149
const models: Model[] = [];
131150
let cursor: string | null = null;

‎src/CodexAcpServer.ts‎

Lines changed: 45 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import {ModelId} from "./ModelId";
1010

1111
export interface SessionState {
1212
sessionMetadata: SessionMetadata;
13-
pendingPrompt: AbortController | null;
13+
currentTurnId: string | null;
1414
}
1515

1616
export class CodexAcpServer implements acp.Agent {
@@ -62,7 +62,7 @@ export class CodexAcpServer implements acp.Agent {
6262
const {sessionId, currentModelId, models} = sessionMetadata;
6363
this.sessions.set(sessionId, {
6464
sessionMetadata: sessionMetadata,
65-
pendingPrompt: null
65+
currentTurnId: null
6666
});
6767

6868
const availableModels = this.buildAvailableModels(models);
@@ -153,25 +153,34 @@ export class CodexAcpServer implements acp.Agent {
153153
async prompt(params: acp.PromptRequest): Promise<acp.PromptResponse> {
154154
const sessionState = this.getSessionState(params.sessionId);
155155

156-
sessionState.pendingPrompt?.abort();
157-
sessionState.pendingPrompt = new AbortController();
156+
sessionState.currentTurnId = null;
158157

159158
try {
160159
const messageHandler = new CodexEventHandler(this.connection, sessionState);
161-
await this.runWithProcessCheck(() => this.codexAcpClient.sendPrompt(params, (event) => messageHandler.handleNotification(event)));
162-
} catch (err) {
163-
if (sessionState.pendingPrompt.signal.aborted) {
164-
return {stopReason: "cancelled"};
160+
const turnCompleted = await this.runWithProcessCheck(() => this.codexAcpClient.sendPrompt(
161+
params,
162+
(event) => messageHandler.handleNotification(event),
163+
(turnId) => {
164+
sessionState.currentTurnId = turnId;
165+
}
166+
));
167+
168+
// Check if turn was interrupted (cancelled)
169+
if (turnCompleted.turn.status === "interrupted") {
170+
return {
171+
stopReason: "cancelled",
172+
};
165173
}
166174

175+
return {
176+
stopReason: "end_turn",
177+
};
178+
} catch (err) {
179+
console.error(`[CodexAcpServer] prompt() failed:`, err);
167180
throw err;
181+
} finally {
182+
sessionState.currentTurnId = null;
168183
}
169-
170-
sessionState.pendingPrompt = null;
171-
172-
return {
173-
stopReason: "end_turn",
174-
};
175184
}
176185

177186
private async runWithProcessCheck<T>(operation: () => Promise<T>): Promise<T> {
@@ -191,7 +200,27 @@ export class CodexAcpServer implements acp.Agent {
191200
}
192201

193202
async cancel(params: acp.CancelNotification): Promise<void> {
194-
//TODO not supported yet
195-
this.sessions.get(params.sessionId)?.pendingPrompt?.abort();
203+
const sessionState = this.sessions.get(params.sessionId);
204+
if (!sessionState) {
205+
console.info(`Can not cancel: session ${params.sessionId} not found`);
206+
return;
207+
}
208+
209+
if (!sessionState.currentTurnId) {
210+
console.info(`Can not cancel: session ${params.sessionId} has no current turn`);
211+
return;
212+
}
213+
214+
console.info(`Cancel session ${params.sessionId}, currentTurnId: ${sessionState.currentTurnId}...`);
215+
try {
216+
// After turnInterrupt(), Codex will send turn/completed event, which will naturally complete awaitTurnCompleted()
217+
await this.codexAcpClient.turnInterrupt({
218+
threadId: params.sessionId,
219+
turnId: sessionState.currentTurnId
220+
});
221+
console.log(`Cancel - turnInterrupt succeeded`);
222+
} catch (err) {
223+
console.error(`Cancel - turnInterrupt failed:`, err);
224+
}
196225
}
197226
}

‎src/CodexAppServerClient.ts‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@ import type {
1313
ThreadStartParams,
1414
ThreadStartResponse,
1515
TurnCompletedNotification,
16+
TurnInterruptParams,
17+
TurnInterruptResponse,
1618
TurnStartParams,
1719
TurnStartResponse
1820
} from "./app-server/v2";
@@ -45,6 +47,10 @@ export class CodexAppServerClient {
4547
return await this.sendRequest({ method: "turn/start", params: params });
4648
}
4749

50+
async turnInterrupt(params: TurnInterruptParams): Promise<TurnInterruptResponse> {
51+
return await this.sendRequest({ method: "turn/interrupt", params: params });
52+
}
53+
4854
async threadStart(params: ThreadStartParams): Promise<ThreadStartResponse> {
4955
return await this.sendRequest({ method: "thread/start", params: params });
5056
}

‎src/__tests__/CodexACPAgent/CodexAcpClient.test.ts‎

Lines changed: 25 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -77,10 +77,15 @@ describe('ACP server test', { timeout: 40_000 }, () => {
7777

7878
const codexAcpAgent = fixture.getCodexAcpAgent();
7979

80-
fixture.getCodexAppServerClient().turnStart = vi.fn().mockResolvedValue(undefined);
81-
fixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockResolvedValue(undefined);
80+
fixture.getCodexAppServerClient().turnStart = vi.fn().mockResolvedValue({
81+
turn: { id: "turn-id", items: [], status: "inProgress", error: null }
82+
});
83+
fixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockResolvedValue({
84+
threadId: "id",
85+
turn: { id: "turn-id", items: [], status: "completed", error: null }
86+
});
8287
const sessionState: SessionState = {
83-
pendingPrompt: null,
88+
currentTurnId: null,
8489
sessionMetadata: {
8590
sessionId: "id",
8691
currentModelId: "model-id",
@@ -99,11 +104,16 @@ describe('ACP server test', { timeout: 40_000 }, () => {
99104
const mockFixture = createCodexMockTestFixture();
100105
const codexAcpAgent = mockFixture.getCodexAcpAgent();
101106

102-
mockFixture.getCodexAppServerClient().turnStart = vi.fn().mockResolvedValue(undefined);
103-
mockFixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockResolvedValue(undefined);
107+
mockFixture.getCodexAppServerClient().turnStart = vi.fn().mockResolvedValue({
108+
turn: { id: "turn-id", items: [], status: "inProgress", error: null }
109+
});
110+
mockFixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockResolvedValue({
111+
threadId: "id",
112+
turn: { id: "turn-id", items: [], status: "completed", error: null }
113+
});
104114

105115
const sessionState: SessionState = {
106-
pendingPrompt: null,
116+
currentTurnId: null,
107117
sessionMetadata: {
108118
sessionId: "id",
109119
currentModelId: "model-id",
@@ -143,19 +153,24 @@ describe('ACP server test', { timeout: 40_000 }, () => {
143153
const mockFixture = createCodexMockTestFixture();
144154
const codexAcpAgent = mockFixture.getCodexAcpAgent();
145155

146-
mockFixture.getCodexAppServerClient().turnStart = vi.fn().mockResolvedValue(undefined);
147-
mockFixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockResolvedValue(undefined);
156+
mockFixture.getCodexAppServerClient().turnStart = vi.fn().mockResolvedValue({
157+
turn: { id: "turn-id", items: [], status: "inProgress", error: null }
158+
});
159+
mockFixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockResolvedValue({
160+
threadId: "id",
161+
turn: { id: "turn-id", items: [], status: "completed", error: null }
162+
});
148163

149164
const sessionState1: SessionState = {
150-
pendingPrompt: null,
165+
currentTurnId: null,
151166
sessionMetadata: {
152167
sessionId: "session-1",
153168
currentModelId: "model-id",
154169
models: [],
155170
}
156171
};
157172
const sessionState2: SessionState = {
158-
pendingPrompt: null,
173+
currentTurnId: null,
159174
sessionMetadata: {
160175
sessionId: "session-2",
161176
currentModelId: "model-id",

0 commit comments

Comments
 (0)