feat: Add experimental ACP session compaction updates

docs/session-compaction.md

@@ -0,0 +1,23 @@
+# Session compaction
+
+The adapter implements the [ACP session compaction RFD](https://agentclientprotocol.com/rfds/session-compaction) for ACP v1. Clients opt in during initialization:
+
+```json
+{
+  "clientCapabilities": {
+    "session": {
+      "compaction": {}
+    }
+  }
+}
+```
+
+For these clients, both automatic compaction and `/compact` produce `session/update` notifications with `sessionUpdate: "compaction_update"`. A Codex `contextCompaction` item starts an `in_progress` entity; its completion updates the same `compactionId` to `completed`. A failure or interruption closes an unfinished entity as `failed` or `cancelled`. Successful compactions stay completed if the surrounding turn subsequently fails or is interrupted.
+
+The adapter uses Codex's item ID as the compaction ID. It suppresses duplicate completion signals, including the older `thread/compacted` notification. If only that older notification is available, it emits a completed entity with an ID derived from the turn. That legacy signal cannot distinguish multiple compactions within one turn.
+
+Loading a session replays each persisted compaction as one completed update in its history position. Current Codex paginated history preserves the live item ID. Older legacy histories can reconstruct item IDs, and Codex does not persist failed or interrupted compaction items, so those entries are not available for replay.
+
+Codex's app-server compaction items expose lifecycle identity without a user-displayable summary. The adapter therefore omits `summary` and does not emit `compaction_summary_chunk`. It does not extract internal replacement history or encrypted compaction data. Context utilization continues to arrive separately through `usage_update`.
+
+When the client omits `session.compaction` or sets it to `null`, the adapter preserves its existing synthetic tool-call and text-message fallback.

src/CodexAcpClient.ts

@@ -662,8 +662,12 @@ export class CodexAcpClient {
         }, onTurnStarted);
     }
 
-    async runCompact(sessionId: string): Promise<void> {
-        await this.codexClient.runCompact({threadId: sessionId});
+    async runCompact(
+        sessionId: string,
+        onTurnStarted?: (turnId: string) => void,
+    ): Promise<TurnCompletedNotification | undefined> {
+        const completed = await this.codexClient.runCompact({threadId: sessionId}, onTurnStarted);
+        return completed.method === "turn/completed" ? completed.params : undefined;
     }
 
     async getGoal(sessionId: string): Promise<ThreadGoal | null> {

src/CodexAcpServer.ts

@@ -144,6 +144,7 @@ import {
 } from "./AirExtension";
 import {ASYNC_TASK_STOP_METHOD} from "./async-tasks/AsyncTaskExtension";
 import {CodexBackgroundTerminalTasks} from "./async-tasks/CodexBackgroundTerminalTasks";
+import {clientSupportsCompaction, CodexSessionCompactions, createCompactionUpdate} from "./CodexSessionCompactions";
 import {
     type AgentFileChangeReport,
     type AgentFileChangeReportRequest,
@@ -184,6 +185,7 @@ export interface SessionState {
     titleGen?: TitleGenerator;
     subagents: CodexSubagentEventRouter;
     asyncTasks: CodexBackgroundTerminalTasks;
+    compactions: CodexSessionCompactions;
 }
 
 export type SessionFailureCategory =
@@ -691,6 +693,7 @@ export class CodexAcpServer {
                 new ACPSessionConnection(this.connection, sessionId),
             ),
             asyncTasks: this.createAsyncTasks(sessionId),
+            compactions: new CodexSessionCompactions(),
         };
         sessionState.titleGen = new TitleGenerator(
             this.codexAcpClient.appServerClient,
@@ -1948,6 +1951,7 @@ export class CodexAcpServer {
                 new ACPSessionConnection(this.connection, sessionId),
             ),
             asyncTasks: this.createAsyncTasks(sessionId),
+            compactions: new CodexSessionCompactions(),
         };
         sessionState.titleGen = new TitleGenerator(
             this.codexAcpClient.appServerClient,
@@ -2276,7 +2280,9 @@ export class CodexAcpServer {
             case "exitedReviewMode":
                 return [this.createReviewModeUpdate(item, false)];
             case "contextCompaction":
-                return [createCompletedContextCompactionUpdate(item)];
+                return [clientSupportsCompaction(this.clientCapabilities)
+                    ? createCompactionUpdate(item.id, "completed")
+                    : createCompletedContextCompactionUpdate(item)];
             case "plan":
                 return item.text.length > 0 ? [this.createPlanHistoryUpdate(item)] : [];
         }
@@ -2792,6 +2798,7 @@ export class CodexAcpServer {
                 this.sessionFailureEpoch,
                 sessionState.subagents,
                 (accountUpdated) => this.handleAccountUpdated(accountUpdated),
+                clientSupportsCompaction(this.clientCapabilities),
             );
             eventHandler = promptEventHandler;
             const permissionLifecycle = this.permissionLifecycleContext(sessionState);
@@ -3190,7 +3197,7 @@ export class CodexAcpServer {
                         : "failed",
                 );
             } catch (error) {
-                logger.error("Failed to publish terminal subagent state during prompt cleanup", error);
+                logger.error("Failed to publish terminal compaction or subagent state during prompt cleanup", error);
             }
             if (agentFileChangeReportRequest !== null) {
                 await this.publishAgentFileChangeReport(

src/CodexAppServerClient.ts

@@ -161,7 +161,7 @@ export class CodexAppServerClient {
     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 turnRoutingCaptures = new Map<string, Set<(turnId: string) => void>>();
+    private readonly turnRoutingCaptures = new Map<string, Set<(turnId: string, notification: ServerNotification) => void>>();
     private readonly threadStatusCaptures = new Map<string, Set<(status: ThreadStatus) => void>>();
     private readonly threadGoalUpdateCaptures = new Map<string, Set<(event: ThreadGoalUpdatedNotification) => void>>();
     private readonly threadGoalClearedCaptures = new Map<string, Set<() => void>>();
@@ -204,7 +204,7 @@ export class CodexAppServerClient {
             if (this.handleStaleTurnNotification(serverNotification, routing)) {
                 return;
             }
-            this.recordTurnRouting(routing);
+            this.recordTurnRouting(routing, serverNotification);
             if (this.handleStaleTurnNotification(serverNotification, routing)) {
                 return;
             }
@@ -518,10 +518,48 @@ export class CodexAppServerClient {
         };
     }
 
-    async runCompact(params: ThreadCompactStartParams): Promise<CompactionCompletedNotification> {
-        const compactionCompleted = this.awaitCompactionCompleted(params.threadId);
-        await this.threadCompactStart(params);
-        return await compactionCompleted;
+    async runCompact(
+        params: ThreadCompactStartParams,
+        onTurnStarted?: (turnId: string) => void,
+    ): Promise<CompactionCompletedNotification | Extract<ServerNotification, {method: "turn/completed"}>> {
+        type Result = CompactionCompletedNotification | Extract<ServerNotification, {method: "turn/completed"}>;
+        let compactTurnId: string | null = null;
+        let resolveCompleted: (event: Result) => void = () => {};
+        let rejectCompleted: (error: Error) => void = () => {};
+        const completed = new Promise<Result>((resolve, reject) => {
+            resolveCompleted = resolve;
+            rejectCompleted = reject;
+        });
+        // The request acknowledgement can still be pending when the connection closes.
+        void completed.catch(() => {});
+        const closed = this.connection.onClose?.(() => rejectCompleted(new Error("Codex connection closed during compaction.")));
+        const completeCompaction = (event: CompactionCompletedNotification) => {
+            if (compactTurnId !== null && event.params.turnId !== compactTurnId) return;
+            resolveCompleted(event);
+        };
+        const releaseCompactionCapture = this.captureCompactionCompletions(params.threadId, completeCompaction);
+        const releaseTurnCapture = this.captureTurnCompletions(params.threadId, (event) => {
+            if (compactTurnId === null || event.turn.id !== compactTurnId) return;
+            if (event.turn.status !== "inProgress") {
+                resolveCompleted({method: "turn/completed", params: event});
+            }
+        });
+        const releaseRoutingCapture = this.captureTurnRoutings(params.threadId, (turnId, notification) => {
+            if (compactTurnId !== null) return;
+            if (notification.method !== "turn/started"
+                && !(notification.method === "item/started" && notification.params.item.type === "contextCompaction")) return;
+            compactTurnId = turnId;
+            onTurnStarted?.(turnId);
+        });
+        try {
+            await this.threadCompactStart(params);
+            return await completed;
+        } finally {
+            releaseTurnCapture();
+            releaseRoutingCapture();
+            releaseCompactionCapture();
+            closed?.dispose();
+        }
     }
 
     async turnInterrupt(params: TurnInterruptParams): Promise<TurnInterruptResponse> {
@@ -736,9 +774,10 @@ 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);
+            const releaseCapture = this.captureCompactionCompletions(threadId, (event) => {
+                releaseCapture();
+                resolve(event);
+            });
         });
     }
 
@@ -828,12 +867,29 @@ export class CodexAppServerClient {
         if (!resolvers) {
             return;
         }
-        this.pendingCompactionCompletionResolvers.delete(threadId);
         for (const resolve of resolvers) {
             resolve(event);
         }
     }
 
+    private captureCompactionCompletions(
+        threadId: string,
+        capture: (event: CompactionCompletedNotification) => void,
+    ): () => void {
+        const captures = this.pendingCompactionCompletionResolvers.get(threadId) ?? new Set();
+        captures.add(capture);
+        this.pendingCompactionCompletionResolvers.set(threadId, captures);
+        let released = false;
+        return () => {
+            if (released) return;
+            released = true;
+            captures.delete(capture);
+            if (captures.size === 0) {
+                this.pendingCompactionCompletionResolvers.delete(threadId);
+            }
+        };
+    }
+
     private recordThreadStatusChanged(event: ThreadStatusChangedNotification): void {
         const captures = this.threadStatusCaptures.get(event.threadId);
         if (!captures) {
@@ -864,7 +920,10 @@ export class CodexAppServerClient {
         }
     }
 
-    private recordTurnRouting(routing: { threadId: string | null, turnId: string | null }): void {
+    private recordTurnRouting(
+        routing: { threadId: string | null, turnId: string | null },
+        notification: ServerNotification,
+    ): void {
         if (routing.threadId === null || routing.turnId === null) {
             return;
         }
@@ -873,7 +932,7 @@ export class CodexAppServerClient {
             return;
         }
         for (const capture of captures) {
-            capture(routing.turnId);
+            capture(routing.turnId, notification);
         }
     }
 
@@ -938,8 +997,8 @@ export class CodexAppServerClient {
         };
     }
 
-    private captureTurnRoutings(threadId: string, capture: (turnId: string) => void): () => void {
-        const captures = this.turnRoutingCaptures.get(threadId) ?? new Set<(turnId: string) => void>();
+    private captureTurnRoutings(threadId: string, capture: (turnId: string, notification: ServerNotification) => void): () => void {
+        const captures = this.turnRoutingCaptures.get(threadId) ?? new Set<(turnId: string, notification: ServerNotification) => void>();
         captures.add(capture);
         this.turnRoutingCaptures.set(threadId, captures);
         let released = false;

src/CodexCommands.ts

@@ -226,8 +226,12 @@ export class CodexCommands {
                 return { handled: options.setConfigOption !== undefined };
             }
             case "compact": {
-                await this.runWithProcessCheck(() => this.codexAcpClient.runCompact(sessionId));
-                return { handled: true };
+                options.onTurnStartPending?.();
+                const turnCompleted = await this.runWithProcessCheck(() => this.codexAcpClient.runCompact(
+                    sessionId,
+                    (turnId) => options.onTurnStarted?.(turnId, sessionId),
+                ));
+                return { handled: true, ...(turnCompleted === undefined ? {} : {turnCompleted}) };
             }
             case "goal": {
                 return await this.runGoalCommand(sessionState, command.rest, options);

src/CodexEventHandler.ts

@@ -243,6 +243,7 @@ export class CodexEventHandler {
             new ACPSessionConnection(connection, sessionState.sessionId),
         ),
         onAccountUpdated?: (notification: AccountUpdatedNotification) => void,
+        private readonly supportsCompaction = false,
     ) {
         this.onAccountUpdated = onAccountUpdated;
         this.sessionState = sessionState;
@@ -294,6 +295,7 @@ export class CodexEventHandler {
             await this.handleNotification(notification);
             return;
         }
+        await this.finishCompactionsForNotification(notification);
         if (notification.params.willRetry) {
             await this.session.update(this.createSessionFailureUpdate(this.recordRetryWarning(notification.params, false)));
             return;
@@ -375,8 +377,13 @@ export class CodexEventHandler {
 
     async handleNotification(notification: ServerNotification) {
         await this.flushPendingErrors();
+        await this.finishCompactionsForNotification(notification);
         const closingChildren = this.subagents.closingChildSessions(notification);
         for (const child of closingChildren) {
+            await this.finishOutstandingCompactions(
+                child.state === "cancelled" ? "cancelled" : "failed",
+                child.sessionId,
+            );
             await this.sessionState.asyncTasks.reconcile(child.threadId, child.sessionId);
         }
         const handledBySubagents = await this.subagents.handle(notification);
@@ -411,13 +418,49 @@ export class CodexEventHandler {
     }
 
     async waitForNativeSubagents(signal: AbortSignal): Promise<void> {
-        await this.subagents.wait(signal);
+        if (await this.subagents.wait(signal) === "timed_out") {
+            await this.finishOutstandingNativeSubagents("failed");
+        }
     }
 
     async finishOutstandingNativeSubagents(state: SubagentState): Promise<void> {
+        await this.finishOutstandingCompactions(state === "cancelled" ? "cancelled" : "failed");
         await this.subagents.finishOutstanding(state);
     }
 
+    async finishOutstandingCompactions(status: "failed" | "cancelled", sessionId?: string): Promise<void> {
+        if (!this.supportsCompaction) return;
+        for (const {sessionId: targetSessionId, update} of this.sessionState.compactions.finishOutstanding(status, sessionId)) {
+            await this.session.update(update, targetSessionId);
+        }
+    }
+
+    private async finishCompactionsForNotification(notification: ServerNotification): Promise<void> {
+        if (!this.supportsCompaction) return;
+        let updates: UpdateSessionEvent[];
+        const sessionId = this.subagents.notificationSessionId(notification);
+        if (notification.method === "turn/completed") {
+            const turn = notification.params.turn;
+            if (turn.status === "inProgress") return;
+            updates = this.sessionState.compactions.finishTurn(
+                sessionId,
+                turn.id,
+                turn.status === "interrupted" ? "cancelled" : "failed",
+                turn.error?.message ?? "Codex ended the turn before compaction completed.",
+            );
+        } else if (notification.method === "error" && !notification.params.willRetry) {
+            updates = this.sessionState.compactions.finishTurn(
+                sessionId,
+                notification.params.turnId,
+                "failed",
+                notification.params.error.message,
+            );
+        } else {
+            return;
+        }
+        for (const update of updates) await this.session.update(update, sessionId);
+    }
+
     async flushPendingPlanUpdates(): Promise<void> {
         this.cancelPlanUpdateTimer();
         do {
@@ -538,7 +581,11 @@ export class CodexEventHandler {
             case "item/autoApprovalReview/completed":
                 return this.handleGuardianApprovalReviewCompleted(notification.params);
             case "thread/compacted":
-                return this.createContextCompactedEvent();
+                return this.supportsCompaction
+                    ? this.sessionState.compactions.completeLegacy(
+                        this.subagents.notificationSessionId(notification), notification.params.turnId,
+                    )
+                    : this.createContextCompactedEvent();
             case "item/reasoning/summaryTextDelta":
                 this.completeRetryIncidentOnTurnProgress();
                 return this.createReasoningDeltaEvent(notification.params);
@@ -748,7 +795,12 @@ export class CodexEventHandler {
                 this.rememberAgentMessagePhase(event.item);
                 return null;
             case "contextCompaction":
-                return createContextCompactionStartUpdate(event.item);
+                return this.supportsCompaction
+                    ? this.sessionState.compactions.start(
+                        this.subagents.notificationSessionId({method: "item/started", params: event}),
+                        event.turnId, event.item.id,
+                    )
+                    : createContextCompactionStartUpdate(event.item);
             case "subAgentActivity":
                 return this.subagents.legacyActivityStarted(event.item);
             case "sleep":
@@ -817,7 +869,12 @@ export class CodexEventHandler {
             case "exitedReviewMode":
                 return this.createExitedReviewModeEvent(event.item);
             case "contextCompaction":
-                return createContextCompactionCompleteUpdate(event.item);
+                return this.supportsCompaction
+                    ? this.sessionState.compactions.complete(
+                        this.subagents.notificationSessionId({method: "item/completed", params: event}),
+                        event.turnId, event.item.id,
+                    )
+                    : createContextCompactionCompleteUpdate(event.item);
             //ignored types
             case "subAgentActivity":
                 return this.subagents.legacyActivityCompleted(event.item);