From 19e74456fba72cf1108138af8afdcacf7aa16b23 Mon Sep 17 00:00:00 2001 From: Qianli Liu <73509271+Liu-KM@users.noreply.github.com> Date: Thu, 23 Apr 2026 20:45:33 +0800 Subject: [PATCH] fix(codex): surface thread system errors (#519) Co-authored-by: Liu-KM --- cli/src/codex/codexRemoteLauncher.test.ts | 86 ++++++++++++++++--- cli/src/codex/codexRemoteLauncher.ts | 33 +++++-- .../utils/appServerEventConverter.test.ts | 15 ++++ .../codex/utils/appServerEventConverter.ts | 18 ++++ .../codex/utils/terminalEventGuard.test.ts | 26 ++++++ cli/src/codex/utils/terminalEventGuard.ts | 12 +++ 6 files changed, 171 insertions(+), 19 deletions(-) diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index 6d1b2c57..22307e1d 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -5,7 +5,11 @@ import type { EnhancedMode } from './loop'; const harness = vi.hoisted(() => ({ notifications: [] as Array<{ method: string; params: unknown }>, registerRequestCalls: [] as string[], - initializeCalls: [] as unknown[] + initializeCalls: [] as unknown[], + startThreadIds: [] as string[], + resumeThreadIds: [] as string[], + startTurnThreadIds: [] as string[], + remainingThreadSystemErrors: 0 })); vi.mock('./codexAppServerClient', () => { @@ -28,23 +32,41 @@ vi.mock('./codexAppServerClient', () => { } async startThread(): Promise<{ thread: { id: string }; model: string }> { - return { thread: { id: 'thread-anonymous' }, model: 'gpt-5.4' }; + const id = `thread-${harness.startThreadIds.length + 1}`; + harness.startThreadIds.push(id); + return { thread: { id }, model: 'gpt-5.4' }; } - async resumeThread(): Promise<{ thread: { id: string }; model: string }> { - return { thread: { id: 'thread-anonymous' }, model: 'gpt-5.4' }; + async resumeThread(params?: { threadId?: string }): Promise<{ thread: { id: string }; model: string }> { + const id = params?.threadId ?? 'thread-resumed'; + harness.resumeThreadIds.push(id); + return { thread: { id }, model: 'gpt-5.4' }; } - async startTurn(): Promise<{ turn: Record }> { - const started = { turn: {} }; + async startTurn(params?: { threadId?: string }): Promise<{ turn: { id?: string } }> { + const threadId = params?.threadId ?? 'thread-unknown'; + harness.startTurnThreadIds.push(threadId); + const turnId = `turn-${harness.startTurnThreadIds.length}`; + const started = { turn: { id: turnId } }; harness.notifications.push({ method: 'turn/started', params: started }); this.notificationHandler?.('turn/started', started); - const completed = { status: 'Completed', turn: {} }; + if (harness.remainingThreadSystemErrors > 0) { + harness.remainingThreadSystemErrors -= 1; + const failed = { + thread: { id: threadId }, + status: { type: 'systemError' } + }; + harness.notifications.push({ method: 'thread/status/changed', params: failed }); + this.notificationHandler?.('thread/status/changed', failed); + return { turn: { id: turnId } }; + } + + const completed = { status: 'Completed', turn: { id: turnId } }; harness.notifications.push({ method: 'turn/completed', params: completed }); this.notificationHandler?.('turn/completed', completed); - return { turn: {} }; + return { turn: { id: turnId } }; } async interruptTurn(): Promise> { @@ -80,9 +102,15 @@ function createMode(): EnhancedMode { }; } -function createSessionStub() { +function createSessionStub(messages = ['hello from launcher test']) { const queue = new MessageQueue2((mode) => JSON.stringify(mode)); - queue.push('hello from launcher test', createMode()); + messages.forEach((message, index) => { + if (index === 0 && messages.length > 1) { + queue.pushIsolateAndClear(message, createMode()); + } else { + queue.push(message, createMode()); + } + }); queue.close(); const sessionEvents: Array<{ type: string; [key: string]: unknown }> = []; @@ -168,9 +196,13 @@ describe('codexRemoteLauncher', () => { harness.notifications = []; harness.registerRequestCalls = []; harness.initializeCalls = []; + harness.startThreadIds = []; + harness.resumeThreadIds = []; + harness.startTurnThreadIds = []; + harness.remainingThreadSystemErrors = 0; }); - it('finishes a turn and emits ready when task lifecycle events omit turn_id', async () => { + it('finishes a turn and emits ready when task lifecycle events include turn_id', async () => { const { session, sessionEvents, @@ -182,7 +214,7 @@ describe('codexRemoteLauncher', () => { const exitReason = await codexRemoteLauncher(session as never); expect(exitReason).toBe('exit'); - expect(foundSessionIds).toContain('thread-anonymous'); + expect(foundSessionIds).toContain('thread-1'); expect(getModel()).toBe('gpt-5.4'); expect(harness.initializeCalls).toEqual([{ clientInfo: { @@ -198,4 +230,34 @@ describe('codexRemoteLauncher', () => { expect(thinkingChanges).toContain(true); expect(session.thinking).toBe(false); }); + + it('surfaces thread-level systemError as a visible failure and emits ready', async () => { + harness.remainingThreadSystemErrors = 1; + const { session, sessionEvents } = createSessionStub(); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.notifications.map((entry) => entry.method)).toEqual(['turn/started', 'thread/status/changed']); + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Task failed: Codex thread entered systemError' + }); + expect(sessionEvents.filter((event) => event.type === 'ready').length).toBeGreaterThanOrEqual(1); + expect(session.thinking).toBe(false); + }); + + it('starts a fresh thread for the next queued message after thread-level systemError', async () => { + harness.remainingThreadSystemErrors = 1; + const { session } = createSessionStub(['first message', 'second message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startThreadIds).toEqual(['thread-1', 'thread-2']); + expect(harness.resumeThreadIds).toEqual([]); + expect(harness.startTurnThreadIds).toEqual(['thread-1', 'thread-2']); + expect(session.sessionId).toBe('thread-2'); + expect(session.thinking).toBe(false); + }); }); diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index 53024dee..8251a2a3 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -241,11 +241,13 @@ class CodexRemoteLauncher extends RemoteLauncherBase { let clearReadyAfterTurnTimer: (() => void) | null = null; let turnInFlight = false; let allowAnonymousTerminalEvent = false; + let invalidThreadId: string | null = null; const handleCodexEvent = (msg: Record) => { const msgType = asString(msg.type); if (!msgType) return; const eventTurnId = asString(msg.turn_id ?? msg.turnId); + const eventThreadId = asString(msg.thread_id ?? msg.threadId); const isTerminalEvent = msgType === 'task_complete' || msgType === 'turn_aborted' || msgType === 'task_failed'; if (msgType === 'thread_started') { @@ -267,22 +269,33 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } } + const isThreadStatusFailure = msgType === 'task_failed' && msg.terminal_source === 'thread_status'; + if (isTerminalEvent) { if (shouldIgnoreTerminalEvent({ eventTurnId, currentTurnId: this.currentTurnId, turnInFlight, - allowAnonymousTerminalEvent + allowAnonymousTerminalEvent, + eventThreadId, + currentThreadId: this.currentThreadId, + allowMatchingThreadIdTerminalEvent: msg.terminal_source === 'thread_status' })) { logger.debug( `[Codex] Ignoring terminal event ${msgType} without matching turn context; ` + `eventTurnId=${eventTurnId ?? 'none'}, activeTurn=${this.currentTurnId ?? 'none'}, ` + + `eventThreadId=${eventThreadId ?? 'none'}, activeThread=${this.currentThreadId ?? 'none'}, ` + `turnInFlight=${turnInFlight}, allowAnonymous=${allowAnonymousTerminalEvent}` ); return; } this.currentTurnId = null; allowAnonymousTerminalEvent = false; + if (isThreadStatusFailure) { + invalidThreadId = eventThreadId ?? this.currentThreadId; + this.currentThreadId = null; + hasThread = false; + } } if (msgType === 'agent_message') { @@ -314,7 +327,9 @@ class CodexRemoteLauncher extends RemoteLauncherBase { messageBuffer.addMessage('Turn aborted', 'status'); } else if (msgType === 'task_failed') { const error = asString(msg.error); - messageBuffer.addMessage(error ? `Task failed: ${error}` : 'Task failed', 'status'); + const message = error ? `Task failed: ${error}` : 'Task failed'; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); } if (msgType === 'task_started') { @@ -627,7 +642,9 @@ class CodexRemoteLauncher extends RemoteLauncherBase { cliOverrides: session.codexCliOverrides }); - const resumeCandidate = session.sessionId; + const resumeCandidate = session.sessionId && session.sessionId !== invalidThreadId + ? session.sessionId + : null; let threadId: string | null = null; if (resumeCandidate) { @@ -695,10 +712,12 @@ class CodexRemoteLauncher extends RemoteLauncherBase { const turnRecord = asRecord(turnResponse); const turn = turnRecord ? asRecord(turnRecord.turn) : null; const turnId = asString(turn?.id); - if (turnId) { - this.currentTurnId = turnId; - } else if (!this.currentTurnId) { - allowAnonymousTerminalEvent = true; + if (turnInFlight) { + if (turnId) { + this.currentTurnId = turnId; + } else if (!this.currentTurnId) { + allowAnonymousTerminalEvent = true; + } } } catch (error) { logger.warn('Error in codex session:', error); diff --git a/cli/src/codex/utils/appServerEventConverter.test.ts b/cli/src/codex/utils/appServerEventConverter.test.ts index 27276926..68dd3407 100644 --- a/cli/src/codex/utils/appServerEventConverter.test.ts +++ b/cli/src/codex/utils/appServerEventConverter.test.ts @@ -16,6 +16,21 @@ describe('AppServerEventConverter', () => { expect(events).toEqual([{ type: 'thread_started', thread_id: 'thread-2' }]); }); + it('maps thread systemError to a task failure', () => { + const converter = new AppServerEventConverter(); + const events = converter.handleNotification('thread/status/changed', { + thread: { id: 'thread-1' }, + status: { type: 'systemError' } + }); + + expect(events).toEqual([{ + type: 'task_failed', + thread_id: 'thread-1', + terminal_source: 'thread_status', + error: 'Codex thread entered systemError' + }]); + }); + it('maps turn/started and completed statuses', () => { const converter = new AppServerEventConverter(); diff --git a/cli/src/codex/utils/appServerEventConverter.ts b/cli/src/codex/utils/appServerEventConverter.ts index 08a95763..667f9272 100644 --- a/cli/src/codex/utils/appServerEventConverter.ts +++ b/cli/src/codex/utils/appServerEventConverter.ts @@ -266,6 +266,24 @@ export class AppServerEventConverter { return events; } + if (method === 'thread/status/changed') { + const thread = asRecord(paramsRecord.thread) ?? paramsRecord; + const threadId = asString(thread.threadId ?? thread.thread_id ?? thread.id); + const status = asRecord(paramsRecord.status ?? thread.status); + const statusType = asString(status?.type ?? paramsRecord.statusType ?? paramsRecord.status_type); + if (statusType === 'systemError') { + const error = asString(status?.message ?? status?.error ?? paramsRecord.message ?? paramsRecord.error) + ?? 'Codex thread entered systemError'; + events.push({ + type: 'task_failed', + ...(threadId ? { thread_id: threadId } : {}), + terminal_source: 'thread_status', + error + }); + } + return events; + } + if (method === 'turn/started') { const turn = asRecord(paramsRecord.turn) ?? paramsRecord; const turnId = asString(turn.turnId ?? turn.turn_id ?? turn.id); diff --git a/cli/src/codex/utils/terminalEventGuard.test.ts b/cli/src/codex/utils/terminalEventGuard.test.ts index 5896b11e..5c1e01d4 100644 --- a/cli/src/codex/utils/terminalEventGuard.test.ts +++ b/cli/src/codex/utils/terminalEventGuard.test.ts @@ -44,6 +44,32 @@ describe('shouldIgnoreTerminalEvent', () => { expect(ignored).toBe(true); }); + it('accepts thread-level terminal events for the current thread when explicitly allowed', () => { + const ignored = shouldIgnoreTerminalEvent({ + eventTurnId: null, + currentTurnId: 'turn-1', + turnInFlight: true, + eventThreadId: 'thread-1', + currentThreadId: 'thread-1', + allowMatchingThreadIdTerminalEvent: true + }); + + expect(ignored).toBe(false); + }); + + it('ignores thread-level terminal events for a different thread', () => { + const ignored = shouldIgnoreTerminalEvent({ + eventTurnId: null, + currentTurnId: 'turn-1', + turnInFlight: true, + eventThreadId: 'thread-old', + currentThreadId: 'thread-1', + allowMatchingThreadIdTerminalEvent: true + }); + + expect(ignored).toBe(true); + }); + it('ignores stale terminal events from another turn', () => { const ignored = shouldIgnoreTerminalEvent({ eventTurnId: 'turn-old', diff --git a/cli/src/codex/utils/terminalEventGuard.ts b/cli/src/codex/utils/terminalEventGuard.ts index a4b76b01..e526bec5 100644 --- a/cli/src/codex/utils/terminalEventGuard.ts +++ b/cli/src/codex/utils/terminalEventGuard.ts @@ -3,6 +3,9 @@ export type TerminalEventGuardInput = { currentTurnId: string | null; turnInFlight: boolean; allowAnonymousTerminalEvent?: boolean; + eventThreadId?: string | null; + currentThreadId?: string | null; + allowMatchingThreadIdTerminalEvent?: boolean; }; export function shouldIgnoreTerminalEvent(input: TerminalEventGuardInput): boolean { @@ -13,6 +16,15 @@ export function shouldIgnoreTerminalEvent(input: TerminalEventGuardInput): boole } if (input.currentTurnId) { + const allowMatchingThreadIdTerminalEvent = input.allowMatchingThreadIdTerminalEvent === true; + if ( + allowMatchingThreadIdTerminalEvent && + input.eventThreadId && + input.currentThreadId && + input.eventThreadId === input.currentThreadId + ) { + return false; + } return true; }