diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index f389b4f2..a179bd53 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/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((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']); diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index 5e35c376..d39b5650 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -1827,6 +1827,16 @@ class CodexRemoteLauncher extends RemoteLauncherBase { message: QueuedMessage; timeout: ReturnType | null; } | null = null; + let manualCompact: { + threadId: string; + turnId: string | null; + compacted: boolean; + terminal: { type: 'complete' | 'failed'; turnId: string; error?: string } | null; + timeout: ReturnType | 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 => { + if (manualCompact) { + settleManualCompact(manualCompact, new Error('Compaction superseded')); + } + + return new Promise((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 | 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(); const forwardedGoalClearsByThreadId = new Set(); const adminInterruptedTurnIds = new Set(); @@ -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; }; diff --git a/cli/src/codex/utils/appServerEventConverter.test.ts b/cli/src/codex/utils/appServerEventConverter.test.ts index 62f68489..31e4846e 100644 --- a/cli/src/codex/utils/appServerEventConverter.test.ts +++ b/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(); diff --git a/cli/src/codex/utils/appServerEventConverter.ts b/cli/src/codex/utils/appServerEventConverter.ts index 1944a723..e83253c5 100644 --- a/cli/src/codex/utils/appServerEventConverter.ts +++ b/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)) {