From 00e8fc3a47b901184090031628d3c3859e3c7b10 Mon Sep 17 00:00:00 2001 From: SSU-WEI HUANG Date: Tue, 4 Aug 2026 10:58:49 +0800 Subject: [PATCH] fix(codex): recover ready after stale terminal event (#997) * fix codex stale terminal recovery * fix(codex): ignore stale retry failures * fix(codex): ignore stale retry terminal failures Only task completion may bypass stale-turn duplicate handling during same-thread recovery, preventing delayed failed events from finalizing the active retry. Co-authored-by: Cursor * fix(codex): separate stale turn recovery guard Limit matching-thread status events to missing turn IDs so delayed status failures cannot affect an active retry turn. Co-authored-by: Cursor * fix(codex): scope stale completion recovery turn Accept a stale completion only for the immediately finalized turn, so older retries cannot finalize the active turn. Co-authored-by: Cursor --------- Co-authored-by: Cursor --- cli/src/codex/codexRemoteLauncher.test.ts | 154 ++++++++++++++++++ cli/src/codex/codexRemoteLauncher.ts | 20 ++- .../codex/utils/terminalEventGuard.test.ts | 39 +++++ cli/src/codex/utils/terminalEventGuard.ts | 20 ++- 4 files changed, 226 insertions(+), 7 deletions(-) diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index 36bffd99..ef09648a 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -57,6 +57,10 @@ const harness = vi.hoisted(() => ({ failNextCompact: false, deferCompactCompletion: false, deferThreadStatusNotifications: false, + emitStaleTaskCompleteAfterRetry: false, + emitStaleTaskFailedAfterRetry: false, + emitStaleThreadStatusFailureAfterRetry: false, + emitFirstTurnTaskCompleteAfterSecondRetry: false, emitChildThreadEvents: false, emitChildUsageEvents: false, emitChildGoalEvent: false, @@ -907,6 +911,82 @@ vi.mock('./codexAppServerClient', () => { } } + if (harness.emitStaleTaskCompleteAfterRetry && harness.startTurnThreadIds.length === 2) { + const assistantMessage = { + item: { + id: 'stale-retry-message', + type: 'agentMessage', + content: [{ type: 'text', text: 'done after retry' }] + }, + threadId, + turnId: 'turn-1' + }; + harness.notifications.push({ method: 'item/completed', params: assistantMessage }); + this.notificationHandler?.('item/completed', assistantMessage); + + const usage = { + tokenUsage: { + thread_id: threadId, + turn_id: 'turn-1', + last_token_usage: { + input_tokens: 10, + output_tokens: 2 + }, + model_context_window: 200_000 + } + }; + harness.notifications.push({ method: 'thread/tokenUsage/updated', params: usage }); + this.notificationHandler?.('thread/tokenUsage/updated', usage); + + const staleCompleted = { + msg: { + type: 'task_complete', + thread_id: threadId, + turn_id: 'turn-1' + } + }; + harness.notifications.push({ method: 'codex/event/task_complete', params: staleCompleted }); + this.notificationHandler?.('codex/event/task_complete', staleCompleted); + return { turn: { id: turnId } }; + } + + if (harness.emitStaleTaskFailedAfterRetry && harness.startTurnThreadIds.length === 2) { + const staleFailed = { + msg: { + type: 'task_failed', + thread_id: threadId, + turn_id: 'turn-1', + error: 'Codex thread entered systemError' + } + }; + harness.notifications.push({ method: 'codex/event/task_failed', params: staleFailed }); + this.notificationHandler?.('codex/event/task_failed', staleFailed); + } + + if (harness.emitStaleThreadStatusFailureAfterRetry && harness.startTurnThreadIds.length === 2) { + const staleThreadStatus = { + thread: { id: threadId }, + turnId: 'turn-1', + status: { type: 'systemError' } + }; + harness.notifications.push({ method: 'thread/status/changed', params: staleThreadStatus }); + this.notificationHandler?.('thread/status/changed', staleThreadStatus); + await new Promise((resolve) => setTimeout(resolve, 300)); + } + + if (harness.emitFirstTurnTaskCompleteAfterSecondRetry && harness.startTurnThreadIds.length === 3) { + const staleCompleted = { + msg: { + type: 'task_complete', + thread_id: threadId, + turn_id: 'turn-1' + } + }; + harness.notifications.push({ method: 'codex/event/task_complete', params: staleCompleted }); + this.notificationHandler?.('codex/event/task_complete', staleCompleted); + await new Promise((resolve) => setTimeout(resolve, 300)); + } + const completed = { status: 'Completed', turn: { id: turnId } }; harness.notifications.push({ method: 'turn/completed', params: completed }); this.notificationHandler?.('turn/completed', completed); @@ -1161,6 +1241,10 @@ describe('codexRemoteLauncher', () => { harness.failNextCompact = false; harness.deferCompactCompletion = false; harness.deferThreadStatusNotifications = false; + harness.emitStaleTaskCompleteAfterRetry = false; + harness.emitStaleTaskFailedAfterRetry = false; + harness.emitStaleThreadStatusFailureAfterRetry = false; + harness.emitFirstTurnTaskCompleteAfterSecondRetry = false; harness.emitChildThreadEvents = false; harness.emitChildUsageEvents = false; harness.emitChildGoalEvent = false; @@ -1650,6 +1734,76 @@ describe('codexRemoteLauncher', () => { expect(session.thinking).toBe(false); }); + it('emits ready when same-thread retry completes with a stale terminal turn id', async () => { + harness.remainingThreadSystemErrors = 1; + harness.emitStaleTaskCompleteAfterRetry = true; + const { + session, + sessionEvents, + codexMessages, + rpcHandlers + } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + const timeout = new Promise<'timeout'>((resolve) => { + setTimeout(() => resolve('timeout'), 500); + }); + const result = await Promise.race([running, timeout]); + if (result === 'timeout') { + await rpcHandlers.get('switch')?.({}); + await running; + } + + expect(result).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1']); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1']); + expect(harness.startTurnMessages).toEqual(['first message', 'first message']); + expect(codexMessages).toContainEqual(expect.objectContaining({ + type: 'message', + message: 'done after retry' + })); + expect(sessionEvents.filter((event) => event.type === 'ready').length).toBeGreaterThanOrEqual(1); + expect(session.thinking).toBe(false); + }); + + it('ignores a stale same-thread failure after retry has started', async () => { + harness.remainingThreadSystemErrors = 1; + harness.emitStaleTaskFailedAfterRetry = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1']); + expect(sessionEvents.filter((event) => event.type === 'ready').length).toBeGreaterThanOrEqual(1); + expect(session.thinking).toBe(false); + }); + + it('ignores a stale thread-status failure after retry has started', async () => { + harness.remainingThreadSystemErrors = 1; + harness.emitStaleThreadStatusFailureAfterRetry = true; + const { session } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1']); + expect(session.thinking).toBe(false); + }); + + it('ignores a first-turn completion while the second retry is running', async () => { + harness.remainingThreadSystemErrors = 2; + harness.emitFirstTurnTaskCompleteAfterSecondRetry = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-1', 'thread-1']); + expect(sessionEvents.filter((event) => event.type === 'ready')).toHaveLength(1); + expect(session.thinking).toBe(false); + }); + it('still retries a generic systemError when an empty failed turn completion confirms it', async () => { harness.remainingThreadSystemErrors = 1; harness.emitFailedCompletionAfterThreadSystemError = true; diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index 3cbc12f5..9436db75 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -2457,7 +2457,22 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return; } - if (isTerminalEvent && eventTurnId && eventTurnId === lastFinalizedTurnId) { + const isStaleSameThreadRecoveryTerminal = msgType === 'task_complete' + && turnInFlight + && (sameThreadRetryAttempt > 0 || sameThreadCompactAttempt > 0) + && Boolean(eventTurnId) + && eventTurnId === lastFinalizedTurnId + && Boolean(this.currentTurnId) + && eventTurnId !== this.currentTurnId + && Boolean(eventThreadId) + && eventThreadId === this.currentThreadId; + + if ( + isTerminalEvent + && eventTurnId + && eventTurnId === lastFinalizedTurnId + && !isStaleSameThreadRecoveryTerminal + ) { logger.debug(`[Codex] Ignoring duplicate terminal event for turn ${eventTurnId}`); return; } @@ -2643,7 +2658,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { allowAnonymousTerminalEvent, eventThreadId, currentThreadId: this.currentThreadId, - allowMatchingThreadIdTerminalEvent: msg.terminal_source === 'thread_status' + allowMatchingThreadIdTerminalEvent: msg.terminal_source === 'thread_status', + allowMismatchedTurnIdTerminalEvent: isStaleSameThreadRecoveryTerminal })) { logger.debug( `[Codex] Ignoring terminal event ${msgType} without matching turn context; ` + diff --git a/cli/src/codex/utils/terminalEventGuard.test.ts b/cli/src/codex/utils/terminalEventGuard.test.ts index 5c1e01d4..37cdc78e 100644 --- a/cli/src/codex/utils/terminalEventGuard.test.ts +++ b/cli/src/codex/utils/terminalEventGuard.test.ts @@ -80,6 +80,45 @@ describe('shouldIgnoreTerminalEvent', () => { expect(ignored).toBe(true); }); + it('accepts stale-turn terminal events for the current thread when explicitly allowed', () => { + const ignored = shouldIgnoreTerminalEvent({ + eventTurnId: 'turn-old', + currentTurnId: 'turn-current', + turnInFlight: true, + eventThreadId: 'thread-1', + currentThreadId: 'thread-1', + allowMismatchedTurnIdTerminalEvent: true + }); + + expect(ignored).toBe(false); + }); + + it('ignores stale-turn terminal events when only thread-level events are allowed', () => { + const ignored = shouldIgnoreTerminalEvent({ + eventTurnId: 'turn-old', + currentTurnId: 'turn-current', + turnInFlight: true, + eventThreadId: 'thread-1', + currentThreadId: 'thread-1', + allowMatchingThreadIdTerminalEvent: true + }); + + expect(ignored).toBe(true); + }); + + it('still ignores stale-turn terminal events for a different thread when thread matching is allowed', () => { + const ignored = shouldIgnoreTerminalEvent({ + eventTurnId: 'turn-old', + currentTurnId: 'turn-current', + turnInFlight: true, + eventThreadId: 'thread-old', + currentThreadId: 'thread-1', + allowMismatchedTurnIdTerminalEvent: true + }); + + expect(ignored).toBe(true); + }); + it('accepts terminal events that match the current turn id', () => { const ignored = shouldIgnoreTerminalEvent({ eventTurnId: 'turn-current', diff --git a/cli/src/codex/utils/terminalEventGuard.ts b/cli/src/codex/utils/terminalEventGuard.ts index e526bec5..7879393c 100644 --- a/cli/src/codex/utils/terminalEventGuard.ts +++ b/cli/src/codex/utils/terminalEventGuard.ts @@ -6,22 +6,32 @@ export type TerminalEventGuardInput = { eventThreadId?: string | null; currentThreadId?: string | null; allowMatchingThreadIdTerminalEvent?: boolean; + allowMismatchedTurnIdTerminalEvent?: boolean; }; export function shouldIgnoreTerminalEvent(input: TerminalEventGuardInput): boolean { const allowAnonymousTerminalEvent = input.allowAnonymousTerminalEvent === true; + const allowMatchingThreadIdTerminalEvent = input.allowMatchingThreadIdTerminalEvent === true; + const hasMatchingThreadId = Boolean( + input.eventThreadId && + input.currentThreadId && + input.eventThreadId === input.currentThreadId + ); if (input.eventTurnId) { - return Boolean(input.currentTurnId && input.eventTurnId !== input.currentTurnId); + if (!input.currentTurnId || input.eventTurnId === input.currentTurnId) { + return false; + } + return !( + input.allowMismatchedTurnIdTerminalEvent === true + && hasMatchingThreadId + ); } if (input.currentTurnId) { - const allowMatchingThreadIdTerminalEvent = input.allowMatchingThreadIdTerminalEvent === true; if ( allowMatchingThreadIdTerminalEvent && - input.eventThreadId && - input.currentThreadId && - input.eventThreadId === input.currentThreadId + hasMatchingThreadId ) { return false; }