From c2cc73159c16f3e47542be2c982a111dc4a14343 Mon Sep 17 00:00:00 2001 From: weishu Date: Mon, 10 Aug 2026 10:14:29 +0800 Subject: [PATCH] fix(codex): preserve transcript final answer ordering --- cli/src/codex/codexLocalLauncher.test.ts | 169 ++++++++ cli/src/codex/codexLocalLauncher.ts | 179 ++++---- .../codex/utils/codexEventConverter.test.ts | 386 +++++++++++++++++- cli/src/codex/utils/codexEventConverter.ts | 239 ++++++++++- .../codex/utils/codexSessionScanner.test.ts | 6 + cli/src/codex/utils/codexSessionScanner.ts | 6 + 6 files changed, 886 insertions(+), 99 deletions(-) diff --git a/cli/src/codex/codexLocalLauncher.test.ts b/cli/src/codex/codexLocalLauncher.test.ts index c11f0423..6d7ad7e1 100644 --- a/cli/src/codex/codexLocalLauncher.test.ts +++ b/cli/src/codex/codexLocalLauncher.test.ts @@ -76,6 +76,15 @@ function createSessionStub( const sessionEvents: Array<{ type: string; message?: string }> = []; const userMessages: string[] = []; const agentMessages: unknown[] = []; + const messageEvents: Array<{ + type: 'user-message'; + message: string; + } | { + type: 'user-activity'; + } | { + type: 'agent-message'; + message: unknown; + }> = []; let userActivityCount = 0; let localLaunchFailure: { message: string; exitReason: 'switch' | 'exit' } | null = null; let sessionId: string | null = null; @@ -142,18 +151,22 @@ function createSessionStub( }, sendUserMessage: (message: string) => { userMessages.push(message); + messageEvents.push({ type: 'user-message', message }); }, notifyUserActivity: () => { userActivityCount += 1; + messageEvents.push({ type: 'user-activity' }); }, sendAgentMessage: (message: unknown) => { agentMessages.push(message); + messageEvents.push({ type: 'agent-message', message }); }, queue: createQueueStub() }, sessionEvents, userMessages, agentMessages, + messageEvents, getUserActivityCount: () => userActivityCount, getLocalLaunchFailure: () => localLaunchFailure, getModelReasoningEffort: () => modelReasoningEffort, @@ -1027,6 +1040,102 @@ describe('codexLocalLauncher', () => { }); }); + it('dispatches buffered finals before later user and semantic actions', async () => { + const transcriptPath = join(tempDir, 'codex-ordered-actions-transcript.jsonl'); + const { session, messageEvents } = 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-ordered' } }), + JSON.stringify({ type: 'turn_context', payload: { turn_id: 'turn-ordered' } }), + JSON.stringify({ + type: 'response_item', + payload: { + type: 'message', + id: 'final-a', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible final A' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-ordered' } + } + }), + JSON.stringify({ + type: 'event_msg', + payload: { type: 'user_message', message: 'queued follow-up' } + }), + JSON.stringify({ + type: 'response_item', + payload: { + type: 'message', + id: 'final-b', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible final B' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-ordered' } + } + }), + JSON.stringify({ + type: 'event_msg', + payload: { + type: 'item_completed', + turn_id: 'turn-ordered', + item: { + type: 'AgentMessage', + id: 'final-b', + phase: 'final_answer', + content: [{ type: 'Text', text: 'visible final B' }] + } + } + }), + JSON.stringify({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-ordered' } + }) + ].join('\n') + '\n' + ); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[0]?.('codex-thread-ordered', { + transcript_path: transcriptPath + }); + await wait(300); + + releaseRunBarrier?.(); + await launcherPromise; + + expect(messageEvents).toEqual([{ + type: 'agent-message', + message: { + type: 'message', + message: 'visible final A', + id: 'final-a' + } + }, { + type: 'user-message', + message: 'queued follow-up' + }, { + type: 'agent-message', + message: { + type: 'message', + message: 'visible final B', + id: 'final-b' + } + }]); + }); + it('replays Codex 0.147 completed messages and response-only final answers once', async () => { const transcriptPath = join(tempDir, 'codex-import-0.147-transcript.jsonl'); const { session, userMessages, agentMessages, getUserActivityCount } = createSessionStub( @@ -1095,6 +1204,10 @@ describe('codexLocalLauncher', () => { content: [{ type: 'output_text', text: 'visible 0.147 final answer' }], internal_chat_message_metadata_passthrough: { turn_id: 'turn-147' } } + }), + JSON.stringify({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-147' } }) ].join('\n') + '\n' ); @@ -1123,6 +1236,62 @@ describe('codexLocalLauncher', () => { }]); }); + it('finalizes a response-only answer when an imported transcript ends without a boundary', async () => { + const transcriptPath = join(tempDir, 'codex-import-final-at-eof.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-final-at-eof' } }), + JSON.stringify({ type: 'turn_context', payload: { turn_id: 'turn-final-at-eof' } }), + JSON.stringify({ + type: 'response_item', + payload: { + type: 'message', + id: 'final-at-eof', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible answer at EOF' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-final-at-eof' } + } + }) + ].join('\n') + '\n' + ); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[0]?.('codex-thread-final-at-eof', { + transcript_path: transcriptPath + }); + await wait(300); + expect(agentMessages).toEqual([{ + type: 'message', + message: 'visible answer at EOF', + id: 'final-at-eof' + }]); + + releaseRunBarrier?.(); + await launcherPromise; + + expect(agentMessages).toEqual([{ + type: 'message', + message: 'visible answer at EOF', + id: 'final-at-eof' + }]); + }); + it('replays a plan-only turn when the turn completes', async () => { const transcriptPath = join(tempDir, 'codex-import-plan-only-transcript.jsonl'); const { session, agentMessages } = createSessionStub('default', undefined, '/tmp/worktree', null, true); diff --git a/cli/src/codex/codexLocalLauncher.ts b/cli/src/codex/codexLocalLauncher.ts index 06ca931b..7cc58645 100644 --- a/cli/src/codex/codexLocalLauncher.ts +++ b/cli/src/codex/codexLocalLauncher.ts @@ -5,7 +5,12 @@ import { codexLocal } from './codexLocal'; import type { ReasoningEffort } from './appServerTypes'; import { CodexSession } from './session'; import { createCodexSessionScanner, type CodexSessionScanner } from './utils/codexSessionScanner'; -import { createCodexEventConverter, type CodexMessage, type CodexSessionEvent } from './utils/codexEventConverter'; +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'; @@ -89,21 +94,6 @@ 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}`); @@ -173,6 +163,100 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch pendingExecWrappers.clear(); }; + 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 message = action.message; + 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, + model: transcriptModel, + threadId: primarySessionId, + thread_id: primarySessionId, + hapiUsageScope: 'managed' + } + : { ...message, model: transcriptModel }; + session.sendAgentMessage(scopedMessage); + } + } + }; + + const finalizeTranscriptConversion = (replayedExistingHistory: boolean): void => { + dispatchTranscriptActions(convertTranscriptEvent.finalize(), { + replayedHistory: replayedExistingHistory + }); + }; + + 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 { + 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}`); @@ -196,6 +280,8 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch } if (scanner) { if (scannerTranscriptPath !== transcriptPath) { + await scanner.flush(); + finalizeTranscriptConversion(scannerReplayedExistingHistory); flushAllPendingPlans(); convertTranscriptEvent = createCodexEventConverter(); } @@ -208,6 +294,9 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch 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}`); @@ -224,63 +313,7 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch if (observedReasoningEffort !== undefined) { session.setModelReasoningEffort(observedReasoningEffort); } - const converted = convertTranscriptEvent(event); - if (converted?.sessionId) { - if (!isPrimarySessionId(converted.sessionId)) { - logger.debug(`[codex-local]: Ignoring converted session id ${converted.sessionId}; primary is ${primarySessionId}`); - return; - } - session.onSessionFound(converted.sessionId); - } - if (converted?.userMessage) { - session.sendUserMessage(converted.userMessage); - } else if (converted?.userActivity) { - session.notifyUserActivity(); - } - for (const message of converted?.messages ?? []) { - 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, - ...(converted?.turnId ? { turnId: converted.turnId } : {}) - }); - } - } else if (message.type === 'tool-call-result' && pendingExecWrappers.has(message.callId)) { - const pending = pendingExecWrappers.get(message.callId); - const turnId = pending?.turnId ?? converted?.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, - model: transcriptModel, - threadId: primarySessionId, - thread_id: primarySessionId, - hapiUsageScope: 'managed' - } - : { ...message, model: transcriptModel }; - session.sendAgentMessage(scopedMessage); - } - } - if (converted?.finishedTurnId) { - flushPendingPlan(converted.finishedTurnId); - for (const message of toolHookBridge.finishTurn(converted.finishedTurnId)) { - session.sendAgentMessage(message); - } - } + dispatchTranscriptActions(convertTranscriptEvent(event), context); } }); if (shuttingDown) { diff --git a/cli/src/codex/utils/codexEventConverter.test.ts b/cli/src/codex/utils/codexEventConverter.test.ts index 9efd3533..de94af45 100644 --- a/cli/src/codex/utils/codexEventConverter.test.ts +++ b/cli/src/codex/utils/codexEventConverter.test.ts @@ -1,5 +1,14 @@ import { describe, expect, it } from 'vitest'; -import { convertCodexEvent, createCodexEventConverter } from './codexEventConverter'; +import { + convertCodexEvent, + createCodexEventConverter, + type CodexConversionAction, + type CodexMessage +} from './codexEventConverter'; + +function getAgentMessages(actions: CodexConversionAction[]): CodexMessage[] { + return actions.flatMap((action) => action.type === 'agent-message' ? [action.message] : []); +} describe('convertCodexEvent', () => { it('extracts session_meta id', () => { @@ -247,14 +256,14 @@ describe('convertCodexEvent', () => { it('deduplicates legacy and 0.147 semantic messages against response items', () => { const convert = createCodexEventConverter(); - convert({ + expect(convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } - }); - expect(convert({ + })).toEqual([]); + expect(getAgentMessages(convert({ type: 'event_msg', payload: { type: 'agent_message', phase: 'commentary', message: 'legacy commentary' } - })?.messages?.[0]).toMatchObject({ message: 'legacy commentary' }); + }))[0]).toMatchObject({ message: 'legacy commentary' }); expect(convert({ type: 'response_item', payload: { @@ -265,9 +274,9 @@ describe('convertCodexEvent', () => { content: [{ type: 'output_text', text: 'legacy commentary' }], internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } } - })).toBeNull(); + })).toEqual([]); - expect(convert({ + expect(getAgentMessages(convert({ type: 'event_msg', payload: { type: 'item_completed', @@ -279,7 +288,7 @@ describe('convertCodexEvent', () => { content: [{ type: 'Text', text: 'new commentary' }] } } - })?.messages?.[0]).toMatchObject({ message: 'new commentary' }); + }))[0]).toMatchObject({ message: 'new commentary' }); expect(convert({ type: 'response_item', payload: { @@ -290,7 +299,7 @@ describe('convertCodexEvent', () => { content: [{ type: 'output_text', text: 'new commentary' }], internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } } - })).toBeNull(); + })).toEqual([]); expect(convert({ type: 'response_item', @@ -302,7 +311,364 @@ describe('convertCodexEvent', () => { content: [{ type: 'output_text', text: 'response-only final' }], internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } } - })?.messages?.[0]).toMatchObject({ message: 'response-only final' }); + })).toEqual([]); + const completed = convert({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-1' } + }); + expect(getAgentMessages(completed)).toEqual([ + expect.objectContaining({ message: 'response-only final' }) + ]); + expect(completed.at(-1)).toEqual({ type: 'turn-finished', turnId: 'turn-1' }); + }); + + it('flushes older finals before emitting a later semantic mirror', () => { + const convert = createCodexEventConverter(); + + convert({ + type: 'turn_context', + payload: { turn_id: 'turn-1' } + }); + expect(convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'final-a', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible final A' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + })).toEqual([]); + expect(convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'final-b', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible final B' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + })).toEqual([]); + + const mirrored = convert({ + type: 'event_msg', + payload: { + type: 'item_completed', + turn_id: 'turn-1', + item: { + type: 'AgentMessage', + id: 'final-b', + phase: 'final_answer', + content: [{ type: 'Text', text: 'visible final B' }] + } + } + }); + expect(getAgentMessages(mirrored)).toEqual([{ + type: 'message', + message: 'visible final A', + id: 'final-a' + }, { + type: 'message', + message: 'visible final B', + id: 'final-b' + }]); + expect(convert({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-1' } + })).toEqual([{ type: 'turn-finished', turnId: 'turn-1' }]); + }); + + it('does not register a semantic message after pairing it with a buffered response', () => { + const convert = createCodexEventConverter(); + + convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } }); + convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'paired-final', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'same visible text' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + }); + expect(getAgentMessages(convert({ + type: 'event_msg', + payload: { + type: 'item_completed', + turn_id: 'turn-1', + item: { + type: 'AgentMessage', + id: 'paired-final', + phase: 'final_answer', + content: [{ type: 'Text', text: 'same visible text' }] + } + } + }))).toEqual([{ + type: 'message', + message: 'same visible text', + id: 'paired-final' + }]); + + expect(convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'response-only-final', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'same visible text' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + })).toEqual([]); + expect(getAgentMessages(convert({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-1' } + }))).toEqual([{ + type: 'message', + message: 'same visible text', + id: 'response-only-final' + }]); + }); + + it('finalizes a response-only answer when the event stream ends', () => { + const convert = createCodexEventConverter(); + + convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } }); + expect(convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'final-at-eof', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible answer at EOF' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + })).toEqual([]); + + expect(getAgentMessages(convert.finalize())).toEqual([{ + type: 'message', + message: 'visible answer at EOF', + id: 'final-at-eof' + }]); + expect(convert.finalize()).toEqual([]); + }); + + it('flushes a pending final before subsequent user activity', () => { + const convert = createCodexEventConverter(); + + convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } }); + expect(convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'visible-final', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible final answer' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + })).toEqual([]); + + expect(convert({ + type: 'event_msg', + payload: { type: 'user_message', message: 'queued follow-up' } + })).toEqual([{ + type: 'agent-message', + turnId: 'turn-1', + message: { + type: 'message', + message: 'visible final answer', + id: 'visible-final' + } + }, { + type: 'user-message', + message: 'queued follow-up' + }]); + }); + + it('flushes a pending final before subsequent tool activity', () => { + const convert = createCodexEventConverter(); + + convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } }); + convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'visible-final', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible final answer' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + }); + + expect(getAgentMessages(convert({ + type: 'response_item', + payload: { + type: 'function_call', + name: 'ReadFile', + call_id: 'call-read', + arguments: '{"path":"README.md"}' + } + }))).toEqual([{ + type: 'message', + message: 'visible final answer', + id: 'visible-final' + }, expect.objectContaining({ + type: 'tool-call', + name: 'ReadFile', + callId: 'call-read' + })]); + }); + + it('drops only a confirmed compaction summary and flushes earlier finals', () => { + const convert = createCodexEventConverter(); + + convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } }); + convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'earlier-visible-final', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'earlier visible final answer' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + }); + convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'compaction-summary', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'internal context summary for the next model' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + }); + + expect(getAgentMessages(convert({ + type: 'compacted', + payload: { + message: 'Compaction checkpoint prefix\ninternal context summary for the next model' + } + }))).toEqual([{ + type: 'message', + message: 'earlier visible final answer', + id: 'earlier-visible-final' + }]); + expect(convert({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-1' } + })).toEqual([{ type: 'turn-finished', turnId: 'turn-1' }]); + }); + + it('matches an indented compaction summary after the checkpoint prefix', () => { + const convert = createCodexEventConverter(); + + convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } }); + convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'indented-compaction-summary', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: ' internal context summary' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + }); + + expect(convert({ + type: 'compacted', + payload: { message: 'Compaction checkpoint prefix\n internal context summary' } + })).toEqual([]); + expect(convert({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-1' } + })).toEqual([{ type: 'turn-finished', turnId: 'turn-1' }]); + }); + + it('does not let token counts confirm a pending final', () => { + const convert = createCodexEventConverter(); + + convert({ type: 'turn_context', payload: { turn_id: 'turn-1' } }); + convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'visible-final', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible final answer' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + }); + + const tokenCount = convert({ + type: 'event_msg', + payload: { type: 'token_count', info: { total_token_usage: {} } } + }); + expect(getAgentMessages(tokenCount)).toEqual([ + expect.objectContaining({ type: 'token_count' }) + ]); + + const completed = convert({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-1' } + }); + expect(getAgentMessages(completed)).toEqual([{ + type: 'message', + message: 'visible final answer', + id: 'visible-final' + }]); + }); + + it('keeps a pending visible final when remote compaction has no raw summary item', () => { + const convert = createCodexEventConverter(); + + convert({ + type: 'turn_context', + payload: { turn_id: 'turn-1' } + }); + expect(convert({ + type: 'response_item', + payload: { + type: 'message', + id: 'visible-before-remote-compaction', + role: 'assistant', + phase: 'final_answer', + content: [{ type: 'output_text', text: 'visible before remote compaction' }], + internal_chat_message_metadata_passthrough: { turn_id: 'turn-1' } + } + })).toEqual([]); + expect(convert({ + type: 'compacted', + payload: { + message: '', + replacement_history: [{ + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'remote checkpoint summary' }] + }] + } + })).toEqual([]); + const completed = convert({ + type: 'event_msg', + payload: { type: 'task_complete', turn_id: 'turn-1' } + }); + expect(getAgentMessages(completed)).toEqual([{ + type: 'message', + message: 'visible before remote compaction', + id: 'visible-before-remote-compaction' + }]); + expect(completed.at(-1)).toEqual({ type: 'turn-finished', turnId: 'turn-1' }); }); it('converts reasoning events', () => { diff --git a/cli/src/codex/utils/codexEventConverter.ts b/cli/src/codex/utils/codexEventConverter.ts index 3c18dfa0..04df3cb6 100644 --- a/cli/src/codex/utils/codexEventConverter.ts +++ b/cli/src/codex/utils/codexEventConverter.ts @@ -47,7 +47,7 @@ export type CodexMessage = { is_error?: boolean; }; -export type CodexConversionResult = { +export type CodexEventProjection = { sessionId?: string; turnId?: string; messages?: CodexMessage[]; @@ -56,6 +56,28 @@ export type CodexConversionResult = { finishedTurnId?: string; }; +export type CodexConversionAction = { + type: 'session-found'; + sessionId: string; +} | { + type: 'user-message'; + message: string; +} | { + type: 'user-activity'; +} | { + type: 'agent-message'; + message: CodexMessage; + turnId?: string; +} | { + type: 'turn-finished'; + turnId: string; +}; + +export interface CodexEventConverter { + (rawEvent: unknown): CodexConversionAction[]; + finalize: () => CodexConversionAction[]; +} + function asRecord(value: unknown): Record | null { if (!value || typeof value !== 'object') { return null; @@ -150,6 +172,15 @@ type AssistantMessageProjection = { source: 'semantic' | 'response'; text: string; turnId: string | null; + itemId: string | null; +}; + +type PendingResponseFinal = { + key: string; + text: string; + itemId: string | null; + turnId: string | null; + messages: CodexMessage[]; }; function extractEventTurnId(event: CodexSessionEvent): string | null { @@ -173,7 +204,8 @@ function extractAssistantMessageProjection( return { source: 'semantic', text, - turnId: extractEventTurnId(event) ?? currentTurnId + turnId: extractEventTurnId(event) ?? currentTurnId, + itemId: asString(payload.id) }; } @@ -185,7 +217,8 @@ function extractAssistantMessageProjection( return { source: 'semantic', text, - turnId: extractEventTurnId(event) ?? currentTurnId + turnId: extractEventTurnId(event) ?? currentTurnId, + itemId: asString(item?.id) }; } @@ -195,7 +228,8 @@ function extractAssistantMessageProjection( return { source: 'response', text, - turnId: extractEventTurnId(event) ?? currentTurnId + turnId: extractEventTurnId(event) ?? currentTurnId, + itemId: asString(payload.id) }; } @@ -214,50 +248,223 @@ function rememberProjection(counts: Map, key: string): void { counts.set(key, (counts.get(key) ?? 0) + 1); } +function isFinalAnswerResponse(event: CodexSessionEvent): boolean { + const payload = asRecord(event.payload); + return event.type === 'response_item' + && payload?.type === 'message' + && payload.role === 'assistant' + && normalizeItemType(payload.phase) === 'finalanswer'; +} + +function findPendingResponseFinalIndex( + pending: PendingResponseFinal[], + projection: AssistantMessageProjection, + key: string +): number { + if (projection.itemId) { + const itemIndex = pending.findIndex((entry) => entry.itemId === projection.itemId); + if (itemIndex !== -1) return itemIndex; + } + return pending.findIndex((entry) => entry.key === key); +} + +function normalizeCompactionText(value: string): string { + return value.replace(/\r\n/g, '\n').trim(); +} + +function checkpointEndsWithSummary(checkpointMessage: string, summaryText: string): boolean { + if (checkpointMessage === summaryText) return true; + if (!checkpointMessage.endsWith(summaryText)) return false; + + const prefix = checkpointMessage.slice(0, -summaryText.length); + return /\n[\t ]*$/.test(prefix); +} + +function findCompactionSummaryIndex( + pending: PendingResponseFinal[], + payload: Record | null +): number { + const rawCheckpointMessage = asString(payload?.message); + const checkpointMessage = rawCheckpointMessage ? normalizeCompactionText(rawCheckpointMessage) : ''; + if (!checkpointMessage) return -1; + + for (let index = pending.length - 1; index >= 0; index -= 1) { + const text = normalizeCompactionText(pending[index]?.text ?? ''); + if (text && checkpointEndsWithSummary(checkpointMessage, text)) { + return index; + } + } + return -1; +} + +function convertProjectionToActions( + projection: CodexEventProjection | null, + fallbackTurnId: string | null = null +): CodexConversionAction[] { + if (!projection) return []; + + const actions: CodexConversionAction[] = []; + if (projection.sessionId) { + actions.push({ type: 'session-found', sessionId: projection.sessionId }); + } + if (projection.userMessage) { + actions.push({ type: 'user-message', message: projection.userMessage }); + } else if (projection.userActivity) { + actions.push({ type: 'user-activity' }); + } + + const turnId = projection.turnId ?? fallbackTurnId; + for (const message of projection.messages ?? []) { + actions.push({ + type: 'agent-message', + message, + ...(turnId ? { turnId } : {}) + }); + } + if (projection.finishedTurnId) { + actions.push({ type: 'turn-finished', turnId: projection.finishedTurnId }); + } + return actions; +} + +function confirmsPendingFinalVisibility(actions: CodexConversionAction[]): boolean { + return actions.some((action) => ( + action.type === 'user-message' + || action.type === 'user-activity' + || (action.type === 'agent-message' && action.message.type !== 'token_count') + )); +} + /** - * Transcript chat text is duplicated across Codex's semantic events and raw - * response items. Keep both as compatible sources, but project each visible - * assistant message once. Some Codex versions only persist final answers as - * response_item messages, while 0.147+ replaced legacy agent_message events - * with item_completed AgentMessage records. + * Project transcript records into one ordered action stream. Raw final-answer + * response items are ambiguous: they can be visible fallback messages or + * internal compaction summaries. Hold them in arrival order until a semantic + * mirror, subsequent visible activity, compaction checkpoint, or turn boundary + * resolves them. Later actions must never overtake an older pending final. */ -export function createCodexEventConverter(): (rawEvent: unknown) => CodexConversionResult | null { +export function createCodexEventConverter(): CodexEventConverter { let currentTurnId: string | null = null; const unmatchedSemanticMessages = new Map(); const unmatchedResponseMessages = new Map(); + const pendingResponseFinals: PendingResponseFinal[] = []; - return (rawEvent: unknown): CodexConversionResult | null => { + const emitPendingFinals = (entries: PendingResponseFinal[]): CodexConversionAction[] => { + const actions: CodexConversionAction[] = []; + for (const entry of entries) { + rememberProjection(unmatchedResponseMessages, entry.key); + for (const message of entry.messages) { + actions.push({ + type: 'agent-message', + message, + ...(entry.turnId ? { turnId: entry.turnId } : {}) + }); + } + } + return actions; + }; + + const drainPendingPrefix = (count: number): CodexConversionAction[] => ( + emitPendingFinals(pendingResponseFinals.splice(0, count)) + ); + + const drainAllPendingFinals = (): CodexConversionAction[] => ( + drainPendingPrefix(pendingResponseFinals.length) + ); + + const convert = (rawEvent: unknown): CodexConversionAction[] => { const parsed = CodexSessionEventSchema.safeParse(rawEvent); - if (!parsed.success) return null; + if (!parsed.success) return []; if (parsed.data.type === 'session_meta') { currentTurnId = null; unmatchedSemanticMessages.clear(); unmatchedResponseMessages.clear(); + pendingResponseFinals.length = 0; } currentTurnId = extractEventTurnId(parsed.data) ?? currentTurnId; const projection = extractAssistantMessageProjection(parsed.data, currentTurnId); const converted = convertCodexEvent(parsed.data); - if (!projection || !converted) return converted; + const convertedActions = convertProjectionToActions(converted, projection?.turnId ?? null); + const payload = asRecord(parsed.data.payload); + + if (parsed.data.type === 'compacted') { + // Local compaction stores its raw assistant summary at the end of + // checkpoint.message. Remote compaction leaves message empty and + // stores the summary only in replacement_history, so adjacency is + // insufficient evidence that the latest pending final is internal. + const summaryIndex = findCompactionSummaryIndex(pendingResponseFinals, payload); + if (summaryIndex === -1) return convertedActions; + + const earlierVisibleFinals = drainPendingPrefix(summaryIndex); + pendingResponseFinals.shift(); + return [...earlierVisibleFinals, ...convertedActions]; + } + + const eventType = parsed.data.type === 'event_msg' ? asString(payload?.type) : null; + if (eventType === 'task_complete') { + return [...drainAllPendingFinals(), ...convertedActions]; + } + if (eventType === 'turn_aborted' || eventType === 'task_failed') { + pendingResponseFinals.length = 0; + return convertedActions; + } + + if (!projection || !converted) { + return confirmsPendingFinalVisibility(convertedActions) + ? [...drainAllPendingFinals(), ...convertedActions] + : convertedActions; + } const key = `${projection.turnId ?? ''}\u0000${projection.text}`; + + const pendingIndex = projection.source === 'semantic' + ? findPendingResponseFinalIndex(pendingResponseFinals, projection, key) + : -1; + if (pendingIndex !== -1) { + const earlierVisibleFinals = drainPendingPrefix(pendingIndex); + pendingResponseFinals.shift(); + return [...earlierVisibleFinals, ...convertedActions]; + } + + if (projection.source === 'response' && isFinalAnswerResponse(parsed.data)) { + if (consumeProjection(unmatchedSemanticMessages, key)) { + return []; + } + // A response-only final answer is ambiguous until the next + // boundary: task_complete confirms visible output, while compacted + // identifies an internal context summary. Hold it briefly. + pendingResponseFinals.push({ + key, + text: projection.text, + itemId: projection.itemId, + turnId: projection.turnId, + messages: converted.messages ?? [] + }); + return []; + } + const opposite = projection.source === 'semantic' ? unmatchedResponseMessages : unmatchedSemanticMessages; if (consumeProjection(opposite, key)) { - return null; + return []; } + const earlierVisibleFinals = drainAllPendingFinals(); rememberProjection( projection.source === 'semantic' ? unmatchedSemanticMessages : unmatchedResponseMessages, key ); - return converted; + return [...earlierVisibleFinals, ...convertedActions]; }; + + return Object.assign(convert, { + finalize: drainAllPendingFinals + }); } -export function convertCodexEvent(rawEvent: unknown): CodexConversionResult | null { +export function convertCodexEvent(rawEvent: unknown): CodexEventProjection | null { const parsed = CodexSessionEventSchema.safeParse(rawEvent); if (!parsed.success) { return null; diff --git a/cli/src/codex/utils/codexSessionScanner.test.ts b/cli/src/codex/utils/codexSessionScanner.test.ts index 51362f03..3e05872e 100644 --- a/cli/src/codex/utils/codexSessionScanner.test.ts +++ b/cli/src/codex/utils/codexSessionScanner.test.ts @@ -105,9 +105,13 @@ describe('codexSessionScanner', () => { ); const replayFlags: boolean[] = []; + let replayCompleteCount = 0; scanner = await createCodexSessionScanner({ transcriptPath, replayExistingHistory: true, + onReplayComplete: () => { + replayCompleteCount += 1; + }, onEvent: (event, context) => { events.push(event); replayFlags.push(context.replayedHistory); @@ -119,6 +123,7 @@ describe('codexSessionScanner', () => { expect(events[0]?.type).toBe('session_meta'); expect(events[1]?.payload).toEqual({ type: 'agent_message', message: 'old' }); expect(replayFlags).toEqual([true, true]); + expect(replayCompleteCount).toBe(1); await appendFile( transcriptPath, @@ -126,6 +131,7 @@ describe('codexSessionScanner', () => { ); await scanner.flush(); expect(replayFlags).toEqual([true, true, false]); + expect(replayCompleteCount).toBe(1); }); it('reports session id from the transcript metadata', async () => { diff --git a/cli/src/codex/utils/codexSessionScanner.ts b/cli/src/codex/utils/codexSessionScanner.ts index 73c9c695..053b5cbf 100644 --- a/cli/src/codex/utils/codexSessionScanner.ts +++ b/cli/src/codex/utils/codexSessionScanner.ts @@ -7,6 +7,7 @@ interface CodexSessionScannerOptions { transcriptPath: string | null; onEvent: (event: CodexSessionEvent, context: { replayedHistory: boolean }) => void; onSessionId?: (sessionId: string) => void; + onReplayComplete?: () => void; replayExistingHistory?: boolean; } @@ -37,6 +38,7 @@ class CodexSessionScannerImpl extends BaseSessionScanner { private transcriptPath: string | null; private readonly onEvent: (event: CodexSessionEvent, context: { replayedHistory: boolean }) => void; private readonly onSessionId?: (sessionId: string) => void; + private readonly onReplayComplete?: () => void; private readonly fileEpochByPath = new Map(); private readonly fileStateByPath = new Map { this.transcriptPath = opts.transcriptPath; this.onEvent = opts.onEvent; this.onSessionId = opts.onSessionId; + this.onReplayComplete = opts.onReplayComplete; this.replayExistingHistoryOnNextAttach = opts.replayExistingHistory ?? false; } @@ -101,6 +104,9 @@ class CodexSessionScannerImpl extends BaseSessionScanner { } finally { this.replayingExistingHistory = false; } + if (replayedHistory) { + this.onReplayComplete?.(); + } if (stats.newCount > 0) { logger.debug(`[codex-session-scanner] ${stats.newCount} new events from ${stats.filePath}`); }