fix(codex): preserve native safety behavior (#1024)

* test: reproduce issue #1020 * fix: preserve Codex safety behavior (closes #1020) * test: cover Codex safety buffering dismissal * test: cover Codex safety retry edge cases * fix: keep dismissed safety buffering prompts hidden
2026-08-08 07:17:39 +00:00 · 2026-07-12 18:39:56 +08:00
parent d160203bb2
commit d9ea9db507
8 changed files with 1166 additions and 14 deletions
@@ -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<string, unknown>, 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<string, unknown>;
+            threadId: string;
+            turnId: string;
+            timer: ReturnType<typeof setTimeout>;
+        } | null = null;
+        let activeSafetyBufferingRequest: {
+            requestId: string;
+            threadId: string;
+            turnId: string;
+            fasterModel: string;
+            message: QueuedMessage;
+        } | null = null;
+        const dismissedSafetyBufferingKeys = new Set<string>();
+        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<typeof activeSafetyBufferingRequest>) => {
+            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<string, unknown>, 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();
    }