From 92958890de9d03e84e6b2bbf1cb0b185177230db Mon Sep 17 00:00:00 2001 From: weishu Date: Tue, 28 Jul 2026 12:17:38 +0800 Subject: [PATCH] fix(codex): preserve transcript sync across mode switches --- cli/src/codex/codexLocalLauncher.test.ts | 59 +++++++++++++++++-- cli/src/codex/codexLocalLauncher.ts | 24 +++++++- cli/src/codex/session.ts | 12 +++- .../codex/utils/codexSessionScanner.test.ts | 21 +++++++ cli/src/codex/utils/codexSessionScanner.ts | 4 ++ .../common/session/BaseSessionScanner.ts | 7 +++ 6 files changed, 118 insertions(+), 9 deletions(-) diff --git a/cli/src/codex/codexLocalLauncher.test.ts b/cli/src/codex/codexLocalLauncher.test.ts index 48c649a7..e2728de4 100644 --- a/cli/src/codex/codexLocalLauncher.test.ts +++ b/cli/src/codex/codexLocalLauncher.test.ts @@ -80,6 +80,7 @@ function createSessionStub( let localLaunchFailure: { message: string; exitReason: 'switch' | 'exit' } | null = null; let sessionId: string | null = null; let transcriptPath: string | null = initialTranscriptPath; + let transcriptHistoryReplayPending = replayTranscriptHistoryOnStart; let modelReasoningEffort: string | null = null; const modelReasoningEffortUpdates: Array = []; const transcriptPathCallbacks: Array<(path: string) => void> = []; @@ -96,7 +97,10 @@ function createSessionStub( startedBy: 'terminal' as const, startingMode: 'local' as const, codexArgs, - replayTranscriptHistoryOnStart, + shouldReplayTranscriptHistory: () => transcriptHistoryReplayPending, + markTranscriptHistoryReplayConsumed: () => { + transcriptHistoryReplayPending = false; + }, client: { isPending: () => pendingClient, rpcHandlerManager: { @@ -628,9 +632,9 @@ describe('codexLocalLauncher', () => { } }); - it('replays existing transcript messages when importing a Codex thread into a new Hapi session', async () => { + it('replays imported transcript history only on the first local attachment', async () => { const transcriptPath = join(tempDir, 'codex-import-transcript.jsonl'); - const { session, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true); + const { session, userMessages, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true); let releaseRunBarrier: (() => void) | undefined; harness.runBarrier = new Promise((resolve) => { releaseRunBarrier = resolve; @@ -640,6 +644,7 @@ describe('codexLocalLauncher', () => { transcriptPath, [ JSON.stringify({ type: 'session_meta', payload: { id: 'codex-thread-import' } }), + JSON.stringify({ type: 'event_msg', payload: { type: 'user_message', message: 'old imported prompt' } }), JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'old imported message' } }) ].join('\n') + '\n' ); @@ -652,14 +657,60 @@ describe('codexLocalLauncher', () => { }); await wait(300); + await appendFile( + transcriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'user_message', message: 'tail before switch' } }) + '\n' + ); + if (releaseRunBarrier) { releaseRunBarrier(); } await launcherPromise; + expect(userMessages).toEqual(['old imported prompt', 'tail before switch']); + expect(agentMessages.filter((message) => ( + message as { message?: string } + ).message === 'old imported message')).toHaveLength(1); + + let releaseSecondRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseSecondRunBarrier = resolve; + }); + + const secondLauncherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[1]?.('codex-thread-import', { + transcript_path: transcriptPath + }); + await wait(300); + + expect(userMessages).toEqual(['old imported prompt', 'tail before switch']); + expect(agentMessages.filter((message) => ( + message as { message?: string } + ).message === 'old imported message')).toHaveLength(1); + + await appendFile( + transcriptPath, + [ + JSON.stringify({ type: 'event_msg', payload: { type: 'user_message', message: 'new local prompt' } }), + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'new local response' } }) + ].join('\n') + '\n' + ); + await wait(700); + + if (releaseSecondRunBarrier) { + releaseSecondRunBarrier(); + } + await secondLauncherPromise; + + expect(userMessages).toEqual(['old imported prompt', 'tail before switch', 'new local prompt']); + expect(agentMessages.filter((message) => ( + message as { message?: string } + ).message === 'old imported message')).toHaveLength(1); expect(agentMessages).toContainEqual({ type: 'message', - message: 'old imported message', + message: 'new local response', id: expect.any(String) }); }); diff --git a/cli/src/codex/codexLocalLauncher.ts b/cli/src/codex/codexLocalLauncher.ts index 9b66e231..f09da448 100644 --- a/cli/src/codex/codexLocalLauncher.ts +++ b/cli/src/codex/codexLocalLauncher.ts @@ -46,6 +46,7 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch let pendingScannerSetup: Promise | null = null; let transcriptLocator: CodexTranscriptLocator | null = null; let scannerTranscriptPath: string | null = null; + let scannerReplayedExistingHistory = false; const pendingPlansByTurnId = new Map(); const pendingExecWrappers = new Map(); const toolHookBridge = new CodexToolHookBridge(); @@ -76,6 +77,21 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch }); }; + const drainAndCleanupScanner = async ( + activeScanner: CodexSessionScanner, + replayedExistingHistory: boolean + ): Promise => { + try { + // Codex can flush its final transcript records after the last watcher tick. + await activeScanner.flush(); + if (replayedExistingHistory) { + session.markTranscriptHistoryReplayConsumed(); + } + } finally { + await activeScanner.cleanup(); + } + }; + 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}`); @@ -174,10 +190,11 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch scannerTranscriptPath = transcriptPath; return; } + const replayExistingHistory = session.shouldReplayTranscriptHistory(); const createdScanner = await createCodexSessionScanner({ transcriptPath, // 中文注释：导入模式下允许 scanner 首次回放 transcript 全量内容，补齐 Codex 客户端里已有但 Hapi 还未看到的消息。 - replayExistingHistory: session.replayTranscriptHistoryOnStart, + replayExistingHistory, onSessionId: (sessionId) => { if (!isPrimarySessionId(sessionId)) { logger.debug(`[codex-local]: Ignoring transcript session id ${sessionId}; primary is ${primarySessionId}`); @@ -237,11 +254,12 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch } }); if (shuttingDown) { - await createdScanner.cleanup(); + await drainAndCleanupScanner(createdScanner, replayExistingHistory); return; } scanner = createdScanner; scannerTranscriptPath = transcriptPath; + scannerReplayedExistingHistory = replayExistingHistory; }; const handleTranscriptPath = (transcriptPath: string): Promise => { @@ -376,7 +394,7 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch } const activeScanner = scanner as CodexSessionScanner | null; if (activeScanner) { - await activeScanner.cleanup(); + await drainAndCleanupScanner(activeScanner, scannerReplayedExistingHistory); } flushAllPendingExecWrappers(); for (const message of toolHookBridge.finish()) { diff --git a/cli/src/codex/session.ts b/cli/src/codex/session.ts index ae9b133e..7f6a4e9f 100644 --- a/cli/src/codex/session.ts +++ b/cli/src/codex/session.ts @@ -17,11 +17,11 @@ export class CodexSession extends AgentSessionBase { readonly codexCliOverrides?: CodexCliOverrides; readonly startedBy: 'runner' | 'terminal'; readonly startingMode: 'local' | 'remote'; - readonly replayTranscriptHistoryOnStart: boolean; readonly sourceSessionId?: string; localLaunchFailure: LocalLaunchFailure | null = null; private transcriptPathCallbacks: Array<(path: string) => void> = []; + private transcriptHistoryReplayPending: boolean; constructor(opts: { api: ApiClient; @@ -68,7 +68,7 @@ export class CodexSession extends AgentSessionBase { this.codexCliOverrides = opts.codexCliOverrides; this.startedBy = opts.startedBy; this.startingMode = opts.startingMode; - this.replayTranscriptHistoryOnStart = opts.replayTranscriptHistoryOnStart ?? false; + this.transcriptHistoryReplayPending = opts.replayTranscriptHistoryOnStart ?? false; this.sourceSessionId = opts.sourceSessionId; this.permissionMode = opts.permissionMode; this.model = opts.model; @@ -76,6 +76,14 @@ export class CodexSession extends AgentSessionBase { this.collaborationMode = opts.collaborationMode; } + shouldReplayTranscriptHistory(): boolean { + return this.transcriptHistoryReplayPending; + } + + markTranscriptHistoryReplayConsumed(): void { + this.transcriptHistoryReplayPending = false; + } + onTranscriptPathFound(path: string): void { if (this.transcriptPath === path) { return; diff --git a/cli/src/codex/utils/codexSessionScanner.test.ts b/cli/src/codex/utils/codexSessionScanner.test.ts index d78d436b..06c205c1 100644 --- a/cli/src/codex/utils/codexSessionScanner.test.ts +++ b/cli/src/codex/utils/codexSessionScanner.test.ts @@ -59,6 +59,27 @@ describe('codexSessionScanner', () => { expect(events[0]?.type).toBe('event_msg'); }); + it('flushes an appended tail without waiting for the watcher', async () => { + await writeFile( + transcriptPath, + JSON.stringify({ type: 'session_meta', payload: { id: 'session-flush' } }) + '\n' + ); + + scanner = await createCodexSessionScanner({ + transcriptPath, + onEvent: (event) => events.push(event) + }); + + await appendFile( + transcriptPath, + JSON.stringify({ type: 'event_msg', payload: { type: 'agent_message', message: 'final tail' } }) + '\n' + ); + await scanner.flush(); + + expect(events).toHaveLength(1); + expect(events[0]?.payload).toEqual({ type: 'agent_message', message: 'final tail' }); + }); + it('reads exactly the requested transcript byte range', async () => { const initial = 'existing transcript\n'; const appended = 'new event\n'; diff --git a/cli/src/codex/utils/codexSessionScanner.ts b/cli/src/codex/utils/codexSessionScanner.ts index a593e813..4c98af00 100644 --- a/cli/src/codex/utils/codexSessionScanner.ts +++ b/cli/src/codex/utils/codexSessionScanner.ts @@ -11,6 +11,7 @@ interface CodexSessionScannerOptions { } export interface CodexSessionScanner { + flush: () => Promise; cleanup: () => Promise; setTranscriptPath: (transcriptPath: string) => Promise; } @@ -20,6 +21,9 @@ export async function createCodexSessionScanner(opts: CodexSessionScannerOptions await scanner.start(); return { + flush: async () => { + await scanner.flush(); + }, cleanup: async () => { await scanner.cleanup(); }, diff --git a/cli/src/modules/common/session/BaseSessionScanner.ts b/cli/src/modules/common/session/BaseSessionScanner.ts index e19d0e75..2f75aeca 100644 --- a/cli/src/modules/common/session/BaseSessionScanner.ts +++ b/cli/src/modules/common/session/BaseSessionScanner.ts @@ -108,6 +108,13 @@ export abstract class BaseSessionScanner { this.intervalId = setInterval(() => this.sync.invalidate(), this.options.intervalMs); } + public async flush(): Promise { + if (this.stopped) { + return; + } + await this.sync.invalidateAndAwait(); + } + public async cleanup(): Promise { this.stopped = true; if (this.intervalId) {