diff --git a/cli/src/codex/appServerTypes.ts b/cli/src/codex/appServerTypes.ts index ffebf0e6..5fc0262c 100644 --- a/cli/src/codex/appServerTypes.ts +++ b/cli/src/codex/appServerTypes.ts @@ -204,6 +204,19 @@ export interface TurnInterruptResponse { [key: string]: unknown; } +export interface ThreadRollbackParams { + threadId: string; + numTurns: number; +} + +export interface ThreadRollbackResponse { + thread: { + id: string; + [key: string]: unknown; + }; + [key: string]: unknown; +} + export interface ThreadCompactStartParams { threadId: string; } diff --git a/cli/src/codex/codexAppServerClient.ts b/cli/src/codex/codexAppServerClient.ts index 0aa2e72b..31569e13 100644 --- a/cli/src/codex/codexAppServerClient.ts +++ b/cli/src/codex/codexAppServerClient.ts @@ -16,6 +16,8 @@ import type { TurnStartResponse, TurnInterruptParams, TurnInterruptResponse, + ThreadRollbackParams, + ThreadRollbackResponse, ThreadCompactStartParams, ThreadCompactStartResponse, ThreadGoalSetParams, @@ -207,6 +209,18 @@ export class CodexAppServerClient extends JsonLineParser { return response as TurnInterruptResponse; } + /** + * Deprecated upstream, but still required to match Codex's native + * safety-buffering retry flow. Keep the protocol call isolated here so it + * can be replaced when app-server exposes a successor. + */ + async rollbackThread(params: ThreadRollbackParams): Promise { + const response = await this.sendRequest('thread/rollback', params, { + timeoutMs: 30_000 + }); + return response as ThreadRollbackResponse; + } + async compactThread( params: ThreadCompactStartParams, options?: { signal?: AbortSignal } diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index 28e931a1..731b030d 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -4,6 +4,7 @@ import type { EnhancedMode } from './loop'; const harness = vi.hoisted(() => ({ notifications: [] as Array<{ method: string; params: unknown }>, + dispatchNotification: null as ((method: string, params: unknown) => void) | null, registerRequestCalls: [] as string[], requestHandlers: new Map Promise | unknown>(), initializeCalls: [] as unknown[], @@ -18,6 +19,9 @@ const harness = vi.hoisted(() => ({ startTurnParams: [] as Array>, startTurnErrors: [] as Error[], interruptedTurns: [] as Array<{ threadId: string; turnId: string }>, + interruptErrors: [] as Error[], + rollbackCalls: [] as Array<{ threadId: string; numTurns: number }>, + rollbackErrors: [] as Error[], compactThreadIds: [] as string[], goalSetCalls: [] as unknown[], goalGetCalls: [] as unknown[], @@ -26,6 +30,11 @@ const harness = vi.hoisted(() => ({ suppressGoalNotifications: false, suppressTurnCompletion: false, remainingThreadSystemErrors: 0, + emitFailedCompletionAfterThreadSystemError: false, + emitCyberPolicyAfterThreadSystemError: false, + emitSafetyBuffering: false, + safetyBufferingFasterModel: null as string | null, + emitModelSafetyNotices: false, startTurnMessages: [] as string[], failResumeThreadIds: [] as string[], nextThreadSystemErrorMessage: null as string | null, @@ -72,6 +81,7 @@ vi.mock('./codexAppServerClient', () => { setNotificationHandler(handler: ((method: string, params: unknown) => void) | null): void { this.notificationHandler = handler; + harness.dispatchNotification = handler; } setStderrHandler(handler: ((text: string) => void) | null): void { @@ -194,9 +204,75 @@ vi.mock('./codexAppServerClient', () => { } else { notify(); } + if (harness.emitCyberPolicyAfterThreadSystemError) { + const policyError = { + threadId, + turnId, + error: { + message: 'This content was flagged for possible cybersecurity risk.', + codexErrorInfo: 'cyberPolicy' + }, + willRetry: false + }; + harness.notifications.push({ method: 'error', params: policyError }); + this.notificationHandler?.('error', policyError); + + const completed = { + threadId, + turnId, + turn: { id: turnId, status: 'failed' } + }; + harness.notifications.push({ method: 'turn/completed', params: completed }); + this.notificationHandler?.('turn/completed', completed); + } else if (harness.emitFailedCompletionAfterThreadSystemError) { + const completed = { + threadId, + turnId, + turn: { id: turnId, status: 'failed' } + }; + harness.notifications.push({ method: 'turn/completed', params: completed }); + this.notificationHandler?.('turn/completed', completed); + } return { turn: { id: turnId } }; } + if (harness.emitSafetyBuffering) { + harness.emitSafetyBuffering = false; + const notification = { + threadId, + turnId, + model: 'gpt-5.4', + useCases: ['cyber'], + reasons: ['review'], + showBufferingUi: true, + fasterModel: harness.safetyBufferingFasterModel + }; + harness.notifications.push({ method: 'model/safetyBuffering/updated', params: notification }); + this.notificationHandler?.('model/safetyBuffering/updated', notification); + return { turn: { id: turnId } }; + } + + if (harness.emitModelSafetyNotices) { + harness.emitModelSafetyNotices = false; + const rerouted = { + threadId, + turnId, + fromModel: 'gpt-5.4', + toModel: 'gpt-5.4-codex', + reason: 'highRiskCyberActivity' + }; + harness.notifications.push({ method: 'model/rerouted', params: rerouted }); + this.notificationHandler?.('model/rerouted', rerouted); + + const verification = { + threadId, + turnId, + verifications: ['trustedAccessForCyber'] + }; + harness.notifications.push({ method: 'model/verification', params: verification }); + this.notificationHandler?.('model/verification', verification); + } + if ( harness.emitRunningChildTurnBeforeSuppressedParent || harness.emitCompletedChildTurnBeforeSuppressedParent @@ -772,6 +848,10 @@ vi.mock('./codexAppServerClient', () => { const threadId = params?.threadId ?? 'thread-unknown'; const turnId = params?.turnId ?? 'turn-unknown'; harness.interruptedTurns.push({ threadId, turnId }); + const error = harness.interruptErrors.shift(); + if (error) { + throw error; + } if (harness.emitTurnAbortedOnInterrupt) { const interrupted = { threadId, @@ -785,6 +865,16 @@ vi.mock('./codexAppServerClient', () => { return {}; } + async rollbackThread(params?: { threadId?: string; numTurns?: number }): Promise<{ thread: { id: string } }> { + const threadId = params?.threadId ?? 'thread-unknown'; + harness.rollbackCalls.push({ threadId, numTurns: params?.numTurns ?? 0 }); + const error = harness.rollbackErrors.shift(); + if (error) { + throw error; + } + return { thread: { id: threadId } }; + } + async disconnect(): Promise {} } @@ -838,6 +928,7 @@ function createSessionStub(messages = ['hello from launcher test'], mode = creat const collaborationModes: Array = []; let currentPermissionMode: EnhancedMode['permissionMode'] = mode.permissionMode; let currentModel: string | null | undefined = mode.model; + let currentModelReasoningEffort = mode.modelReasoningEffort; let currentCollaborationMode: EnhancedMode['collaborationMode'] | undefined = mode.collaborationMode; let agentState: FakeAgentState = { requests: {}, @@ -884,6 +975,9 @@ function createSessionStub(messages = ['hello from launcher test'], mode = creat getModel() { return currentModel; }, + setModelReasoningEffort(nextEffort: EnhancedMode['modelReasoningEffort']) { + currentModelReasoningEffort = nextEffort; + }, getCollaborationMode() { return currentCollaborationMode; }, @@ -927,6 +1021,7 @@ function createSessionStub(messages = ['hello from launcher test'], mode = creat currentPermissionMode = nextMode; }, getModel: () => currentModel, + getModelReasoningEffort: () => currentModelReasoningEffort, getCollaborationMode: () => currentCollaborationMode, collaborationModes, getAgentState: () => agentState @@ -936,6 +1031,7 @@ function createSessionStub(messages = ['hello from launcher test'], mode = creat describe('codexRemoteLauncher', () => { afterEach(() => { harness.notifications = []; + harness.dispatchNotification = null; harness.registerRequestCalls = []; harness.requestHandlers = new Map(); harness.initializeCalls = []; @@ -950,6 +1046,9 @@ describe('codexRemoteLauncher', () => { harness.startTurnParams = []; harness.startTurnErrors = []; harness.interruptedTurns = []; + harness.interruptErrors = []; + harness.rollbackCalls = []; + harness.rollbackErrors = []; harness.compactThreadIds = []; harness.goalSetCalls = []; harness.goalGetCalls = []; @@ -957,6 +1056,11 @@ describe('codexRemoteLauncher', () => { harness.goal = null; harness.suppressGoalNotifications = false; harness.suppressTurnCompletion = false; + harness.emitFailedCompletionAfterThreadSystemError = false; + harness.emitCyberPolicyAfterThreadSystemError = false; + harness.emitSafetyBuffering = false; + harness.safetyBufferingFasterModel = null; + harness.emitModelSafetyNotices = false; harness.startTurnMessages = []; harness.failResumeThreadIds = []; harness.remainingThreadSystemErrors = 0; @@ -1376,6 +1480,373 @@ describe('codexRemoteLauncher', () => { 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; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startTurnMessages).toEqual(['first message', 'first message']); + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Task failed: Codex thread entered systemError; retrying same conversation (1/3)' + }); + expect(session.thinking).toBe(false); + }); + + it('does not retry when a generic systemError is followed by a cyber-policy block', async () => { + harness.remainingThreadSystemErrors = 1; + harness.emitCyberPolicyAfterThreadSystemError = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const exitReason = await codexRemoteLauncher(session as never); + + expect(exitReason).toBe('exit'); + expect(harness.startTurnThreadIds).toEqual(['thread-1']); + expect(harness.startTurnMessages).toEqual(['first message']); + const failureMessages = sessionEvents.filter((event) => event.type === 'message'); + expect(failureMessages).toHaveLength(1); + expect(failureMessages[0]?.message).toContain('This content was flagged for possible cybersecurity risk.'); + expect(failureMessages[0]?.message).toContain('https://openai.com/form/enterprise-trusted-access-for-cyber/'); + expect(failureMessages[0]?.message).toContain('https://help.openai.com/en/articles/20001326'); + expect(sessionEvents.filter((event) => event.type === 'ready').length).toBeGreaterThanOrEqual(1); + expect(session.thinking).toBe(false); + }); + + it('does not retry an explicitly non-retryable error even when its text is retryable', async () => { + harness.suppressTurnCompletion = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(harness.startTurnMessages).toEqual(['first message']); + }); + + harness.dispatchNotification?.('error', { + threadId: 'thread-1', + turnId: 'turn-1', + error: { message: 'Selected model is at capacity' }, + willRetry: false + }); + + await expect(running).resolves.toBe('exit'); + expect(harness.startTurnMessages).toEqual(['first message']); + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Task failed: Selected model is at capacity' + }); + expect(sessionEvents.some((event) => String(event.message ?? '').includes('retrying same conversation'))).toBe(false); + expect(session.thinking).toBe(false); + }); + + it('retries a safety-buffered turn with the offered faster model only after user opt-in', async () => { + harness.emitSafetyBuffering = true; + harness.safetyBufferingFasterModel = 'gpt-5.4-mini'; + harness.emitTurnAbortedOnInterrupt = true; + const { + session, + rpcHandlers, + getAgentState, + getModel, + getModelReasoningEffort + } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + + await vi.waitFor(() => { + expect(Object.values(getAgentState().requests)).toContainEqual(expect.objectContaining({ + tool: 'request_user_input' + })); + }); + expect(harness.startTurnThreadIds).toEqual(['thread-1']); + expect(harness.interruptedTurns).toEqual([]); + expect(harness.rollbackCalls).toEqual([]); + + const requestId = Object.keys(getAgentState().requests)[0]; + await rpcHandlers.get('permission')?.({ + id: requestId, + approved: true, + answers: { + safety_buffering_action: { + answers: ['Retry with a faster model'] + } + } + }); + + await expect(running).resolves.toBe('exit'); + expect(harness.interruptedTurns).toEqual([{ threadId: 'thread-1', turnId: 'turn-1' }]); + expect(harness.rollbackCalls).toEqual([{ threadId: 'thread-1', numTurns: 1 }]); + expect(harness.startTurnMessages).toEqual(['first message', 'first message']); + expect(harness.startTurnParams[1]).toMatchObject({ + threadId: 'thread-1', + effort: 'low', + input: [{ type: 'text', text: 'first message' }], + collaborationMode: { + mode: 'default', + settings: { + model: 'gpt-5.4-mini', + reasoning_effort: 'low' + } + } + }); + expect(getModel()).toBe('gpt-5.4-mini'); + expect(getModelReasoningEffort()).toBe('low'); + }); + + it('keeps the original safety-buffered turn running when the user dismisses the retry', async () => { + harness.emitSafetyBuffering = true; + harness.safetyBufferingFasterModel = 'gpt-5.4-mini'; + const { session, rpcHandlers, getAgentState } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(Object.keys(getAgentState().requests)).toHaveLength(1); + }); + + const requestId = Object.keys(getAgentState().requests)[0]; + await rpcHandlers.get('permission')?.({ + id: requestId, + approved: true, + answers: { + safety_buffering_action: { + answers: ['Keep waiting'] + } + } + }); + + expect(harness.interruptedTurns).toEqual([]); + expect(harness.rollbackCalls).toEqual([]); + expect(harness.startTurnMessages).toEqual(['first message']); + await new Promise((resolve) => setTimeout(resolve, 0)); + + harness.dispatchNotification?.('model/safetyBuffering/updated', { + threadId: 'thread-1', + turnId: 'turn-1', + model: 'gpt-5.4', + useCases: ['cyber'], + reasons: ['review'], + showBufferingUi: true, + fasterModel: 'gpt-5.4-mini' + }); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(getAgentState().requests).toEqual({}); + + harness.dispatchNotification?.('model/safetyBuffering/updated', { + threadId: 'thread-1', + turnId: 'turn-1', + model: 'gpt-5.4', + useCases: ['cyber'], + reasons: ['review'], + showBufferingUi: false, + fasterModel: 'gpt-5.4-mini' + }); + harness.dispatchNotification?.('model/safetyBuffering/updated', { + threadId: 'thread-1', + turnId: 'turn-1', + model: 'gpt-5.4', + useCases: ['cyber'], + reasons: ['review'], + showBufferingUi: true, + fasterModel: 'gpt-5.4-mini' + }); + await vi.waitFor(() => { + expect(Object.keys(getAgentState().requests)).toHaveLength(1); + }); + + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turnId: 'turn-1', + turn: { id: 'turn-1', status: 'completed' } + }); + await expect(running).resolves.toBe('exit'); + }); + + it('dismisses safety-buffering choices when hidden or when agent output starts', async () => { + harness.emitSafetyBuffering = true; + harness.safetyBufferingFasterModel = 'gpt-5.4-mini'; + const { session, getAgentState } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(Object.keys(getAgentState().requests)).toHaveLength(1); + }); + const hiddenRequestId = Object.keys(getAgentState().requests)[0]; + + harness.dispatchNotification?.('model/safetyBuffering/updated', { + threadId: 'thread-1', + turnId: 'turn-1', + model: 'gpt-5.4', + useCases: ['cyber'], + reasons: ['review'], + showBufferingUi: false, + fasterModel: 'gpt-5.4-mini' + }); + await vi.waitFor(() => { + expect(getAgentState().requests).toEqual({}); + expect(getAgentState().completedRequests[hiddenRequestId]).toMatchObject({ + status: 'canceled', + reason: 'Safety buffering ended' + }); + }); + + harness.dispatchNotification?.('model/safetyBuffering/updated', { + threadId: 'thread-1', + turnId: 'turn-1', + model: 'gpt-5.4', + useCases: ['cyber'], + reasons: ['review'], + showBufferingUi: true, + fasterModel: 'gpt-5.4-mini' + }); + await vi.waitFor(() => { + expect(Object.keys(getAgentState().requests)).toHaveLength(1); + }); + const outputRequestId = Object.keys(getAgentState().requests)[0]; + expect(outputRequestId).not.toBe(hiddenRequestId); + + harness.dispatchNotification?.('item/agentMessage/delta', { + threadId: 'thread-1', + turnId: 'turn-1', + itemId: 'message-1', + delta: 'Visible response' + }); + await vi.waitFor(() => { + expect(getAgentState().requests).toEqual({}); + expect(getAgentState().completedRequests[outputRequestId]).toMatchObject({ + status: 'canceled', + reason: 'Agent output started' + }); + }); + expect(harness.interruptedTurns).toEqual([]); + expect(harness.rollbackCalls).toEqual([]); + + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turnId: 'turn-1', + turn: { id: 'turn-1', status: 'completed' } + }); + await expect(running).resolves.toBe('exit'); + }); + + it('surfaces safety buffering without offering retry when fasterModel is null', async () => { + harness.emitSafetyBuffering = true; + harness.safetyBufferingFasterModel = null; + const { session, sessionEvents, getAgentState } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Codex is taking extra time to review this request. Learn more: https://help.openai.com/en/articles/20001326' + }); + }); + expect(getAgentState().requests).toEqual({}); + expect(harness.interruptedTurns).toEqual([]); + expect(harness.rollbackCalls).toEqual([]); + + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turnId: 'turn-1', + turn: { id: 'turn-1', status: 'completed' } + }); + await expect(running).resolves.toBe('exit'); + }); + + it('does not replay a safety-buffered turn when rollback is unavailable', async () => { + harness.emitSafetyBuffering = true; + harness.safetyBufferingFasterModel = 'gpt-5.4-mini'; + harness.emitTurnAbortedOnInterrupt = true; + harness.rollbackErrors.push(new Error('thread/rollback is unsupported')); + const { session, sessionEvents, rpcHandlers, getAgentState } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(Object.keys(getAgentState().requests)).toHaveLength(1); + }); + + const requestId = Object.keys(getAgentState().requests)[0]; + await rpcHandlers.get('permission')?.({ + id: requestId, + approved: true, + answers: { + safety_buffering_action: { + answers: ['Retry with a faster model'] + } + } + }); + + await expect(running).resolves.toBe('exit'); + expect(harness.startTurnMessages).toEqual(['first message']); + expect(harness.rollbackCalls).toEqual([{ threadId: 'thread-1', numTurns: 1 }]); + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Failed to retry with a faster model: thread/rollback is unsupported' + }); + await vi.waitFor(() => { + expect(sessionEvents).toContainEqual({ type: 'ready' }); + }); + expect(session.thinking).toBe(false); + }); + + it('keeps the original turn running when safety-buffering interrupt fails', async () => { + harness.emitSafetyBuffering = true; + harness.safetyBufferingFasterModel = 'gpt-5.4-mini'; + harness.interruptErrors.push(new Error('turn/interrupt failed')); + const { session, sessionEvents, rpcHandlers, getAgentState } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(Object.keys(getAgentState().requests)).toHaveLength(1); + }); + + const requestId = Object.keys(getAgentState().requests)[0]; + await rpcHandlers.get('permission')?.({ + id: requestId, + approved: true, + answers: { + safety_buffering_action: { + answers: ['Retry with a faster model'] + } + } + }); + await vi.waitFor(() => { + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Failed to retry with a faster model: turn/interrupt failed' + }); + }); + + expect(harness.startTurnMessages).toEqual(['first message']); + expect(harness.rollbackCalls).toEqual([]); + expect(session.thinking).toBe(true); + + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turnId: 'turn-1', + turn: { id: 'turn-1', status: 'completed' } + }); + await expect(running).resolves.toBe('exit'); + expect(session.thinking).toBe(false); + }); + + it('surfaces model reroute and Trusted Access verification notices', async () => { + harness.emitModelSafetyNotices = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + await codexRemoteLauncher(session as never); + + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Codex rerouted the model from gpt-5.4 to gpt-5.4-codex (highRiskCyberActivity).' + }); + expect(sessionEvents).toContainEqual({ + type: 'message', + message: 'Your conversations have multiple flags for possible cybersecurity risk. Responses may take longer because extra safety checks are on. To get authorized for security work, join [Trusted Access for Cyber](https://chatgpt.com/cyber).' + }); + }); + it('compacts the same thread before retrying context-window overflow', async () => { harness.remainingThreadSystemErrors = 1; harness.nextThreadSystemErrorMessage = "Codex ran out of room in the model's context window. Start a new thread or clear earlier history before retrying."; diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index 2cb80fe1..5e35c376 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -92,6 +92,10 @@ const CONTEXT_COMPACT_RETRYABLE_ERROR_PATTERNS = [ const SAME_THREAD_MAX_RETRIES = 3; const SAME_THREAD_MAX_COMPACT_RETRIES = 1; const SAME_THREAD_COMPACT_TIMEOUT_MS = 10 * 60 * 1000; +const THREAD_STATUS_FAILURE_GRACE_MS = 250; +const SAFETY_BUFFERING_LEARN_MORE_URL = 'https://help.openai.com/en/articles/20001326'; +const TRUSTED_ACCESS_FOR_CYBER_URL = 'https://chatgpt.com/cyber'; +const CYBER_POLICY_TRUSTED_ACCESS_URL = 'https://openai.com/form/enterprise-trusted-access-for-cyber/'; const CODEX_GOALS_UNSUPPORTED_MESSAGE = 'Codex goals are not supported by this Codex runtime. Upgrade Codex or enable features.goals.'; const MAX_CODEX_GOAL_OBJECTIVE_CHARS = 4_000; @@ -137,6 +141,30 @@ function isSameThreadRetryableCodexError(error: string | null): boolean { return SAME_THREAD_RETRYABLE_ERROR_PATTERNS.some((pattern) => normalized.includes(pattern)); } +function normalizePolicyToken(value: unknown): string { + return typeof value === 'string' + ? value.toLowerCase().replace(/[^a-z0-9]/g, '') + : ''; +} + +function isPolicyBlockedCodexFailure(msg: Record, error: string | null): boolean { + if (normalizePolicyToken(msg.codex_error_info ?? msg.codexErrorInfo) === 'cyberpolicy') { + return true; + } + + const normalizedError = error?.toLowerCase() ?? ''; + return normalizedError.includes('flagged for possible cybersecurity risk') + || normalizedError.includes('flagged for potentially high-risk cyber activity') + || normalizedError.includes('cyber policy') + || normalizedError.includes('cyberpolicy') + || normalizedError.includes('limited access to this content for safety reasons') + || normalizedError.includes("this content can't be shown"); +} + +function isGenericThreadSystemError(error: string | null): boolean { + return error?.trim().toLowerCase() === 'codex thread entered systemerror'; +} + function isContextCompactRetryableCodexError(error: string | null): boolean { if (!error) { return false; @@ -1778,6 +1806,22 @@ class CodexRemoteLauncher extends RemoteLauncherBase { let sameThreadRetryAttempt = 0; let sameThreadCompactAttempt = 0; let recoveryInFlight = false; + let lastFinalizedTurnId: string | null = null; + let deferredThreadStatusFailure: { + event: Record; + threadId: string; + turnId: string; + timer: ReturnType; + } | null = null; + let activeSafetyBufferingRequest: { + requestId: string; + threadId: string; + turnId: string; + fasterModel: string; + message: QueuedMessage; + } | null = null; + const dismissedSafetyBufferingKeys = new Set(); + let agentMessageStartedForTurn = false; let compactRecovery: { threadId: string; message: QueuedMessage; @@ -1921,6 +1965,190 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return true; }; + const clearDeferredThreadStatusFailure = () => { + if (!deferredThreadStatusFailure) { + return; + } + clearTimeout(deferredThreadStatusFailure.timer); + deferredThreadStatusFailure = null; + recoveryInFlight = false; + }; + + const cancelSafetyBufferingRequest = (reason: string) => { + const request = activeSafetyBufferingRequest; + if (!request) { + return; + } + activeSafetyBufferingRequest = null; + permissionHandler.cancelUserInputRequest(request.requestId, reason); + }; + + const safetyBufferingTurnKey = (threadId: string, turnId: string) => `${threadId}\u0000${turnId}`; + const safetyBufferingKey = (threadId: string, turnId: string, fasterModel: string) => { + return `${safetyBufferingTurnKey(threadId, turnId)}\u0000${fasterModel}`; + }; + const clearDismissedSafetyBufferingForTurn = (threadId: string | null, turnId: string | null) => { + if (!threadId || !turnId) { + return; + } + const prefix = `${safetyBufferingTurnKey(threadId, turnId)}\u0000`; + for (const key of dismissedSafetyBufferingKeys) { + if (key.startsWith(prefix)) { + dismissedSafetyBufferingKeys.delete(key); + } + } + }; + + const safetyBufferingChoice = (answers: unknown): string | null => { + const answersRecord = asRecord(answers); + const action = asRecord(answersRecord?.safety_buffering_action); + const values = action?.answers ?? answersRecord?.safety_buffering_action; + if (!Array.isArray(values)) { + return null; + } + return values.find((value): value is string => typeof value === 'string') ?? null; + }; + + const retrySafetyBufferedTurn = async (request: NonNullable) => { + if ( + this.currentThreadId !== request.threadId + || this.currentTurnId !== request.turnId + || !turnInFlight + || agentMessageStartedForTurn + ) { + return; + } + + recoveryInFlight = true; + suppressReadyForInterruptedTurn(request.turnId); + clearReadyAfterTurnTimer?.(); + let interrupted = false; + try { + await appServerClient.interruptTurn({ + threadId: request.threadId, + turnId: request.turnId + }); + interrupted = true; + await appServerClient.rollbackThread({ + threadId: request.threadId, + numTurns: 1 + }); + + lastFinalizedTurnId = request.turnId; + turnInFlight = false; + allowAnonymousTerminalEvent = false; + this.currentTurnId = null; + sameThreadRetryAttempt = 0; + sameThreadCompactAttempt = 0; + + const retryMode: EnhancedMode = { + ...request.message.mode, + model: request.fasterModel, + modelReasoningEffort: 'low' + }; + session.setModel(request.fasterModel); + session.setModelReasoningEffort('low'); + pending = { + ...request.message, + mode: retryMode + }; + const message = `Retrying with the faster model ${request.fasterModel}.`; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } catch (error) { + if (interrupted) { + lastFinalizedTurnId = request.turnId; + turnInFlight = false; + allowAnonymousTerminalEvent = false; + this.currentTurnId = null; + activeMessage = null; + } else { + consumeInterruptedTurnReadySuppression(request.turnId); + } + const message = `Failed to retry with a faster model: ${errorMessage(error)}`; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } finally { + recoveryInFlight = false; + wakeLoop(); + if (interrupted && !pending) { + scheduleReadyAfterTurn?.(); + } + } + }; + + const showSafetyBufferingRequest = (args: { + threadId: string; + turnId: string; + fasterModel: string; + }) => { + if (!activeMessage || agentMessageStartedForTurn) { + return; + } + if (dismissedSafetyBufferingKeys.has(safetyBufferingKey(args.threadId, args.turnId, args.fasterModel))) { + return; + } + if ( + activeSafetyBufferingRequest?.threadId === args.threadId + && activeSafetyBufferingRequest.turnId === args.turnId + && activeSafetyBufferingRequest.fasterModel === args.fasterModel + ) { + return; + } + + cancelSafetyBufferingRequest('Safety buffering prompt replaced'); + const request = { + requestId: `codex-safety-buffering:${args.threadId}:${args.turnId}:${randomUUID()}`, + ...args, + message: activeMessage + }; + activeSafetyBufferingRequest = request; + + void permissionHandler.handleUserInputRequest(request.requestId, { + questions: [{ + id: 'safety_buffering_action', + question: 'Codex is taking extra time to review this request. What would you like to do?', + options: [ + { + label: 'Retry with a faster model', + description: `Interrupt this turn and retry with ${args.fasterModel} using low reasoning effort.` + }, + { + label: 'Keep waiting', + description: 'Dismiss this choice and let the current turn continue.' + }, + { + label: 'Learn more', + description: `[Read about safety checks](${SAFETY_BUFFERING_LEARN_MORE_URL}); the current turn will keep waiting.` + } + ] + }] + }).then((answers) => { + if (activeSafetyBufferingRequest !== request) { + return; + } + activeSafetyBufferingRequest = null; + const choice = safetyBufferingChoice(answers); + if (choice === 'Retry with a faster model') { + void retrySafetyBufferedTurn(request); + } else if (choice === 'Keep waiting' || choice === 'Learn more') { + dismissedSafetyBufferingKeys.add( + safetyBufferingKey(request.threadId, request.turnId, request.fasterModel) + ); + if (choice === 'Learn more') { + const message = `Learn more about Codex safety checks: ${SAFETY_BUFFERING_LEARN_MORE_URL}`; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } + } + }).catch((error) => { + if (activeSafetyBufferingRequest === request) { + activeSafetyBufferingRequest = null; + } + logger.debug(`[Codex] Safety buffering choice dismissed: ${errorMessage(error)}`); + }); + }; + const shouldForwardGoalUpdate = (msg: Record, threadId: string | null): boolean => { const goal = asRecord(msg.goal); const scopedThreadId = threadId @@ -1962,9 +2190,6 @@ class CodexRemoteLauncher extends RemoteLauncherBase { 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'; - const suppressReadyForThisTerminalEvent = isTerminalEvent - ? consumeInterruptedTurnReadySuppression(eventTurnId) - : false; if (msgType === 'thread_started') { const threadId = asString(msg.thread_id ?? msg.threadId); @@ -2035,8 +2260,15 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return; } + if (isTerminalEvent && eventTurnId && eventTurnId === lastFinalizedTurnId) { + logger.debug(`[Codex] Ignoring duplicate terminal event for turn ${eventTurnId}`); + return; + } + if (msgType === 'task_started') { const turnId = eventTurnId; + agentMessageStartedForTurn = false; + dismissedSafetyBufferingKeys.clear(); if (turnId) { this.currentTurnId = turnId; allowAnonymousTerminalEvent = false; @@ -2045,20 +2277,167 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } } + if (msgType === 'agent_message_delta') { + agentMessageStartedForTurn = true; + if ( + activeSafetyBufferingRequest + && (!eventTurnId || activeSafetyBufferingRequest.turnId === eventTurnId) + ) { + cancelSafetyBufferingRequest('Agent output started'); + } + return; + } + + if (msgType === 'model_safety_buffering') { + const showBufferingUi = msg.show_buffering_ui === true; + if (!showBufferingUi) { + clearDismissedSafetyBufferingForTurn( + eventThreadId ?? this.currentThreadId, + eventTurnId ?? this.currentTurnId + ); + if ( + activeSafetyBufferingRequest + && (!eventTurnId || activeSafetyBufferingRequest.turnId === eventTurnId) + ) { + cancelSafetyBufferingRequest('Safety buffering ended'); + } + return; + } + + const fasterModel = asString(msg.faster_model ?? msg.fasterModel); + if ( + fasterModel + && eventThreadId + && eventTurnId + && eventThreadId === this.currentThreadId + && eventTurnId === this.currentTurnId + && turnInFlight + ) { + showSafetyBufferingRequest({ + threadId: eventThreadId, + turnId: eventTurnId, + fasterModel + }); + } else if (!fasterModel) { + const message = `Codex is taking extra time to review this request. Learn more: ${SAFETY_BUFFERING_LEARN_MORE_URL}`; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } + return; + } + + if (msgType === 'model_rerouted') { + const fromModel = asString(msg.from_model ?? msg.fromModel); + const toModel = asString(msg.to_model ?? msg.toModel); + const reason = asString(msg.reason); + if (fromModel && toModel) { + const message = `Codex rerouted the model from ${fromModel} to ${toModel}${reason ? ` (${reason})` : ''}.`; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } + return; + } + + if (msgType === 'model_verification') { + const verifications = Array.isArray(msg.verifications) ? msg.verifications : []; + if (verifications.includes('trustedAccessForCyber')) { + const message = 'Your conversations have multiple flags for possible cybersecurity risk. ' + + 'Responses may take longer because extra safety checks are on. To get authorized for ' + + `security work, join [Trusted Access for Cyber](${TRUSTED_ACCESS_FOR_CYBER_URL}).`; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } + return; + } + const isThreadStatusFailure = msgType === 'task_failed' && msg.terminal_source === 'thread_status'; const error = msgType === 'task_failed' ? asString(msg.error) : null; + const explicitlyNonRetryable = msgType === 'task_failed' + && (msg.retryable === false || isPolicyBlockedCodexFailure(msg, error)); + + if (deferredThreadStatusFailure && isTerminalEvent && !isThreadStatusFailure) { + const sameThread = !eventThreadId || eventThreadId === deferredThreadStatusFailure.threadId; + const sameTurn = !eventTurnId || eventTurnId === deferredThreadStatusFailure.turnId; + if (sameThread && sameTurn) { + if ( + msgType === 'task_failed' + && msg.terminal_source === 'turn_completed' + && !error + && !explicitlyNonRetryable + ) { + const deferred = deferredThreadStatusFailure; + clearDeferredThreadStatusFailure(); + await handleCodexEvent({ + ...deferred.event, + turn_id: deferred.turnId, + deferred_thread_status: true + }); + return; + } + clearDeferredThreadStatusFailure(); + } + } + + if ( + isThreadStatusFailure + && isGenericThreadSystemError(error) + && msg.deferred_thread_status !== true + ) { + if (shouldIgnoreTerminalEvent({ + eventTurnId, + currentTurnId: this.currentTurnId, + turnInFlight, + allowAnonymousTerminalEvent, + eventThreadId, + currentThreadId: this.currentThreadId, + allowMatchingThreadIdTerminalEvent: true + })) { + return; + } + const threadId = eventThreadId ?? this.currentThreadId; + const turnId = eventTurnId ?? this.currentTurnId; + if (!threadId || !turnId) { + return; + } + clearDeferredThreadStatusFailure(); + const event = { ...msg }; + const timer = setTimeout(() => { + if (deferredThreadStatusFailure?.event !== event) { + return; + } + deferredThreadStatusFailure = null; + recoveryInFlight = false; + void handleCodexEvent({ + ...event, + turn_id: turnId, + deferred_thread_status: true + }).catch((deferredError) => { + logger.debug(`[Codex] Failed to handle deferred thread status: ${errorMessage(deferredError)}`); + }); + }, THREAD_STATUS_FAILURE_GRACE_MS); + deferredThreadStatusFailure = { event, threadId, turnId, timer }; + recoveryInFlight = true; + return; + } + const shouldCompactAndRetrySameThread = msgType === 'task_failed' + && !explicitlyNonRetryable && isContextCompactRetryableCodexError(error) && Boolean(activeMessage) && Boolean(this.currentThreadId) && sameThreadCompactAttempt < SAME_THREAD_MAX_COMPACT_RETRIES; const shouldRetrySameThread = msgType === 'task_failed' + && !explicitlyNonRetryable && !shouldCompactAndRetrySameThread && isSameThreadRetryableCodexError(error) && Boolean(activeMessage) && Boolean(this.currentThreadId) && sameThreadRetryAttempt < SAME_THREAD_MAX_RETRIES; + const suppressReadyForThisTerminalEvent = isTerminalEvent + ? consumeInterruptedTurnReadySuppression(eventTurnId) + : false; + if (isTerminalEvent) { if (shouldIgnoreTerminalEvent({ eventTurnId, @@ -2077,6 +2456,20 @@ class CodexRemoteLauncher extends RemoteLauncherBase { ); return; } + const finalizedTurnId = eventTurnId ?? this.currentTurnId; + if (finalizedTurnId) { + lastFinalizedTurnId = finalizedTurnId; + } + clearDismissedSafetyBufferingForTurn( + eventThreadId ?? this.currentThreadId, + finalizedTurnId + ); + if ( + activeSafetyBufferingRequest + && (!finalizedTurnId || activeSafetyBufferingRequest.turnId === finalizedTurnId) + ) { + cancelSafetyBufferingRequest('Turn completed'); + } if (shouldCompactAndRetrySameThread) { const threadId = this.currentThreadId; const messageToRetry = activeMessage; @@ -2139,7 +2532,10 @@ class CodexRemoteLauncher extends RemoteLauncherBase { messageBuffer.addMessage(retryMessage, 'status'); session.sendSessionEvent({ type: 'message', message: retryMessage }); } else { - const message = error ? `Task failed: ${error}` : 'Task failed'; + const visibleError = error && isPolicyBlockedCodexFailure(msg, error) + ? `${error}\n\nTrusted Access: ${CYBER_POLICY_TRUSTED_ACCESS_URL}\nLearn more: ${SAFETY_BUFFERING_LEARN_MORE_URL}` + : error; + const message = visibleError ? `Task failed: ${visibleError}` : 'Task failed'; messageBuffer.addMessage(message, 'status'); session.sendSessionEvent({ type: 'message', message }); } @@ -2172,7 +2568,16 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } if (isTerminalEvent && !turnInFlight && !suppressReadyForThisTerminalEvent) { - scheduleReadyAfterTurn?.(); + if (msg.deferred_thread_status === true) { + emitReadyIfIdle({ + pending: pending ?? (recoveryInFlight ? activeMessage : null), + queueSize: () => session.queue.size(), + shouldExit: this.shouldExit, + sendReady + }); + } else { + scheduleReadyAfterTurn?.(); + } } else if (readyAfterTurnTimer && msgType !== 'task_started' && !suppressReadyForThisTerminalEvent) { scheduleReadyAfterTurn?.(); } @@ -2647,6 +3052,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { }; const resetCurrentTurnState = () => { + clearDeferredThreadStatusFailure(); + cancelSafetyBufferingRequest('Session reset'); turnInFlight = false; allowAnonymousTerminalEvent = false; this.currentTurnId = null; @@ -2935,7 +3342,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { while (!this.shouldExit) { logActiveHandles('loop-top'); - if (!pending && (turnInFlight || recoveryInFlight) && session.queue.size() === 0) { + if (!pending && (recoveryInFlight || (turnInFlight && session.queue.size() === 0))) { await waitForTurnOrRecovery(this.abortController.signal); if (this.abortController.signal.aborted && !this.shouldExit) { logger.debug('[codex]: Internal wait aborted while turn/recovery was active; continuing'); @@ -3156,6 +3563,8 @@ class CodexRemoteLauncher extends RemoteLauncherBase { } failPendingAgentStarts('spawn_agent did not return an agent id before the Codex session ended'); + clearDeferredThreadStatusFailure(); + cancelSafetyBufferingRequest('Session ended'); cancelAllPendingThrottledAgentRunUpdates(); } diff --git a/cli/src/codex/utils/appServerEventConverter.test.ts b/cli/src/codex/utils/appServerEventConverter.test.ts index 6bb8d9c4..62f68489 100644 --- a/cli/src/codex/utils/appServerEventConverter.test.ts +++ b/cli/src/codex/utils/appServerEventConverter.test.ts @@ -72,7 +72,12 @@ describe('AppServerEventConverter', () => { expect(interrupted).toEqual([{ type: 'turn_aborted', turn_id: 'turn-1' }]); const failed = converter.handleNotification('turn/completed', { turn: { id: 'turn-1' }, status: 'Failed', message: 'boom' }); - expect(failed).toEqual([{ type: 'task_failed', turn_id: 'turn-1', error: 'boom' }]); + expect(failed).toEqual([{ + type: 'task_failed', + turn_id: 'turn-1', + terminal_source: 'turn_completed', + error: 'boom' + }]); }); it('accumulates agent message deltas', () => { @@ -693,6 +698,120 @@ describe('AppServerEventConverter', () => { expect(events).toEqual([{ type: 'task_failed', error: 'fatal' }]); }); + it('preserves typed non-retryable cyber-policy errors', () => { + const converter = new AppServerEventConverter(); + + const events = converter.handleNotification('error', { + threadId: 'thread-1', + turnId: 'turn-1', + error: { + message: 'This content was flagged for possible cybersecurity risk.', + codexErrorInfo: 'cyberPolicy' + }, + willRetry: false + }); + + expect(events).toEqual([{ + type: 'task_failed', + thread_id: 'thread-1', + turn_id: 'turn-1', + terminal_source: 'error', + retryable: false, + codex_error_info: 'cyberPolicy', + error: 'This content was flagged for possible cybersecurity risk.' + }]); + }); + + it('preserves cyber-policy metadata from wrapped and completed-turn errors', () => { + const converter = new AppServerEventConverter(); + + expect(converter.handleNotification('codex/event/error', { + msg: { + type: 'error', + thread_id: 'thread-1', + turn_id: 'turn-1', + message: 'wrapped policy failure', + codex_error_info: 'cyber_policy', + will_retry: false + } + })).toEqual([{ + type: 'task_failed', + thread_id: 'thread-1', + turn_id: 'turn-1', + retryable: false, + codex_error_info: 'cyber_policy', + error: 'wrapped policy failure' + }]); + + expect(converter.handleNotification('turn/completed', { + threadId: 'thread-1', + turn: { + id: 'turn-1', + status: 'failed', + error: { + message: 'completed policy failure', + codexErrorInfo: 'CyberPolicy' + } + } + })).toEqual([{ + type: 'task_failed', + thread_id: 'thread-1', + turn_id: 'turn-1', + terminal_source: 'turn_completed', + codex_error_info: 'CyberPolicy', + error: 'completed policy failure' + }]); + }); + + it('maps Codex model safety notifications', () => { + const converter = new AppServerEventConverter(); + + expect(converter.handleNotification('model/safetyBuffering/updated', { + threadId: 'thread-1', + turnId: 'turn-1', + model: 'gpt-5.4', + useCases: ['cyber'], + reasons: ['review'], + showBufferingUi: true, + fasterModel: 'gpt-5.4-mini' + })).toEqual([{ + type: 'model_safety_buffering', + thread_id: 'thread-1', + turn_id: 'turn-1', + model: 'gpt-5.4', + use_cases: ['cyber'], + reasons: ['review'], + show_buffering_ui: true, + faster_model: 'gpt-5.4-mini' + }]); + + expect(converter.handleNotification('model/rerouted', { + threadId: 'thread-1', + turnId: 'turn-1', + fromModel: 'gpt-5.4', + toModel: 'gpt-5.4-codex', + reason: 'highRiskCyberActivity' + })).toEqual([{ + type: 'model_rerouted', + thread_id: 'thread-1', + turn_id: 'turn-1', + from_model: 'gpt-5.4', + to_model: 'gpt-5.4-codex', + reason: 'highRiskCyberActivity' + }]); + + expect(converter.handleNotification('model/verification', { + threadId: 'thread-1', + turnId: 'turn-1', + verifications: ['trustedAccessForCyber'] + })).toEqual([{ + type: 'model_verification', + thread_id: 'thread-1', + turn_id: 'turn-1', + verifications: ['trustedAccessForCyber'] + }]); + }); + it('maps thread/compacted notifications', () => { const converter = new AppServerEventConverter(); const events = converter.handleNotification('thread/compacted', { diff --git a/cli/src/codex/utils/appServerEventConverter.ts b/cli/src/codex/utils/appServerEventConverter.ts index 28bae2be..1944a723 100644 --- a/cli/src/codex/utils/appServerEventConverter.ts +++ b/cli/src/codex/utils/appServerEventConverter.ts @@ -306,6 +306,18 @@ function extractStringArray(value: unknown): string[] { : []; } +function extractCodexErrorInfo( + record: Record, + errorRecord: Record | null +): string | null { + return asString( + record.codexErrorInfo + ?? record.codex_error_info + ?? errorRecord?.codexErrorInfo + ?? errorRecord?.codex_error_info + ); +} + function buildCollabAgentInput(item: Record, toolName: string): Record { const targets = extractStringArray(item.receiverThreadIds ?? item.receiver_thread_ids ?? item.targets); const input: Record = {}; @@ -526,12 +538,19 @@ export class AppServerEventConverter { if (msgType === 'error') { const errorRecord = asRecord(msg.error); - const willRetry = asBoolean(msg.will_retry ?? msg.willRetry ?? errorRecord?.will_retry ?? errorRecord?.willRetry) ?? false; + const retryable = asBoolean(msg.will_retry ?? msg.willRetry ?? errorRecord?.will_retry ?? errorRecord?.willRetry); + const willRetry = retryable ?? false; if (willRetry) { return []; } const error = asString(msg.message ?? msg.reason ?? errorRecord?.message); - return error ? addEventScope([{ type: 'task_failed', error }], msgScope) : []; + const codexErrorInfo = extractCodexErrorInfo(msg, errorRecord); + return error ? addEventScope([{ + type: 'task_failed', + ...(retryable !== null ? { retryable } : {}), + ...(codexErrorInfo ? { codex_error_info: codexErrorInfo } : {}), + error + }], msgScope) : []; } if (msgType === 'plan_update') { @@ -672,7 +691,9 @@ export class AppServerEventConverter { const statusRaw = asString(paramsRecord.status ?? turn.status); const status = statusRaw?.toLowerCase(); const turnId = asString(turn.turnId ?? turn.turn_id ?? turn.id); - const errorMessage = asString(paramsRecord.error ?? paramsRecord.message ?? paramsRecord.reason); + const turnError = asRecord(paramsRecord.error ?? turn.error); + const errorMessage = asString(paramsRecord.error ?? paramsRecord.message ?? paramsRecord.reason) + ?? asString(turnError?.message); if (status === 'interrupted' || status === 'cancelled' || status === 'canceled') { events.push(scoped({ type: 'turn_aborted', ...(turnId ? { turn_id: turnId } : {}) })); @@ -680,7 +701,14 @@ export class AppServerEventConverter { } if (status === 'failed' || status === 'error') { - events.push(scoped({ type: 'task_failed', ...(turnId ? { turn_id: turnId } : {}), ...(errorMessage ? { error: errorMessage } : {}) })); + const codexErrorInfo = extractCodexErrorInfo(paramsRecord, turnError); + events.push(scoped({ + type: 'task_failed', + ...(turnId ? { turn_id: turnId } : {}), + terminal_source: 'turn_completed', + ...(codexErrorInfo ? { codex_error_info: codexErrorInfo } : {}), + ...(errorMessage ? { error: errorMessage } : {}) + })); return events; } @@ -702,12 +730,66 @@ export class AppServerEventConverter { return events; } + if (method === 'model/safetyBuffering/updated') { + const model = asString(paramsRecord.model); + const showBufferingUi = asBoolean(paramsRecord.showBufferingUi ?? paramsRecord.show_buffering_ui); + if (!model || showBufferingUi === null) { + return events; + } + events.push(scoped({ + type: 'model_safety_buffering', + model, + use_cases: extractStringArray(paramsRecord.useCases ?? paramsRecord.use_cases), + reasons: extractStringArray(paramsRecord.reasons), + show_buffering_ui: showBufferingUi, + faster_model: asString(paramsRecord.fasterModel ?? paramsRecord.faster_model) + })); + return events; + } + + if (method === 'model/rerouted') { + const fromModel = asString(paramsRecord.fromModel ?? paramsRecord.from_model); + const toModel = asString(paramsRecord.toModel ?? paramsRecord.to_model); + const reason = asString(paramsRecord.reason); + if (fromModel && toModel && reason) { + events.push(scoped({ + type: 'model_rerouted', + from_model: fromModel, + to_model: toModel, + reason + })); + } + return events; + } + + if (method === 'model/verification') { + events.push(scoped({ + type: 'model_verification', + verifications: extractStringArray(paramsRecord.verifications) + })); + return events; + } + if (method === 'error') { - const willRetry = asBoolean(paramsRecord.will_retry ?? paramsRecord.willRetry) ?? false; + const errorRecord = asRecord(paramsRecord.error); + const retryable = asBoolean( + paramsRecord.will_retry + ?? paramsRecord.willRetry + ?? errorRecord?.will_retry + ?? errorRecord?.willRetry + ); + const willRetry = retryable ?? false; if (willRetry) return events; - const message = asString(paramsRecord.message) ?? asString(asRecord(paramsRecord.error)?.message); + const message = asString(paramsRecord.message) ?? asString(errorRecord?.message); if (message) { - events.push(scoped({ type: 'task_failed', error: message })); + const codexErrorInfo = extractCodexErrorInfo(paramsRecord, errorRecord); + events.push(scoped({ + type: 'task_failed', + terminal_source: 'error', + ...(retryable !== null ? { retryable } : {}), + ...(codexErrorInfo ? { codex_error_info: codexErrorInfo } : {}), + error: message + })); } return events; } @@ -723,6 +805,7 @@ export class AppServerEventConverter { this.lastAgentMessageDeltaByItemId.set(itemId, delta); const prev = this.agentMessageBuffers.get(itemId) ?? ''; this.agentMessageBuffers.set(itemId, prev + delta); + events.push(scoped({ type: 'agent_message_delta' })); } return events; } diff --git a/cli/src/codex/utils/permissionHandler.test.ts b/cli/src/codex/utils/permissionHandler.test.ts index 8de0adca..f141b661 100644 --- a/cli/src/codex/utils/permissionHandler.test.ts +++ b/cli/src/codex/utils/permissionHandler.test.ts @@ -172,4 +172,31 @@ describe('CodexPermissionHandler', () => { } }); }); + + it('cancels one request_user_input without resetting other pending requests', async () => { + const { handler, getAgentState } = createHarness('default'); + const first = handler.handleUserInputRequest('input-1', { + questions: [{ id: 'action', question: 'First?' }] + }); + const second = handler.handleUserInputRequest('input-2', { + questions: [{ id: 'action', question: 'Second?' }] + }); + + handler.cancelUserInputRequest('input-1', 'No longer relevant'); + + await expect(first).rejects.toThrow('No longer relevant'); + expect(getAgentState().requests).toMatchObject({ + 'input-2': { tool: 'request_user_input' } + }); + expect(getAgentState().requests).not.toHaveProperty('input-1'); + expect(getAgentState().completedRequests).toMatchObject({ + 'input-1': { + status: 'canceled', + reason: 'No longer relevant' + } + }); + + handler.reset(); + await expect(second).rejects.toThrow('Session reset'); + }); }); diff --git a/cli/src/codex/utils/permissionHandler.ts b/cli/src/codex/utils/permissionHandler.ts index 89e2d73a..e37fd39b 100644 --- a/cli/src/codex/utils/permissionHandler.ts +++ b/cli/src/codex/utils/permissionHandler.ts @@ -155,6 +155,22 @@ export class CodexPermissionHandler extends BasePermissionHandler