hapi/cli/src/codex/codexLocalLauncher.ts

import { CodexRoundDuration } from './utils/roundDuration';
import { logger } from '@/ui/logger';
import { resolve } from 'node:path';
import { startHookServer } from '@/claude/utils/startHookServer';
import { codexLocal } from './codexLocal';
import type { ReasoningEffort } from './appServerTypes';
import { CodexSession } from './session';
import { createCodexSessionScanner, type CodexSessionScanner } from './utils/codexSessionScanner';
import {
    createCodexEventConverter,
    type CodexConversionAction,
    type CodexMessage,
    type CodexSessionEvent
} from './utils/codexEventConverter';
import { buildHapiMcpBridge } from './utils/buildHapiMcpBridge';
import { parseCodexCliOverrides, stripCodexCliOverrides } from './utils/codexCliOverrides';
import { buildCodexPermissionModeCliArgs } from './utils/permissionModeConfig';
import { BaseLocalLauncher } from '@/modules/common/launcher/BaseLocalLauncher';
import { createCodexTranscriptLocator, type CodexTranscriptLocator } from './utils/codexTranscriptLocator';
import { CodexToolHookBridge, isCodexToolHookEvent } from './utils/codexToolHookBridge';
import { countHookCoveredExecCalls } from './utils/codexExecWrapper';

type ProposedPlanMessage = Extract<CodexMessage, { type: 'proposed_plan' }>;
type ToolCallMessage = Extract<CodexMessage, { type: 'tool-call' }>;

type PendingExecWrapper = {
    message: ToolCallMessage;
    turnId?: string;
};

function extractTurnContextReasoningEffort(event: CodexSessionEvent): ReasoningEffort | null | undefined {
    if (event.type !== 'turn_context') {
        return undefined;
    }
    if (!event.payload || typeof event.payload !== 'object') {
        return null;
    }
    const effort = (event.payload as Record<string, unknown>).effort;
    if (typeof effort !== 'string' || !effort.trim()) {
        return null;
    }
    return effort.trim().toLowerCase();
}

function extractTurnContextModel(event: CodexSessionEvent): string | null | undefined {
    if (event.type !== 'turn_context' || !event.payload || typeof event.payload !== 'object') {
        return undefined;
    }
    const model = (event.payload as Record<string, unknown>).model;
    if (model === null) return null;
    if (typeof model !== 'string' || !model.trim()) return undefined;
    return model.trim();
}

export async function codexLocalLauncher(session: CodexSession): Promise<'switch' | 'exit'> {
    const resumeSessionId = session.sessionId;
    let primarySessionId = resumeSessionId;
    let primaryTranscriptPath: string | null = null;
    let scanner: CodexSessionScanner | null = null;
    let hookReady = false;
    let shuttingDown = false;
    let readyTimer: ReturnType<typeof setTimeout> | null = null;
    const completedTurns = new Set<string>();
    const cancelReady = (): void => {
        if (readyTimer !== null) {
            clearTimeout(readyTimer);
            readyTimer = null;
        }
    };
    let pendingScannerSetup: Promise<void> | null = null;
    let transcriptLocator: CodexTranscriptLocator | null = null;
    let scannerTranscriptPath: string | null = null;
    let scannerReplayedExistingHistory = false;
    let transcriptModel: string | null = null;
    let convertTranscriptEvent = createCodexEventConverter();
    const pendingPlansByTurnId = new Map<string, ProposedPlanMessage>();
    const pendingExecWrappers = new Map<string, PendingExecWrapper>();
    const toolHookBridge = new CodexToolHookBridge();
    const permissionMode = session.getPermissionMode();
    const managedPermissionMode = permissionMode === 'read-only' || permissionMode === 'yolo'
        ? permissionMode
        : null;
    const codexArgs = managedPermissionMode
        ? [
            ...buildCodexPermissionModeCliArgs(managedPermissionMode),
            ...stripCodexCliOverrides(session.codexArgs)
        ]
        : session.codexArgs;
    const cwdOverride = parseCodexCliOverrides(session.codexArgs).cwd;
    const effectiveCodexCwd = cwdOverride ? resolve(session.path, cwdOverride) : session.path;

    // Start hapi hub for MCP bridge (same as remote mode)
    const { server: happyServer, mcpServers } = await buildHapiMcpBridge(session.client);
    logger.debug(`[codex-local]: Started hapi MCP bridge server at ${happyServer.url}`);

    const reportTranscriptSyncFailure = (transcriptPath: string, error: unknown): void => {
        const detail = error instanceof Error ? error.message : String(error);
        const message = `Codex transcript sync failed for ${transcriptPath}: ${detail}`;
        logger.warn(`[codex-local]: ${message}`);
        session.sendSessionEvent({
            type: 'message',
            message: `${message} Keeping local Codex running; remote transcript sync is unavailable for this launch.`
        });
    };

    const handleSessionFound = (sessionId: string, allowSwitch = false): void => {
        if (primarySessionId && primarySessionId !== sessionId && !allowSwitch) {
            logger.debug(`[codex-local]: Ignoring non-primary Codex session id ${sessionId}; primary is ${primarySessionId}`);
            return;
        }
        primarySessionId = sessionId;
        session.onSessionFound(sessionId);
    };

    const isPrimarySessionId = (sessionId: string): boolean => {
        return primarySessionId === null || primarySessionId === sessionId;
    };

    const sendProposedPlan = (message: ProposedPlanMessage): void => {
        const callId = `codex-proposed-plan:${message.id}`;
        session.sendAgentMessage({
            type: 'tool-call',
            name: 'ExitPlanMode',
            callId,
            input: { plan: message.plan },
            id: message.id,
            turn_id: message.turnId
        });
        session.sendAgentMessage({
            type: 'tool-call-result',
            callId,
            output: null,
            id: `${message.id}:result`,
            turn_id: message.turnId
        });
    };

    const flushPendingPlan = (turnId: string): void => {
        const message = pendingPlansByTurnId.get(turnId);
        if (!message) {
            return;
        }
        pendingPlansByTurnId.delete(turnId);
        sendProposedPlan(message);
    };

    const flushAllPendingPlans = (): void => {
        for (const turnId of pendingPlansByTurnId.keys()) {
            flushPendingPlan(turnId);
        }
    };

    const flushPendingExecWrapper = (callId: string, result?: CodexMessage): void => {
        const pending = pendingExecWrappers.get(callId);
        if (!pending) return;
        pendingExecWrappers.delete(callId);
        session.sendAgentMessage(pending.message);
        if (result) {
            session.sendAgentMessage(result);
        }
    };

    const flushAllPendingExecWrappers = (): void => {
        for (const [callId, pending] of pendingExecWrappers) {
            session.sendAgentMessage(pending.message);
            session.sendAgentMessage({
                type: 'tool-call-result',
                callId,
                output: { error: 'Codex ended before the exec wrapper returned a result.' },
                is_error: true,
                id: `${pending.message.id}:incomplete`
            });
        }
        pendingExecWrappers.clear();
    };

    const roundDuration = new CodexRoundDuration();

    const dispatchTranscriptActions = (
        actions: CodexConversionAction[],
        context: { replayedHistory: boolean }
    ): void => {
        for (const action of actions) {
            if (action.type === 'session-found') {
                if (!isPrimarySessionId(action.sessionId)) {
                    logger.debug(`[codex-local]: Ignoring converted session id ${action.sessionId}; primary is ${primarySessionId}`);
                    return;
                }
                session.onSessionFound(action.sessionId);
                continue;
            }
            if (action.type === 'user-message') {
                session.sendUserMessage(action.message);
                continue;
            }
            if (action.type === 'user-activity') {
                session.notifyUserActivity();
                continue;
            }
            if (action.type === 'turn-finished') {
                flushPendingPlan(action.turnId);
                for (const message of toolHookBridge.finishTurn(action.turnId)) {
                    session.sendAgentMessage(message);
                }
                continue;
            }

            const turnId = action.turnId ?? roundDuration.turnId;
            const message = { ...action.message, ...(turnId ? { turn_id: turnId } : {}) };
            if (message.type === 'proposed_plan') {
                // Codex may complete the Plan item before emitting its final text preface.
                pendingPlansByTurnId.set(message.turnId, message);
            } else if (message.type === 'tool-call' && message.name === 'exec') {
                if (countHookCoveredExecCalls(message.input) === null) {
                    session.sendAgentMessage(message);
                } else {
                    pendingExecWrappers.set(message.callId, {
                        message,
                        ...(action.turnId ? { turnId: action.turnId } : {})
                    });
                }
            } else if (message.type === 'tool-call-result' && pendingExecWrappers.has(message.callId)) {
                const pending = pendingExecWrappers.get(message.callId);
                const turnId = pending?.turnId ?? action.turnId;
                if (pending && toolHookBridge.hasCompletedAllObservedNestedTools(turnId)) {
                    pendingExecWrappers.delete(message.callId);
                } else {
                    flushPendingExecWrapper(message.callId, message);
                }
            } else {
                const scopedMessage = message.type !== 'token_count'
                    ? message
                    : context.replayedHistory
                        ? { ...message, model: transcriptModel, hapiUsageScope: 'imported-history' }
                        : primarySessionId
                            ? {
                                ...message,
                                flavor: 'codex',
                                model: transcriptModel,
                                threadId: primarySessionId,
                                thread_id: primarySessionId,
                                hapiUsageScope: 'managed'
                            }
                            : { ...message, flavor: 'codex', model: transcriptModel };
                session.sendAgentMessage(scopedMessage);
            }
        }
    };

    const finalizeTranscriptConversion = (replayedExistingHistory: boolean): void => {
        dispatchTranscriptActions(convertTranscriptEvent.finalize(), {
            replayedHistory: replayedExistingHistory
        });
    };

    const drainAndCleanupScanner = async (
        activeScanner: CodexSessionScanner,
        replayedExistingHistory: boolean
    ): Promise<void> => {
        try {
            // Codex can flush its final transcript records after the last watcher tick.
            await activeScanner.flush();
            if (replayedExistingHistory) {
                session.markTranscriptHistoryReplayConsumed();
            }
        } finally {
            try {
                await activeScanner.cleanup();
            } finally {
                finalizeTranscriptConversion(replayedExistingHistory);
            }
        }
    };

    const bindPrimarySession = (sessionId: string, transcriptPath: string, allowSwitch = false): void => {
        if (primarySessionId && primarySessionId !== sessionId && !allowSwitch) {
            logger.debug(`[codex-local]: Ignoring non-primary SessionStart hook ${sessionId}; primary is ${primarySessionId}`);
            return;
        }
        primarySessionId = sessionId;
        primaryTranscriptPath = transcriptPath;
        session.onSessionFound(sessionId);
        hookReady = true;
        session.onTranscriptPathFound(transcriptPath);
    };

    const processTranscriptPath = async (transcriptPath: string): Promise<void> => {
        hookReady = true;
        if (shuttingDown) {
            return;
        }
        if (primaryTranscriptPath && transcriptPath !== primaryTranscriptPath) {
            logger.debug(`[codex-local]: Ignoring non-primary transcript path ${transcriptPath}; primary is ${primaryTranscriptPath}`);
            return;
        }
        if (scanner) {
            if (scannerTranscriptPath !== transcriptPath) {
                await scanner.flush();
                finalizeTranscriptConversion(scannerReplayedExistingHistory);
                flushAllPendingPlans();
                convertTranscriptEvent = createCodexEventConverter();
            }
            await scanner.setTranscriptPath(transcriptPath);
            scannerTranscriptPath = transcriptPath;
            return;
        }
        const replayExistingHistory = session.shouldReplayTranscriptHistory();
        const createdScanner = await createCodexSessionScanner({
            transcriptPath,
            // 中文注释：导入模式下允许 scanner 首次回放 transcript 全量内容，补齐 Codex 客户端里已有但 Hapi 还未看到的消息。
            replayExistingHistory,
            onReplayComplete: () => {
                finalizeTranscriptConversion(true);
            },
            onSessionId: (sessionId) => {
                if (!isPrimarySessionId(sessionId)) {
                    logger.debug(`[codex-local]: Ignoring transcript session id ${sessionId}; primary is ${primarySessionId}`);
                    return;
                }
                session.onSessionFound(sessionId);
            },
            onEvent: (event, context) => {
                const observedModel = extractTurnContextModel(event);
                if (observedModel !== undefined) {
                    transcriptModel = observedModel;
                }
                const observedReasoningEffort = extractTurnContextReasoningEffort(event);
                if (observedReasoningEffort !== undefined) {
                    session.setModelReasoningEffort(observedReasoningEffort);
                }
                dispatchTranscriptActions(convertTranscriptEvent(event), context);
                if (event.type === 'event_msg' && event.payload && typeof event.payload === 'object') {
                    const payload = event.payload as Record<string, unknown>;
                    const timestamp = typeof event.timestamp === 'string' ? Date.parse(event.timestamp) : NaN;
                    const turnId = typeof payload.turn_id === 'string' ? payload.turn_id : null;
                    if (!context.replayedHistory) {
                        if (payload.type === 'task_started') roundDuration.start(turnId, timestamp);
                        if (payload.type === 'task_complete' || payload.type === 'turn_aborted' || payload.type === 'task_failed') {
                            const duration = roundDuration.finish(turnId, timestamp, payload.type === 'task_complete');
                            if (duration) session.sendAgentMessage(duration);
                        }
                    }
                    if (payload.type === 'task_started' || payload.type === 'user_message'
                        || payload.type === 'turn_aborted' || payload.type === 'task_failed') {
                        cancelReady();
                    } else if (payload.type === 'task_complete' && typeof payload.turn_id === 'string') {
                        if (completedTurns.has(payload.turn_id)) return;
                        completedTurns.add(payload.turn_id);
                        if (context.replayedHistory || shuttingDown) return;
                        cancelReady();
                        // Finish forwarding this scan before notifying. A following turn in
                        // the same batch cancels the alert; replay and shutdown never alert.
                        readyTimer = setTimeout(() => {
                            readyTimer = null;
                            if (!shuttingDown && session.queue.size() === 0) {
                                session.sendSessionEvent({ type: 'ready' });
                            }
                        }, 0);
                    }
                }
            }
        });
        if (shuttingDown) {
            await drainAndCleanupScanner(createdScanner, replayExistingHistory);
            return;
        }
        scanner = createdScanner;
        scannerTranscriptPath = transcriptPath;
        scannerReplayedExistingHistory = replayExistingHistory;
    };

    const handleTranscriptPath = (transcriptPath: string): Promise<void> => {
        const setupTask = (pendingScannerSetup ?? Promise.resolve()).then(() => processTranscriptPath(transcriptPath));
        const observedTask = setupTask.catch((error) => {
            if (!shuttingDown) {
                reportTranscriptSyncFailure(transcriptPath, error);
            }
        });
        pendingScannerSetup = observedTask.finally(() => {
            if (pendingScannerSetup === observedTask) {
                pendingScannerSetup = null;
            }
        });
        return pendingScannerSetup;
    };

    const handleSessionHook = (sessionId: string, data: Record<string, unknown>): void => {
        if (shuttingDown) {
            return;
        }

        const transcriptPath = typeof data.transcript_path === 'string' && data.transcript_path.length > 0
            ? data.transcript_path
            : null;
        const hookSource = typeof data.source === 'string' ? data.source : null;
        const shouldAllowSessionSwitch = hookSource === 'clear';

        if (transcriptPath) {
            const activeLocator = transcriptLocator;
            transcriptLocator = null;
            void activeLocator?.cleanup();
        }

        if (!transcriptPath) {
            handleSessionFound(sessionId, shouldAllowSessionSwitch);
            return;
        }

        bindPrimarySession(sessionId, transcriptPath, shouldAllowSessionSwitch);
    };

    const hookServer = await startHookServer({
        onSessionHook: (sessionId, data) => {
            if (shuttingDown) {
                return;
            }
            if (isCodexToolHookEvent(data)) {
                if (primarySessionId && primarySessionId !== sessionId) {
                    return;
                }
                for (const message of toolHookBridge.handle(data)) {
                    session.sendAgentMessage({ ...message, turn_id: data.turn_id });
                }
                return;
            }
            handleSessionHook(sessionId, data);
        }
    });
    logger.debug(`[codex-local]: Started Codex SessionStart hook server on port ${hookServer.port}`);

    if (session.client.isPending()) {
        const createdLocator = createCodexTranscriptLocator({
            cwd: effectiveCodexCwd,
            startupTimestampMs: Date.now(),
            resumeSessionId,
            onLocated: ({ sessionId, transcriptPath }) => {
                if (shuttingDown || hookReady || primaryTranscriptPath) {
                    return;
                }
                transcriptLocator = null;
                bindPrimarySession(sessionId, transcriptPath);
            },
            onAmbiguous: (paths) => {
                transcriptLocator = null;
                logger.warn(`[codex-local]: Transcript fallback was ambiguous (${paths.length} active candidates)`);
            }
        });
        transcriptLocator = createdLocator;
        await createdLocator.ready;
    }

    const launcher = new BaseLocalLauncher({
        label: 'codex-local',
        failureLabel: 'Local Codex process failed',
        queue: session.queue,
        rpcHandlerManager: session.client.rpcHandlerManager,
        startedBy: session.startedBy,
        startingMode: session.startingMode,
        launch: async (abortSignal) => {
            await codexLocal({
                path: session.path,
                sessionId: resumeSessionId,
                modelReasoningEffort: (session.getModelReasoningEffort() ?? undefined) as ReasoningEffort | undefined,
                onSessionFound: handleSessionFound,
                abort: abortSignal,
                codexArgs,
                mcpServers,
                sessionHook: {
                    port: hookServer.port,
                    token: hookServer.token
                }
            });
        },
        sendFailureMessage: (message) => {
            session.sendSessionEvent({ type: 'message', message });
        },
        recordLocalLaunchFailure: (message, exitReason) => {
            session.recordLocalLaunchFailure(message, exitReason);
        },
        abortLogMessage: 'doAbort',
        switchLogMessage: 'doSwitch'
    });

    session.resetTranscriptPath();
    const handleTranscriptPathCallback = (transcriptPath: string) => {
        void handleTranscriptPath(transcriptPath);
    };
    session.addTranscriptPathCallback(handleTranscriptPathCallback);

    try {
        return await launcher.run();
    } finally {
        shuttingDown = true;
        cancelReady();
        session.removeTranscriptPathCallback(handleTranscriptPathCallback);
        hookServer.stop();
        const activeLocator = transcriptLocator;
        transcriptLocator = null;
        void activeLocator?.cleanup();
        if (pendingScannerSetup) {
            await pendingScannerSetup;
        }
        const activeScanner = scanner as CodexSessionScanner | null;
        if (activeScanner) {
            await drainAndCleanupScanner(activeScanner, scannerReplayedExistingHistory);
        }
        flushAllPendingExecWrappers();
        for (const message of toolHookBridge.finish()) {
            session.sendAgentMessage(message);
        }
        flushAllPendingPlans();
        happyServer.stop();
        if (!hookReady) {
            logger.debug('[codex-local]: SessionStart hook did not provide transcript path before shutdown');
        }
        logger.debug('[codex-local]: Stopped hapi MCP bridge server');
    }
}