Skip to content

Commit cefdd0e

Browse files
benbrandtbenbrandt
committed
Merge branch 'main' into goal
2 parents 4a88b23 + 1f58237 commit cefdd0e

10 files changed

Lines changed: 460 additions & 95 deletions

‎package-lock.json‎

Lines changed: 4 additions & 4 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@
6161
"vitest": "^4.0.10"
6262
},
6363
"dependencies": {
64-
"@agentclientprotocol/sdk": "^0.28.1",
64+
"@agentclientprotocol/sdk": "^0.29.0",
6565
"@openai/codex": "^0.142.2",
6666
"diff": "^8.0.3",
6767
"open": "^11.0.0",

‎src/CodexAcpClient.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -310,7 +310,7 @@ export class CodexAcpClient {
310310
async runReview(
311311
sessionId: string,
312312
target: ReviewTarget,
313-
onTurnStarted?: (turnId: string) => void,
313+
onTurnStarted?: (turnId: string, threadId: string) => void,
314314
): Promise<TurnCompletedNotification> {
315315
return await this.codexClient.runReview({
316316
threadId: sessionId,

‎src/CodexAcpServer.ts‎

Lines changed: 167 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,10 @@ interface PendingTurnStart {
108108
interface ActivePrompt {
109109
completion: Promise<void>;
110110
closeSignal: Promise<null>;
111+
cancelSignal: Promise<null>;
112+
signal: AbortSignal;
113+
currentTurn: { threadId: string, turnId: string } | null;
114+
requestCancel: () => void;
111115
requestClose: () => void;
112116
complete: () => void;
113117
}
@@ -1124,17 +1128,33 @@ export class CodexAcpServer {
11241128
const closeSignal = new Promise<null>((resolve) => {
11251129
resolveCloseSignal = resolve;
11261130
});
1131+
let resolveCancelSignal: (value: null) => void = () => {};
1132+
const cancelSignal = new Promise<null>((resolve) => {
1133+
resolveCancelSignal = resolve;
1134+
});
1135+
const abortController = new AbortController();
11271136

11281137
let completed = false;
11291138
let closeRequested = false;
11301139
const activePrompt: ActivePrompt = {
11311140
completion,
11321141
closeSignal,
1142+
cancelSignal,
1143+
signal: abortController.signal,
1144+
currentTurn: null,
1145+
requestCancel: () => {
1146+
if (abortController.signal.aborted) {
1147+
return;
1148+
}
1149+
abortController.abort();
1150+
resolveCancelSignal(null);
1151+
},
11331152
requestClose: () => {
11341153
if (closeRequested) {
11351154
return;
11361155
}
11371156
closeRequested = true;
1157+
activePrompt.requestCancel();
11381158
resolveCloseSignal(null);
11391159
},
11401160
complete: () => {
@@ -1153,6 +1173,46 @@ export class CodexAcpServer {
11531173
return activePrompt;
11541174
}
11551175

1176+
private cancelBeforeTurnStarted(activePrompt: ActivePrompt): Promise<null> {
1177+
return activePrompt.cancelSignal.then(() => {
1178+
if (activePrompt.currentTurn === null) {
1179+
return null;
1180+
}
1181+
return new Promise<null>(() => {});
1182+
});
1183+
}
1184+
1185+
private observePromptRequestCancellation(
1186+
signal: AbortSignal | undefined,
1187+
sessionState: SessionState,
1188+
activePrompt: ActivePrompt,
1189+
): () => void {
1190+
if (!signal) {
1191+
return () => {};
1192+
}
1193+
1194+
const onAbort = () => {
1195+
if (this.activePrompts.get(sessionState.sessionId) !== activePrompt) {
1196+
return;
1197+
}
1198+
logger.log("Prompt request cancelled", {sessionId: sessionState.sessionId});
1199+
activePrompt.requestCancel();
1200+
const turn = activePrompt.currentTurn;
1201+
if (!turn) {
1202+
return;
1203+
}
1204+
void this.requestTurnInterrupt(turn, "Cancel");
1205+
};
1206+
1207+
if (signal.aborted) {
1208+
onAbort();
1209+
return () => {};
1210+
}
1211+
1212+
signal.addEventListener("abort", onAbort, {once: true});
1213+
return () => signal.removeEventListener("abort", onAbort);
1214+
}
1215+
11561216
private createPendingTurnStart(): PendingTurnStart {
11571217
let resolve: (turnId: string | null) => void = () => {};
11581218
const promise = new Promise<string | null>((innerResolve) => {
@@ -1161,26 +1221,48 @@ export class CodexAcpServer {
11611221
return {promise, resolve};
11621222
}
11631223

1164-
private interruptLateStartedTurn(sessionId: string, turnId: string): void {
1224+
private async interruptPromptTurn(
1225+
turn: { threadId: string, turnId: string },
1226+
requestName: "Cancel" | "Close",
1227+
): Promise<void> {
11651228
this.codexAcpClient.markTurnStale({
1166-
threadId: sessionId,
1167-
turnId,
1229+
threadId: turn.threadId,
1230+
turnId: turn.turnId,
11681231
});
1169-
void this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({
1170-
threadId: sessionId,
1171-
turnId,
1172-
})).catch((err) => {
1173-
logger.error(`Close - late turnInterrupt failed`, err);
1174-
}).finally(() => {
1232+
try {
1233+
await this.requestTurnInterrupt(turn, requestName);
1234+
} finally {
11751235
this.codexAcpClient.resolveTurnInterrupted({
1176-
threadId: sessionId,
1177-
turnId,
1236+
threadId: turn.threadId,
1237+
turnId: turn.turnId,
11781238
});
1179-
});
1239+
}
1240+
}
1241+
1242+
private async requestTurnInterrupt(
1243+
turn: { threadId: string, turnId: string },
1244+
requestName: "Cancel" | "Close",
1245+
): Promise<void> {
1246+
try {
1247+
await this.runWithProcessCheck(() => this.codexAcpClient.turnInterrupt({
1248+
threadId: turn.threadId,
1249+
turnId: turn.turnId,
1250+
}));
1251+
logger.log(`${requestName} - turnInterrupt succeeded`, {
1252+
sessionId: turn.threadId,
1253+
currentTurnId: turn.turnId,
1254+
});
1255+
} catch (err) {
1256+
logger.error(`${requestName} - turnInterrupt failed`, err);
1257+
}
11801258
}
11811259

1182-
private promptIsClosedOrStale(sessionId: string, activePrompt: ActivePrompt): boolean {
1183-
return this.activePrompts.get(sessionId) !== activePrompt || this.sessionIsClosing(sessionId);
1260+
private interruptLateStartedTurn(turn: { threadId: string, turnId: string }): void {
1261+
void this.interruptPromptTurn(turn, "Close");
1262+
}
1263+
1264+
private promptShouldStop(sessionId: string, activePrompt: ActivePrompt): boolean {
1265+
return activePrompt.signal.aborted || this.activePrompts.get(sessionId) !== activePrompt || this.sessionIsClosing(sessionId);
11841266
}
11851267

11861268
private async interruptSessionTurn(
@@ -1250,7 +1332,7 @@ export class CodexAcpServer {
12501332
return turnId;
12511333
}
12521334

1253-
async prompt(params: acp.PromptRequest): Promise<acp.PromptResponse> {
1335+
async prompt(params: acp.PromptRequest, signal?: AbortSignal): Promise<acp.PromptResponse> {
12541336
logger.log("Prompt received", {
12551337
sessionId: params.sessionId,
12561338
prompt: params.prompt,
@@ -1260,11 +1342,12 @@ export class CodexAcpServer {
12601342
sessionState.lastTokenUsage = null;
12611343
const activePrompt = this.trackActivePrompt(params.sessionId);
12621344
let pendingTurnStart: PendingTurnStart | null = null;
1345+
const disposePromptRequestCancellation = this.observePromptRequestCancellation(signal, sessionState, activePrompt);
12631346

12641347
try {
12651348
const eventHandler = new CodexEventHandler(this.connection, sessionState);
1266-
const approvalHandler = new CodexApprovalHandler(this.connection, sessionState);
1267-
const elicitationHandler = new CodexElicitationHandler(this.connection, sessionState);
1349+
const approvalHandler = new CodexApprovalHandler(this.connection, sessionState, activePrompt.signal);
1350+
const elicitationHandler = new CodexElicitationHandler(this.connection, sessionState, activePrompt.signal);
12681351
await this.codexAcpClient.subscribeToSessionEvents(params.sessionId,
12691352
(event) => {
12701353
elicitationHandler.handleNotification(event);
@@ -1273,29 +1356,40 @@ export class CodexAcpServer {
12731356
approvalHandler,
12741357
elicitationHandler);
12751358

1276-
const commandResult = await this.availableCommands.tryHandleCommand(params.prompt, sessionState);
1359+
if (activePrompt.signal.aborted) {
1360+
return this.cancelledPromptResponse(sessionState);
1361+
}
1362+
1363+
const commandPromise = this.availableCommands.tryHandleCommand(params.prompt, sessionState, {
1364+
onTurnStarted: (turnId, threadId) => {
1365+
const turn = {threadId, turnId};
1366+
activePrompt.currentTurn = turn;
1367+
if (this.promptShouldStop(params.sessionId, activePrompt)) {
1368+
this.interruptLateStartedTurn(turn);
1369+
return;
1370+
}
1371+
sessionState.currentTurnId = turnId;
1372+
},
1373+
});
1374+
void commandPromise.catch((err) => {
1375+
if (this.activePrompts.get(params.sessionId) !== activePrompt) {
1376+
logger.error(`Command for cancelled prompt ${params.sessionId} failed after prompt returned`, err);
1377+
}
1378+
});
1379+
const commandResult = await Promise.race([
1380+
commandPromise,
1381+
activePrompt.closeSignal,
1382+
this.cancelBeforeTurnStarted(activePrompt),
1383+
]);
1384+
if (commandResult === null) {
1385+
return this.cancelledPromptResponse(sessionState);
1386+
}
12771387
if (commandResult.handled) {
12781388
logger.log("Prompt handled by a command");
12791389
await this.codexAcpClient.waitForSessionNotifications(params.sessionId);
1280-
const turnCompleted = commandResult.turnCompleted;
1281-
if (turnCompleted?.turn.status === "interrupted") {
1282-
if (!this.sessionIsClosing(params.sessionId) && this.sessions.has(params.sessionId)) {
1283-
await this.connection.notify(acp.methods.client.session.update, {
1284-
sessionId: params.sessionId,
1285-
update: {
1286-
sessionUpdate: "agent_message_chunk",
1287-
content: {
1288-
type: "text",
1289-
text: "*Conversation interrupted*"
1290-
}
1291-
}
1292-
});
1293-
}
1294-
return {
1295-
stopReason: "cancelled",
1296-
usage: this.buildPromptUsage(sessionState.lastTokenUsage),
1297-
_meta: this.buildQuotaMeta(sessionState),
1298-
};
1390+
if (commandResult.turnCompleted?.turn.status === "interrupted") {
1391+
await this.notifyConversationInterrupted(params.sessionId);
1392+
return this.cancelledPromptResponse(sessionState);
12991393
}
13001394
const error = eventHandler.getFailure();
13011395
if (error) {
@@ -1310,11 +1404,7 @@ export class CodexAcpServer {
13101404
}
13111405

13121406
if (this.sessionIsClosing(params.sessionId)) {
1313-
return {
1314-
stopReason: "cancelled",
1315-
usage: this.buildPromptUsage(sessionState.lastTokenUsage),
1316-
_meta: this.buildQuotaMeta(sessionState),
1317-
};
1407+
return this.cancelledPromptResponse(sessionState);
13181408
}
13191409

13201410
const modelId = ModelId.fromString(sessionState.currentModelId);
@@ -1349,54 +1439,37 @@ export class CodexAcpServer {
13491439
sessionState.cwd,
13501440
sessionState.additionalDirectories,
13511441
(turnId) => {
1352-
if (this.promptIsClosedOrStale(params.sessionId, activePrompt)) {
1353-
this.interruptLateStartedTurn(params.sessionId, turnId);
1442+
const turn = {threadId: params.sessionId, turnId};
1443+
activePrompt.currentTurn = turn;
1444+
if (this.promptShouldStop(params.sessionId, activePrompt)) {
1445+
this.interruptLateStartedTurn(turn);
13541446
return;
13551447
}
13561448
sessionState.currentTurnId = turnId;
13571449
pendingTurnStart?.resolve(turnId);
13581450
},
1359-
() => this.promptIsClosedOrStale(params.sessionId, activePrompt),
1451+
() => this.promptShouldStop(params.sessionId, activePrompt),
13601452
));
13611453
void sendPromptPromise.catch((err) => {
13621454
if (this.activePrompts.get(params.sessionId) !== activePrompt) {
1363-
logger.error(`Prompt for closed session ${params.sessionId} failed after close`, err);
1455+
logger.error(`Prompt for cancelled session ${params.sessionId} failed after prompt returned`, err);
13641456
}
13651457
});
13661458
const turnCompleted = await Promise.race([
13671459
sendPromptPromise,
13681460
activePrompt.closeSignal,
1461+
this.cancelBeforeTurnStarted(activePrompt),
13691462
]);
13701463

13711464
if (turnCompleted === null) {
1372-
return {
1373-
stopReason: "cancelled",
1374-
usage: this.buildPromptUsage(sessionState.lastTokenUsage),
1375-
_meta: this.buildQuotaMeta(sessionState),
1376-
};
1465+
return this.cancelledPromptResponse(sessionState);
13771466
}
13781467

13791468
await this.codexAcpClient.waitForSessionNotifications(params.sessionId);
13801469

1381-
// Check if turn was interrupted (cancelled)
13821470
if (turnCompleted.turn.status === "interrupted") {
1383-
if (!this.sessionIsClosing(params.sessionId) && this.sessions.has(params.sessionId)) {
1384-
await this.connection.notify(acp.methods.client.session.update, {
1385-
sessionId: params.sessionId,
1386-
update: {
1387-
sessionUpdate: "agent_message_chunk",
1388-
content: {
1389-
type: "text",
1390-
text: "*Conversation interrupted*"
1391-
}
1392-
}
1393-
});
1394-
}
1395-
return {
1396-
stopReason: "cancelled",
1397-
usage: this.buildPromptUsage(sessionState.lastTokenUsage),
1398-
_meta: this.buildQuotaMeta(sessionState),
1399-
};
1471+
await this.notifyConversationInterrupted(params.sessionId);
1472+
return this.cancelledPromptResponse(sessionState);
14001473
}
14011474

14021475
const error = eventHandler.getFailure();
@@ -1415,6 +1488,7 @@ export class CodexAcpServer {
14151488
throw err;
14161489
} finally {
14171490
logger.log("Prompt completed", {sessionId: params.sessionId});
1491+
disposePromptRequestCancellation();
14181492
sessionState.currentTurnId = null;
14191493
if (pendingTurnStart !== null && this.pendingTurnStarts.get(params.sessionId) === pendingTurnStart) {
14201494
this.pendingTurnStarts.delete(params.sessionId);
@@ -1424,6 +1498,30 @@ export class CodexAcpServer {
14241498
}
14251499
}
14261500

1501+
private cancelledPromptResponse(sessionState: SessionState): acp.PromptResponse {
1502+
return {
1503+
stopReason: "cancelled",
1504+
usage: this.buildPromptUsage(sessionState.lastTokenUsage),
1505+
_meta: this.buildQuotaMeta(sessionState),
1506+
};
1507+
}
1508+
1509+
private async notifyConversationInterrupted(sessionId: string): Promise<void> {
1510+
if (this.sessionIsClosing(sessionId) || !this.sessions.has(sessionId)) {
1511+
return;
1512+
}
1513+
await this.connection.notify(acp.methods.client.session.update, {
1514+
sessionId,
1515+
update: {
1516+
sessionUpdate: "agent_message_chunk",
1517+
content: {
1518+
type: "text",
1519+
text: "*Conversation interrupted*"
1520+
}
1521+
}
1522+
});
1523+
}
1524+
14271525
private buildQuotaMeta(sessionState: SessionState): { quota: QuotaMeta } {
14281526
const lastTokenUsage = sessionState.lastTokenUsage;
14291527

0 commit comments

Comments
 (0)