feat: Add review and compact slash commands

Author: benbrandt

src/CodexAcpClient.ts

@@ -28,6 +28,7 @@ import type {
     GetAccountResponse,
     ListMcpServerStatusResponse,
     Model,
+    ReviewTarget,
     SkillsListParams,
     SkillsListResponse,
     Thread,
@@ -274,6 +275,22 @@ export class CodexAcpClient {
         await this.codexClient.threadArchive({threadId: sessionId});
     }
 
+    async runReview(
+        sessionId: string,
+        target: ReviewTarget,
+        onTurnStarted?: (turnId: string) => void,
+    ): Promise<TurnCompletedNotification> {
+        return await this.codexClient.runReview({
+            threadId: sessionId,
+            target,
+            delivery: "inline",
+        }, onTurnStarted);
+    }
+
+    async runCompact(sessionId: string): Promise<void> {
+        await this.codexClient.runCompact({threadId: sessionId});
+    }
+
     async awaitMcpServerStartup(serverNames: Array<string>, afterVersion: number): Promise<McpStartupResult> {
         return await this.codexClient.awaitMcpServerStartup(serverNames, afterVersion);
     }

src/CodexAcpServer.ts

@@ -1259,8 +1259,34 @@ export class CodexAcpServer implements acp.Agent {
                 approvalHandler,
                 elicitationHandler);
 
-            if (await this.availableCommands.tryHandleCommand(params.prompt, sessionState)) {
+            const commandResult = await this.availableCommands.tryHandleCommand(params.prompt, 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.sessionUpdate({
+                            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),
+                    };
+                }
+                const error = eventHandler.getFailure()
+                if (error) {
+                    // noinspection ExceptionCaughtLocallyJS
+                    throw error;
+                }
                 return {
                     stopReason: "end_turn",
                     usage: this.buildPromptUsage(sessionState.lastTokenUsage),

src/CodexAppServerClient.ts

@@ -21,11 +21,15 @@ import type {
     McpServerStatusUpdatedNotification,
     ModelListParams,
     ModelListResponse,
+    ReviewStartParams,
+    ReviewStartResponse,
     SkillsExtraRootsSetParams,
     SkillsListParams,
     SkillsListResponse,
     ThreadArchiveParams,
     ThreadArchiveResponse,
+    ThreadCompactStartParams,
+    ThreadCompactStartResponse,
     ThreadLoadedListParams,
     ThreadLoadedListResponse,
     ThreadListParams,
@@ -47,6 +51,7 @@ import type {
     CommandExecutionRequestApprovalResponse,
     FileChangeRequestApprovalParams,
     FileChangeRequestApprovalResponse,
+    ItemCompletedNotification,
 } from "./app-server/v2";
 
 export interface ApprovalHandler {
@@ -99,6 +104,7 @@ export class CodexAppServerClient {
     private readonly mcpServerStartupStates = new Map<string, McpServerStartupSnapshot>();
     private readonly mcpServerStartupResolvers: Array<McpServerStartupResolver> = [];
     private readonly pendingTurnCompletionResolvers = new Map<string, Map<string, (event: TurnCompletedNotification) => void>>();
+    private readonly pendingCompactionCompletionResolvers = new Map<string, Set<(event: CompactionCompletedNotification) => void>>();
     private readonly turnCompletionCaptures = new Map<string, Set<(event: TurnCompletedNotification) => void>>();
     private readonly staleTurnIds = new Map<string, Set<string>>();
 
@@ -118,6 +124,9 @@ export class CodexAppServerClient {
             if (isTurnCompletedNotification(serverNotification)) {
                 this.recordTurnCompleted(serverNotification.params);
             }
+            if (isCompactionCompletedNotification(serverNotification)) {
+                this.recordCompactionCompleted(serverNotification);
+            }
             const routing = extractTurnRouting(serverNotification);
             const staleTurnNotification = this.isStaleTurn(routing.threadId, routing.turnId);
             if (staleTurnNotification) {
@@ -213,10 +222,40 @@ export class CodexAppServerClient {
         }
     }
 
+    async runReview(params: ReviewStartParams, onTurnStarted?: (turnId: string) => void): Promise<TurnCompletedNotification> {
+        const capturedCompletions: Array<TurnCompletedNotification> = [];
+        const releaseCapture = this.captureTurnCompletions(params.threadId, (event) => {
+            capturedCompletions.push(event);
+        });
+
+        try {
+            const reviewStarted = await this.reviewStart(params);
+            onTurnStarted?.(reviewStarted.turn.id);
+            const earlyCompletion = capturedCompletions.find(event => event.turn.id === reviewStarted.turn.id);
+            releaseCapture();
+            if (earlyCompletion) {
+                return earlyCompletion;
+            }
+            return await this.awaitTurnCompleted(reviewStarted.reviewThreadId, reviewStarted.turn.id);
+        } finally {
+            releaseCapture();
+        }
+    }
+
+    async runCompact(params: ThreadCompactStartParams): Promise<CompactionCompletedNotification> {
+        const compactionCompleted = this.awaitCompactionCompleted(params.threadId);
+        await this.threadCompactStart(params);
+        return await compactionCompleted;
+    }
+
     async turnInterrupt(params: TurnInterruptParams): Promise<TurnInterruptResponse> {
         return await this.sendRequest({ method: "turn/interrupt", params: params });
     }
 
+    async reviewStart(params: ReviewStartParams): Promise<ReviewStartResponse> {
+        return await this.sendRequest({ method: "review/start", params: params });
+    }
+
     markTurnStale(threadId: string, turnId: string): void {
         const threadStaleTurns = this.staleTurnIds.get(threadId) ?? new Set<string>();
         threadStaleTurns.add(turnId);
@@ -251,6 +290,10 @@ export class CodexAppServerClient {
         return await this.sendRequest({ method: "thread/unsubscribe", params: params });
     }
 
+    async threadCompactStart(params: ThreadCompactStartParams): Promise<ThreadCompactStartResponse> {
+        return await this.sendRequest({ method: "thread/compact/start", params: params });
+    }
+
     async listMcpServerStatus(params: ListMcpServerStatusParams): Promise<ListMcpServerStatusResponse> {
         return await this.sendRequest({ method: "mcpServerStatus/list", params });
     }
@@ -303,6 +346,14 @@ export class CodexAppServerClient {
         });
     }
 
+    async awaitCompactionCompleted(threadId: string): Promise<CompactionCompletedNotification> {
+        return await new Promise((resolve) => {
+            const resolvers = this.pendingCompactionCompletionResolvers.get(threadId) ?? new Set();
+            resolvers.add(resolve);
+            this.pendingCompactionCompletionResolvers.set(threadId, resolvers);
+        });
+    }
+
     resolveTurnInterrupted(threadId: string, turnId: string): void {
         this.recordTurnCompleted({
             threadId,
@@ -380,6 +431,21 @@ export class CodexAppServerClient {
         }
     }
 
+    private recordCompactionCompleted(event: CompactionCompletedNotification): void {
+        const threadId = extractThreadId(event);
+        if (threadId === null) {
+            return;
+        }
+        const resolvers = this.pendingCompactionCompletionResolvers.get(threadId);
+        if (!resolvers) {
+            return;
+        }
+        this.pendingCompactionCompletionResolvers.delete(threadId);
+        for (const resolve of resolvers) {
+            resolve(event);
+        }
+    }
+
     private isStaleTurn(threadId: string | null, turnId: string | null): boolean {
         if (threadId === null || turnId === null) {
             return false;
@@ -493,6 +559,10 @@ export type CodexConnectionEvent =
     | ({ eventType: "response" } & unknown)
     | ({ eventType: "notification" } & ServerNotification);
 
+export type CompactionCompletedNotification =
+    | { method: "thread/compacted", params: Extract<ServerNotification, { method: "thread/compacted" }>["params"] }
+    | { method: "item/completed", params: ItemCompletedNotification & { item: Extract<ItemCompletedNotification["item"], { type: "contextCompaction" }> } };
+
 type CodexRequest = DistributiveOmit<ClientRequest, "id">
 
 type DistributiveOmit<T, K extends keyof any> = T extends any
@@ -525,6 +595,13 @@ function isTurnCompletedNotification(notification: ServerNotification): notifica
     return notification.method === "turn/completed";
 }
 
+function isCompactionCompletedNotification(notification: ServerNotification): notification is CompactionCompletedNotification {
+    if (notification.method === "thread/compacted") {
+        return true;
+    }
+    return notification.method === "item/completed" && notification.params.item.type === "contextCompaction";
+}
+
 function extractThreadId(notification: ServerNotification): string | null {
     const params = notification.params as { threadId?: unknown } | undefined;
     if (params && typeof params.threadId === "string") {