fix(codex): wait for manual compaction to finish (#1038)

* test: reproduce issue #982

* fix: wait for Codex manual compaction (closes #982)

* test: cover Codex compaction turn completion

cli/src/codex/codexRemoteLauncher.test.ts

@@ -39,6 +39,7 @@ const harness = vi.hoisted(() => ({
     failResumeThreadIds: [] as string[],
     nextThreadSystemErrorMessage: null as string | null,
     failNextCompact: false,
+    deferCompactCompletion: false,
     deferThreadStatusNotifications: false,
     emitChildThreadEvents: false,
     emitChildUsageEvents: false,
@@ -131,6 +132,9 @@ vi.mock('./codexAppServerClient', () => {
                 harness.failNextCompact = false;
                 throw new Error('compact failed');
             }
+            if (harness.deferCompactCompletion) {
+                return {};
+            }
             const compacted = { threadId, turnId: `compact-${harness.compactThreadIds.length}` };
             harness.notifications.push({ method: 'thread/compacted', params: compacted });
             this.notificationHandler?.('thread/compacted', compacted);
@@ -908,10 +912,16 @@ function createMode(): EnhancedMode {
     };
 }
 
-function createSessionStub(messages = ['hello from launcher test'], mode = createMode()) {
+function createSessionStub(
+    messages = ['hello from launcher test'],
+    mode = createMode(),
+    isolateMessages = false
+) {
     const queue = new MessageQueue2<EnhancedMode>((mode) => JSON.stringify(mode));
     messages.forEach((message, index) => {
-        if (index === 0 && messages.length > 1) {
+        if (isolateMessages) {
+            queue.pushIsolated(message, mode);
+        } else if (index === 0 && messages.length > 1) {
             queue.pushIsolateAndClear(message, mode);
         } else {
             queue.push(message, mode);
@@ -1066,6 +1076,7 @@ describe('codexRemoteLauncher', () => {
         harness.remainingThreadSystemErrors = 0;
         harness.nextThreadSystemErrorMessage = null;
         harness.failNextCompact = false;
+        harness.deferCompactCompletion = false;
         harness.deferThreadStatusNotifications = false;
         harness.emitChildThreadEvents = false;
         harness.emitChildUsageEvents = false;
@@ -2684,6 +2695,45 @@ describe('codexRemoteLauncher', () => {
         });
     });
 
+    it('does not start the next turn until manual compaction finishes', async () => {
+        harness.deferCompactCompletion = true;
+        const { session, sessionEvents } = createSessionStub([
+            'first message',
+            '/compact',
+            'after compact'
+        ], createMode(), true);
+
+        const running = codexRemoteLauncher(session as never);
+        await vi.waitFor(() => {
+            expect(harness.compactThreadIds).toEqual(['thread-1']);
+        });
+
+        expect(harness.startTurnMessages).toEqual(['first message']);
+        expect(sessionEvents).not.toContainEqual({
+            type: 'message',
+            message: 'Compaction completed'
+        });
+
+        harness.dispatchNotification?.('item/completed', {
+            threadId: 'thread-1',
+            turnId: 'compact-1',
+            item: { id: 'compact-item-1', type: 'contextCompaction' }
+        });
+        harness.dispatchNotification?.('turn/completed', {
+            threadId: 'thread-1',
+            turn: { id: 'compact-1', status: 'completed' }
+        });
+
+        const exitReason = await running;
+
+        expect(exitReason).toBe('exit');
+        expect(harness.startTurnMessages).toEqual(['first message', 'after compact']);
+        expect(sessionEvents).toContainEqual({
+            type: 'message',
+            message: 'Compaction completed'
+        });
+    });
+
     it('interrupts an in-flight turn before compacting the current thread', async () => {
         harness.suppressTurnCompletion = true;
         const { session, sessionEvents } = createSessionStub(['first message', '/compact']);

cli/src/codex/codexRemoteLauncher.ts

@@ -1827,6 +1827,16 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
             message: QueuedMessage;
             timeout: ReturnType<typeof setTimeout> | null;
         } | null = null;
+        let manualCompact: {
+            threadId: string;
+            turnId: string | null;
+            compacted: boolean;
+            terminal: { type: 'complete' | 'failed'; turnId: string; error?: string } | null;
+            timeout: ReturnType<typeof setTimeout> | null;
+            abortHandler: (() => void) | null;
+            resolve: () => void;
+            reject: (error: Error) => void;
+        } | null = null;
         let loopWakeWaiter: (() => void) | null = null;
 
         const wakeLoop = () => {
@@ -1930,6 +1940,132 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
                 });
         };
 
+        const clearManualCompact = (compact: typeof manualCompact) => {
+            if (!compact) {
+                return;
+            }
+            if (compact.timeout) {
+                clearTimeout(compact.timeout);
+                compact.timeout = null;
+            }
+            if (compact.abortHandler) {
+                this.abortController.signal.removeEventListener('abort', compact.abortHandler);
+                compact.abortHandler = null;
+            }
+            if (manualCompact === compact) {
+                manualCompact = null;
+            }
+        };
+
+        const settleManualCompact = (
+            compact: typeof manualCompact,
+            error?: Error
+        ) => {
+            if (!compact || manualCompact !== compact) {
+                return;
+            }
+            clearManualCompact(compact);
+            if (error) {
+                compact.reject(error);
+            } else {
+                compact.resolve();
+            }
+        };
+
+        const beginManualCompact = (threadId: string): Promise<void> => {
+            if (manualCompact) {
+                settleManualCompact(manualCompact, new Error('Compaction superseded'));
+            }
+
+            return new Promise<void>((resolve, reject) => {
+                const compact = {
+                    threadId,
+                    turnId: null as string | null,
+                    compacted: false,
+                    terminal: null as { type: 'complete' | 'failed'; turnId: string; error?: string } | null,
+                    timeout: null as ReturnType<typeof setTimeout> | null,
+                    abortHandler: null as (() => void) | null,
+                    resolve,
+                    reject
+                };
+                manualCompact = compact;
+                compact.timeout = setTimeout(() => {
+                    settleManualCompact(compact, new Error('timed out waiting for Codex compaction to finish'));
+                }, SAME_THREAD_COMPACT_TIMEOUT_MS);
+                compact.timeout.unref?.();
+                compact.abortHandler = () => {
+                    settleManualCompact(compact, new Error('compaction interrupted'));
+                };
+                this.abortController.signal.addEventListener('abort', compact.abortHandler, { once: true });
+            });
+        };
+
+        const recordManualCompactStarted = (threadId: string | null, turnId: string | null) => {
+            const compact = manualCompact;
+            if (!compact || !turnId || (threadId && threadId !== compact.threadId)) {
+                return;
+            }
+            compact.turnId ??= turnId;
+        };
+
+        const recordManualCompactCompleted = (
+            threadId: string | null,
+            turnId: string | null,
+            awaitTurnCompletion: boolean
+        ) => {
+            const compact = manualCompact;
+            if (!compact || threadId !== compact.threadId) {
+                return;
+            }
+            if (!awaitTurnCompletion) {
+                settleManualCompact(compact);
+                return;
+            }
+            if (!turnId && !compact.turnId) {
+                settleManualCompact(compact);
+                return;
+            }
+            if (turnId && compact.turnId && turnId !== compact.turnId) {
+                return;
+            }
+            compact.turnId ??= turnId;
+            compact.compacted = true;
+            if (!compact.turnId) {
+                settleManualCompact(compact);
+                return;
+            }
+            if (compact.terminal?.turnId === compact.turnId) {
+                settleManualCompact(
+                    compact,
+                    compact.terminal.type === 'failed'
+                        ? new Error(compact.terminal.error ?? 'Codex compaction failed')
+                        : undefined
+                );
+            }
+        };
+
+        const recordManualCompactTerminal = (
+            type: 'complete' | 'failed',
+            threadId: string | null,
+            turnId: string | null,
+            error?: string
+        ) => {
+            const compact = manualCompact;
+            if (!compact || !turnId || (threadId && threadId !== compact.threadId)) {
+                return;
+            }
+            if (!compact.turnId || turnId !== compact.turnId) {
+                return;
+            }
+            compact.terminal = { type, turnId, ...(error ? { error } : {}) };
+            if (type === 'failed' || compact.compacted) {
+                settleManualCompact(
+                    compact,
+                    type === 'failed' ? new Error(error ?? 'Codex compaction failed') : undefined
+                );
+            }
+        };
+
         const forwardedGoalSignaturesByThreadId = new Map<string, string>();
         const forwardedGoalClearsByThreadId = new Set<string>();
         const adminInterruptedTurnIds = new Set<string>();
@@ -2208,10 +2344,32 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
             }
 
             if (msgType === 'thread_compacted') {
+                recordManualCompactCompleted(
+                    eventThreadId,
+                    eventTurnId,
+                    msg.await_turn_completion === true
+                );
                 completeCompactRecovery(eventThreadId);
                 return;
             }
 
+            if (msgType === 'task_started') {
+                recordManualCompactStarted(eventThreadId ?? this.currentThreadId, eventTurnId);
+            } else if (msgType === 'task_complete') {
+                recordManualCompactTerminal(
+                    'complete',
+                    eventThreadId ?? this.currentThreadId,
+                    eventTurnId
+                );
+            } else if (msgType === 'task_failed' || msgType === 'turn_aborted') {
+                recordManualCompactTerminal(
+                    'failed',
+                    eventThreadId ?? this.currentThreadId,
+                    eventTurnId,
+                    asString(msg.error) ?? (msgType === 'turn_aborted' ? 'Codex compaction was aborted' : undefined)
+                );
+            }
+
             if (eventThreadId && this.currentThreadId && eventThreadId !== this.currentThreadId) {
                 logger.debug(
                     `[Codex] Routing event from non-active thread into agent trace; ` +
@@ -3328,14 +3486,23 @@ class CodexRemoteLauncher extends RemoteLauncherBase {
             }
 
             sendVisibleStatus('Compaction started');
+            const compactCompletion = beginManualCompact(threadId);
+            void compactCompletion.catch(() => {});
             try {
                 await appServerClient.compactThread({ threadId }, {
                     signal: this.abortController.signal
                 });
+                await compactCompletion;
                 sendVisibleStatus('Compaction completed');
             } catch (error) {
                 const detail = error instanceof Error ? error.message : String(error);
                 sendVisibleStatus(`Compaction failed: ${detail}`);
+            } finally {
+                if (manualCompact?.threadId === threadId) {
+                    const compact = manualCompact;
+                    clearManualCompact(compact);
+                    compact.resolve();
+                }
             }
             return true;
         };

cli/src/codex/utils/appServerEventConverter.test.ts

@@ -833,6 +833,29 @@ describe('AppServerEventConverter', () => {
         ]);
     });
 
+    it('maps completed contextCompaction items and preserves the turn boundary', () => {
+        const converter = new AppServerEventConverter();
+        const events = converter.handleNotification('item/completed', {
+            threadId: 'thread-1',
+            turnId: 'turn-compact',
+            item: { id: 'compact-item-1', type: 'contextCompaction' }
+        });
+
+        expect(events).toEqual([
+            {
+                type: 'thread_compacted',
+                thread_id: 'thread-1',
+                turn_id: 'turn-compact',
+                await_turn_completion: true
+            },
+            {
+                type: 'context_compacted',
+                thread_id: 'thread-1',
+                turn_id: 'turn-compact'
+            }
+        ]);
+    });
+
     it('ignores compacted notifications without thread ids', () => {
         const converter = new AppServerEventConverter();
 

cli/src/codex/utils/appServerEventConverter.ts

@@ -866,6 +866,23 @@ export class AppServerEventConverter {
                 return events;
             }
 
+            if (itemType === 'contextcompaction') {
+                if (method === 'item/completed') {
+                    const threadId = asString(eventScope.thread_id);
+                    const turnId = asString(eventScope.turn_id);
+                    if (threadId) {
+                        events.push({
+                            type: 'thread_compacted',
+                            thread_id: threadId,
+                            ...(turnId ? { turn_id: turnId } : {}),
+                            await_turn_completion: true
+                        });
+                        events.push(scoped({ type: 'context_compacted' }));
+                    }
+                }
+                return events;
+            }
+
             if (itemType === 'agentmessage') {
                 if (method === 'item/completed') {
                     if (this.completedAgentMessageItems.has(itemId)) {