fix(codex): subagent session hook may override primary session

2026-08-05 06:24:37 +00:00 · 2026-04-27 21:13:06 +08:00
parent 0cfc3b4ed2
commit cfc1edb747
2 changed files with 207 additions and 15 deletions
@@ -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<string> => {
+        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');
@@ -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<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) {
            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<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) {
+            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',