feat(codex): import local Codex sessions into Hapi (#796)

* local: add Codex Desktop session sync controls * feat(codex): import local Codex sessions into Hapi --------- Co-authored-by: Codex Local <codex-local@example.invalid>
2026-08-05 06:24:37 +00:00 · 2026-06-04 17:53:12 +08:00
parent 39fba5292c
commit f9ef3a4489
25 changed files with 3156 additions and 15 deletions
@@ -69,9 +69,11 @@ function createSessionStub(
    permissionMode: 'default' | 'read-only' | 'safe-yolo' | 'yolo',
    codexArgs?: string[],
    path = '/tmp/worktree',
-    initialTranscriptPath: string | null = null
+    initialTranscriptPath: string | null = null,
+    replayTranscriptHistoryOnStart = false
 ) {
    const sessionEvents: Array<{ type: string; message?: string }> = [];
+    const userMessages: string[] = [];
    const agentMessages: unknown[] = [];
    let localLaunchFailure: { message: string; exitReason: 'switch' | 'exit' } | null = null;
    let sessionId: string | null = null;
@@ -90,6 +92,7 @@ function createSessionStub(
            startedBy: 'terminal' as const,
            startingMode: 'local' as const,
            codexArgs,
+            replayTranscriptHistoryOnStart,
            client: {
                rpcHandlerManager: {
                    registerHandler: () => {}
@@ -124,13 +127,16 @@ function createSessionStub(
            recordLocalLaunchFailure: (message: string, exitReason: 'switch' | 'exit') => {
                localLaunchFailure = { message, exitReason };
            },
-            sendUserMessage: () => {},
+            sendUserMessage: (message: string) => {
+                userMessages.push(message);
+            },
            sendAgentMessage: (message: unknown) => {
                agentMessages.push(message);
            },
            queue: createQueueStub()
        },
        sessionEvents,
+        userMessages,
        agentMessages,
        getLocalLaunchFailure: () => localLaunchFailure
    };
@@ -348,6 +354,94 @@ describe('codexLocalLauncher', () => {
        });
    });

+    it('replays existing transcript messages when importing a Codex thread into a new Hapi session', async () => {
+        const transcriptPath = join(tempDir, 'codex-import-transcript.jsonl');
+        const { session, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true);
+        let releaseRunBarrier: (() => void) | undefined;
+        harness.runBarrier = new Promise((resolve) => {
+            releaseRunBarrier = resolve;
+        });
+
+        await writeFile(
+            transcriptPath,
+            [
+                JSON.stringify({ type: 'session_meta', payload: { id: 'codex-thread-import' } }),
+                JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'old imported message' } })
+            ].join('\n') + '\n'
+        );
+
+        const launcherPromise = codexLocalLauncher(session as never);
+        await wait(50);
+
+        harness.sessionHookHandlers[0]?.('codex-thread-import', {
+            transcript_path: transcriptPath
+        });
+        await wait(300);
+
+        if (releaseRunBarrier) {
+            releaseRunBarrier();
+        }
+        await launcherPromise;
+
+        expect(agentMessages).toContainEqual({
+            type: 'message',
+            message: 'old imported message',
+            id: expect.any(String)
+        });
+    });
+
+    it('replays existing response_item chat messages when importing a Codex thread into a new Hapi session', async () => {
+        const transcriptPath = join(tempDir, 'codex-import-response-item-transcript.jsonl');
+        const { session, userMessages, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true);
+        let releaseRunBarrier: (() => void) | undefined;
+        harness.runBarrier = new Promise((resolve) => {
+            releaseRunBarrier = resolve;
+        });
+
+        await writeFile(
+            transcriptPath,
+            [
+                JSON.stringify({ type: 'session_meta', payload: { id: 'codex-thread-import-response-item' } }),
+                JSON.stringify({
+                    type: 'response_item',
+                    payload: {
+                        type: 'message',
+                        role: 'user',
+                        content: [{ type: 'input_text', text: 'old response_item user message' }]
+                    }
+                }),
+                JSON.stringify({
+                    type: 'response_item',
+                    payload: {
+                        type: 'message',
+                        role: 'assistant',
+                        content: [{ type: 'output_text', text: 'old response_item assistant message' }]
+                    }
+                })
+            ].join('\n') + '\n'
+        );
+
+        const launcherPromise = codexLocalLauncher(session as never);
+        await wait(50);
+
+        harness.sessionHookHandlers[0]?.('codex-thread-import-response-item', {
+            transcript_path: transcriptPath
+        });
+        await wait(300);
+
+        if (releaseRunBarrier) {
+            releaseRunBarrier();
+        }
+        await launcherPromise;
+
+        expect(userMessages).toContain('old response_item user message');
+        expect(agentMessages).toContainEqual({
+            type: 'message',
+            message: 'old response_item assistant message',
+            id: expect.any(String)
+        });
+    });
+
    it('does not let a later non-clear hook replace the primary session', async () => {
        const primaryTranscriptPath = await writeTranscriptMeta('primary-later-hook.jsonl', 'primary-thread');
        const otherTranscriptPath = await writeTranscriptMeta('later-other-transcript.jsonl', 'other-thread');
@@ -83,6 +83,8 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch
        }
        const createdScanner = await createCodexSessionScanner({
            transcriptPath,
+            // 中文注释：导入模式下允许 scanner 首次回放 transcript 全量内容，补齐 Codex 客户端里已有但 Hapi 还未看到的消息。
+            replayExistingHistory: session.replayTranscriptHistoryOnStart,
            onSessionId: (sessionId) => {
                if (!isPrimarySessionId(sessionId)) {
                    logger.debug(`[codex-local]: Ignoring transcript session id ${sessionId}; primary is ${primarySessionId}`);
@@ -33,6 +33,7 @@ interface LoopOptions {
    modelReasoningEffort?: ReasoningEffort;
    collaborationMode?: CodexCollaborationMode;
    resumeSessionId?: string;
+    replayTranscriptHistoryOnStart?: boolean;
    onSessionReady?: (session: CodexSession) => void;
 }

@@ -56,7 +57,8 @@ export async function loop(opts: LoopOptions): Promise<void> {
        permissionMode: opts.permissionMode ?? 'default',
        model: opts.model,
        modelReasoningEffort: opts.modelReasoningEffort,
-        collaborationMode: opts.collaborationMode ?? 'default'
+        collaborationMode: opts.collaborationMode ?? 'default',
+        replayTranscriptHistoryOnStart: opts.replayTranscriptHistoryOnStart ?? false
    });

    await runLocalRemoteSession({
@@ -134,8 +134,21 @@ describe('runCodex', () => {
        }))
        expect(harness.loopArgs[0]).toEqual(expect.objectContaining({
            resumeSessionId: 'codex-thread-1',
-            collaborationMode: 'plan'
+            collaborationMode: 'plan',
+            replayTranscriptHistoryOnStart: false
        }))
        expect(mockCodexSession.setCollaborationMode).toHaveBeenLastCalledWith('plan')
    })
+
+    it('replays transcript history when attaching a new Hapi session to an existing Codex thread', async () => {
+        await runCodexImpl({
+            workingDirectory: '/tmp/project',
+            resumeSessionId: 'codex-thread-2'
+        })
+
+        expect(harness.loopArgs[0]).toEqual(expect.objectContaining({
+            resumeSessionId: 'codex-thread-2',
+            replayTranscriptHistoryOnStart: true
+        }))
+    })
 })
@@ -73,6 +73,9 @@ export async function runCodex(opts: {

    const codexCliOverrides = parseCodexCliOverrides(opts.codexArgs);
    const sessionWrapperRef: { current: CodexSession | null } = { current: null };
+    // 中文注释：当用户直接把现成的 Codex thread 导入到一个全新的 Hapi 会话时，
+    // 需要在首次附着 transcript 时回放已有历史；恢复已有 Hapi 会话时则保持原来的增量模式，避免重复灌入旧消息。
+    const replayTranscriptHistoryOnStart = Boolean(opts.resumeSessionId && !opts.existingSessionId);

    let currentPermissionMode: PermissionMode = opts.permissionMode ?? 'default';
    let currentModel = opts.model;
@@ -353,6 +356,7 @@ export async function runCodex(opts: {
            modelReasoningEffort: currentModelReasoningEffort,
            collaborationMode: currentCollaborationMode,
            resumeSessionId: opts.resumeSessionId,
+            replayTranscriptHistoryOnStart,
            onModeChange: createModeChangeHandler(session),
            onSessionReady: (instance) => {
                sessionWrapperRef.current = instance;
@@ -17,6 +17,7 @@ export class CodexSession extends AgentSessionBase<EnhancedMode> {
    readonly codexCliOverrides?: CodexCliOverrides;
    readonly startedBy: 'runner' | 'terminal';
    readonly startingMode: 'local' | 'remote';
+    readonly replayTranscriptHistoryOnStart: boolean;
    localLaunchFailure: LocalLaunchFailure | null = null;

    private transcriptPathCallbacks: Array<(path: string) => void> = [];
@@ -38,6 +39,7 @@ export class CodexSession extends AgentSessionBase<EnhancedMode> {
        model?: SessionModel;
        modelReasoningEffort?: SessionModelReasoningEffort;
        collaborationMode?: EnhancedMode['collaborationMode'];
+        replayTranscriptHistoryOnStart?: boolean;
    }) {
        super({
            api: opts.api,
@@ -64,6 +66,7 @@ export class CodexSession extends AgentSessionBase<EnhancedMode> {
        this.codexCliOverrides = opts.codexCliOverrides;
        this.startedBy = opts.startedBy;
        this.startingMode = opts.startingMode;
+        this.replayTranscriptHistoryOnStart = opts.replayTranscriptHistoryOnStart ?? false;
        this.permissionMode = opts.permissionMode;
        this.model = opts.model;
        this.modelReasoningEffort = opts.modelReasoningEffort;
@@ -32,6 +32,37 @@ describe('convertCodexEvent', () => {
        expect(result?.userMessage).toBe('hello user');
    });

+    it('converts response_item user messages', () => {
+        const result = convertCodexEvent({
+            type: 'response_item',
+            payload: {
+                type: 'message',
+                role: 'user',
+                content: [{ type: 'input_text', text: 'hello from response_item user' }]
+            }
+        });
+
+        expect(result).toEqual({
+            userMessage: 'hello from response_item user'
+        });
+    });
+
+    it('converts response_item assistant messages', () => {
+        const result = convertCodexEvent({
+            type: 'response_item',
+            payload: {
+                type: 'message',
+                role: 'assistant',
+                content: [{ type: 'output_text', text: 'hello from response_item assistant' }]
+            }
+        });
+
+        expect(result?.message).toMatchObject({
+            type: 'message',
+            message: 'hello from response_item assistant'
+        });
+    });
+
    it('converts reasoning events', () => {
        const result = convertCodexEvent({
            type: 'event_msg',
@@ -55,6 +55,30 @@ function asString(value: unknown): string | null {
    return typeof value === 'string' && value.length > 0 ? value : null;
 }

+function extractCodexText(value: unknown): string {
+    if (typeof value === 'string') {
+        return value.trim();
+    }
+    if (Array.isArray(value)) {
+        return value
+            .map((item) => {
+                const record = asRecord(item);
+                if (record?.type === 'input_text' && typeof record.text === 'string') return record.text;
+                if (record?.type === 'output_text' && typeof record.text === 'string') return record.text;
+                if (record?.type === 'text' && typeof record.text === 'string') return record.text;
+                return null;
+            })
+            .filter((part): part is string => Boolean(part))
+            .join(' ')
+            .trim();
+    }
+    const record = asRecord(value);
+    if (record?.type === 'input_text' && typeof record.text === 'string') return record.text.trim();
+    if (record?.type === 'output_text' && typeof record.text === 'string') return record.text.trim();
+    if (record?.type === 'text' && typeof record.text === 'string') return record.text.trim();
+    return '';
+}
+
 function parseArguments(value: unknown): unknown {
    if (typeof value !== 'string') {
        return value;
@@ -194,6 +218,27 @@ export function convertCodexEvent(rawEvent: unknown): CodexConversionResult | nu
            return null;
        }

+        if (itemType === 'message') {
+            const role = asString(payloadRecord.role);
+            const text = extractCodexText(payloadRecord.content);
+            if (!text) {
+                return null;
+            }
+            if (role === 'user') {
+                return { userMessage: text };
+            }
+            if (role === 'assistant') {
+                return {
+                    message: {
+                        type: 'message',
+                        message: text,
+                        id: randomUUID()
+                    }
+                };
+            }
+            return null;
+        }
+
        if (itemType === 'function_call') {
            const name = asString(payloadRecord.name);
            const callId = extractCallId(payloadRecord);
@@ -59,6 +59,27 @@ describe('codexSessionScanner', () => {
        expect(events[0]?.type).toBe('event_msg');
    });

+    it('can replay existing transcript history on first attach', async () => {
+        await writeFile(
+            transcriptPath,
+            [
+                JSON.stringify({ type: 'session_meta', payload: { id: 'session-replay' } }),
+                JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'old' } })
+            ].join('\n') + '\n'
+        );
+
+        scanner = await createCodexSessionScanner({
+            transcriptPath,
+            replayExistingHistory: true,
+            onEvent: (event) => events.push(event)
+        });
+
+        await wait(300);
+        expect(events).toHaveLength(2);
+        expect(events[0]?.type).toBe('session_meta');
+        expect(events[1]?.payload).toEqual({ type: 'agent_message', message: 'old' });
+    });
+
    it('reports session id from the transcript metadata', async () => {
        await writeFile(
            transcriptPath,
@@ -7,6 +7,7 @@ interface CodexSessionScannerOptions {
    transcriptPath: string | null;
    onEvent: (event: CodexSessionEvent) => void;
    onSessionId?: (sessionId: string) => void;
+    replayExistingHistory?: boolean;
 }

 export interface CodexSessionScanner {
@@ -34,6 +35,7 @@ class CodexSessionScannerImpl extends BaseSessionScanner<CodexSessionEvent> {
    private readonly onSessionId?: (sessionId: string) => void;
    private readonly fileEpochByPath = new Map<string, number>();
    private readonly fileSizeByPath = new Map<string, number>();
+    private replayExistingHistoryOnNextAttach: boolean;
    private observedSessionId: string | null = null;

    constructor(opts: CodexSessionScannerOptions) {
@@ -41,6 +43,7 @@ class CodexSessionScannerImpl extends BaseSessionScanner<CodexSessionEvent> {
        this.transcriptPath = opts.transcriptPath;
        this.onEvent = opts.onEvent;
        this.onSessionId = opts.onSessionId;
+        this.replayExistingHistoryOnNextAttach = opts.replayExistingHistory ?? false;
    }

    async setTranscriptPath(transcriptPath: string): Promise<void> {
@@ -48,14 +51,14 @@ class CodexSessionScannerImpl extends BaseSessionScanner<CodexSessionEvent> {
            return;
        }
        this.transcriptPath = transcriptPath;
-        await this.primeTranscript(transcriptPath);
+        await this.prepareTranscript(transcriptPath);
        this.pruneWatchers(this.transcriptPath ? [this.transcriptPath] : []);
        this.invalidate();
    }

    protected async initialize(): Promise<void> {
        if (this.transcriptPath) {
-            await this.primeTranscript(this.transcriptPath);
+            await this.prepareTranscript(this.transcriptPath);
        }
    }

@@ -89,6 +92,17 @@ class CodexSessionScannerImpl extends BaseSessionScanner<CodexSessionEvent> {
        this.pruneWatchers(this.transcriptPath ? [this.transcriptPath] : []);
    }

+    private async prepareTranscript(filePath: string): Promise<void> {
+        if (this.replayExistingHistoryOnNextAttach) {
+            // 中文注释：导入既有 Codex thread 时，首次挂接 transcript 不能先 prime 到 EOF，
+            // 否则 Hapi 只会看到后续增量，客户端里已经存在的最新消息会被跳过。
+            this.replayExistingHistoryOnNextAttach = false;
+            return;
+        }
+
+        await this.primeTranscript(filePath);
+    }
+
    private async primeTranscript(filePath: string): Promise<void> {
        const { events, nextCursor } = await this.readSessionFile(filePath, 0);
        const keys = events.map((entry) => this.generateEventKey(entry.event, { filePath, lineIndex: entry.lineIndex }));