diff --git a/cli/src/codex/codexLocalLauncher.test.ts b/cli/src/codex/codexLocalLauncher.test.ts index eebcc4f0..3c72576e 100644 --- a/cli/src/codex/codexLocalLauncher.test.ts +++ b/cli/src/codex/codexLocalLauncher.test.ts @@ -140,6 +140,19 @@ describe('codexLocalLauncher', () => { let tempDir = ''; const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + const writeTranscriptMeta = async (fileName: string, sessionId: string): Promise => { + const transcriptPath = join(tempDir, fileName); + await writeFile( + transcriptPath, + JSON.stringify({ + type: 'session_meta', + payload: { + id: sessionId + } + }) + '\n' + ); + return transcriptPath; + }; beforeEach(async () => { tempDir = join(tmpdir(), `codex-local-launcher-${Date.now()}`); @@ -335,6 +348,135 @@ describe('codexLocalLauncher', () => { }); }); + 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'); + const { session } = createSessionStub('default'); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[0]?.('primary-thread', { + transcript_path: primaryTranscriptPath, + source: 'startup' + }); + await wait(100); + + harness.sessionHookHandlers[0]?.('other-thread', { + transcript_path: otherTranscriptPath, + source: 'startup' + }); + await wait(100); + + if (releaseRunBarrier) { + releaseRunBarrier(); + } + await launcherPromise; + + expect(session.sessionId).toBe('primary-thread'); + expect(session.transcriptPath).toBe(primaryTranscriptPath); + }); + + it('does not let a later hook without source replace the primary session', async () => { + const primaryTranscriptPath = await writeTranscriptMeta('primary-no-source.jsonl', 'primary-thread'); + const otherTranscriptPath = await writeTranscriptMeta('other-no-source.jsonl', 'other-thread'); + const { session } = createSessionStub('default'); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[0]?.('primary-thread', { + transcript_path: primaryTranscriptPath + }); + await wait(100); + + harness.sessionHookHandlers[0]?.('other-thread', { + transcript_path: otherTranscriptPath + }); + await wait(100); + + if (releaseRunBarrier) { + releaseRunBarrier(); + } + await launcherPromise; + + expect(session.sessionId).toBe('primary-thread'); + expect(session.transcriptPath).toBe(primaryTranscriptPath); + }); + + it('allows a clear hook to replace the primary session', async () => { + const primaryTranscriptPath = await writeTranscriptMeta('primary-before-clear.jsonl', 'primary-thread'); + const clearTranscriptPath = await writeTranscriptMeta('clear-transcript.jsonl', 'clear-thread'); + const { session } = createSessionStub('default'); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[0]?.('primary-thread', { + transcript_path: primaryTranscriptPath, + source: 'startup' + }); + await wait(100); + + harness.sessionHookHandlers[0]?.('clear-thread', { + transcript_path: clearTranscriptPath, + source: 'clear' + }); + await wait(100); + + if (releaseRunBarrier) { + releaseRunBarrier(); + } + await launcherPromise; + + expect(session.sessionId).toBe('clear-thread'); + expect(session.transcriptPath).toBe(clearTranscriptPath); + }); + + it('ignores mismatched session metadata from the active transcript scanner', async () => { + const transcriptPath = await writeTranscriptMeta('mismatched-scanner.jsonl', 'primary-thread'); + const { session } = createSessionStub('default'); + let releaseRunBarrier: (() => void) | undefined; + harness.runBarrier = new Promise((resolve) => { + releaseRunBarrier = resolve; + }); + + const launcherPromise = codexLocalLauncher(session as never); + await wait(50); + + harness.sessionHookHandlers[0]?.('primary-thread', { + transcript_path: transcriptPath + }); + await wait(100); + + await appendFile( + transcriptPath, + JSON.stringify({ type: 'session_meta', payload: { id: 'unexpected-thread' } }) + '\n' + ); + + await wait(2300); + + if (releaseRunBarrier) { + releaseRunBarrier(); + } + await launcherPromise; + + expect(session.sessionId).toBe('primary-thread'); + expect(session.transcriptPath).toBe(transcriptPath); + }); + it('does not leave transcript scanning alive after launcher teardown', async () => { const transcriptPath = join(tempDir, 'teardown-race-transcript.jsonl'); const { session, agentMessages } = createSessionStub('default'); diff --git a/cli/src/codex/codexLocalLauncher.ts b/cli/src/codex/codexLocalLauncher.ts index 2545f103..5d79340d 100644 --- a/cli/src/codex/codexLocalLauncher.ts +++ b/cli/src/codex/codexLocalLauncher.ts @@ -12,6 +12,8 @@ import { BaseLocalLauncher } from '@/modules/common/launcher/BaseLocalLauncher'; 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; @@ -31,20 +33,6 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch const { server: happyServer, mcpServers } = await buildHapiMcpBridge(session.client); logger.debug(`[codex-local]: Started hapi MCP bridge server at ${happyServer.url}`); - const hookServer = await startHookServer({ - onSessionHook: (sessionId, data) => { - if (shuttingDown) { - return; - } - session.onSessionFound(sessionId); - if (typeof data.transcript_path === 'string' && data.transcript_path.length > 0) { - hookReady = true; - session.onTranscriptPathFound(data.transcript_path); - } - } - }); - logger.debug(`[codex-local]: Started Codex SessionStart hook server on port ${hookServer.port}`); - 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}`; @@ -55,15 +43,40 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch }); }; - const handleSessionFound = (sessionId: string) => { + 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 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 => { 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) { await scanner.setTranscriptPath(transcriptPath); return; @@ -71,11 +84,19 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch const createdScanner = await createCodexSessionScanner({ transcriptPath, onSessionId: (sessionId) => { + if (!isPrimarySessionId(sessionId)) { + logger.debug(`[codex-local]: Ignoring transcript session id ${sessionId}; primary is ${primarySessionId}`); + return; + } session.onSessionFound(sessionId); }, onEvent: (event) => { const converted = convertCodexEvent(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) { @@ -108,6 +129,35 @@ export async function codexLocalLauncher(session: CodexSession): Promise<'switch return pendingScannerSetup; }; + const handleSessionHook = (sessionId: string, data: Record): 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) { + handleSessionFound(sessionId, shouldAllowSessionSwitch); + return; + } + + bindPrimarySession(sessionId, transcriptPath, shouldAllowSessionSwitch); + }; + + const hookServer = await startHookServer({ + onSessionHook: (sessionId, data) => { + if (shuttingDown) { + return; + } + handleSessionHook(sessionId, data); + } + }); + logger.debug(`[codex-local]: Started Codex SessionStart hook server on port ${hookServer.port}`); + const launcher = new BaseLocalLauncher({ label: 'codex-local', failureLabel: 'Local Codex process failed',