Fix session pollution: route notifications by threadId (#123)

* Fix session pollution: route notifications by threadId

notify() broadcasted every Codex notification to every registered
session handler, so a stale handler from a previous prompt would
re-emit the new session's deltas as ACP sessionUpdates on the old
sessionId. awaitTurnCompleted() had a similar issue: it never disposed
its turn/completed listener and didn't filter by threadId, so
concurrent prompts across sessions could resolve to the wrong turn.

Filter thread-scoped notifications by params.threadId in notify(), and
make awaitTurnCompleted thread-scoped with disposal. Non-thread events
(account, mcp, app-list, warnings) still broadcast.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Match turn completions by turn id

Add runTurn to handle turn/completed races and ensure concurrent
prompts resolve only the matching thread and turn. Update tests to
cover out-of-order and early completion notifications.

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Ben Brandt <benjamin.j.brandt@gmail.com>

Author: 3 people

src/CodexAcpClient.ts

@@ -383,7 +383,7 @@ export class CodexAcpClient {
         const effort = modelId.effort as ReasoningEffort | null; //TODO remove unsafe conversion
 
         await this.refreshSkills(cwd, request._meta);
-        await this.codexClient.turnStart({
+        return await this.codexClient.runTurn({
             outputSchema: null,
             threadId: request.sessionId,
             input: input,
@@ -395,10 +395,6 @@ export class CodexAcpClient {
             effort: effort,
             model: modelId.model,
         });
-
-        // Wait for turn completion
-        // If turnInterrupt() was called, Codex will send turn/completed event with status "interrupted"
-        return await this.codexClient.awaitTurnCompleted();
     }
 
     async listSkills(params?: SkillsListParams): Promise<SkillsListResponse> {

src/CodexAppServerClient.ts

@@ -94,6 +94,8 @@ export class CodexAppServerClient {
     private mcpServerStartupVersion = 0;
     private readonly mcpServerStartupStates = new Map<string, McpServerStartupSnapshot>();
     private readonly mcpServerStartupResolvers: Array<McpServerStartupResolver> = [];
+    private readonly pendingTurnCompletionResolvers = new Map<string, Map<string, (event: TurnCompletedNotification) => void>>();
+    private readonly turnCompletionCaptures = new Map<string, Set<(event: TurnCompletedNotification) => void>>();
 
     constructor(connection: MessageConnection) {
         this.connection = connection;
@@ -108,6 +110,9 @@ export class CodexAppServerClient {
                 });
                 this.resolveMcpServerStartupResolvers();
             }
+            if (isTurnCompletedNotification(serverNotification)) {
+                this.recordTurnCompleted(serverNotification.params);
+            }
             this.notify(serverNotification);
             for (const callback of this.codexEventHandlers) {
                 callback({ eventType: "notification", ...serverNotification });
@@ -155,6 +160,27 @@ export class CodexAppServerClient {
         return await this.sendRequest({ method: "turn/start", params: params });
     }
 
+    async runTurn(params: TurnStartParams): Promise<TurnCompletedNotification> {
+        const capturedCompletions: Array<TurnCompletedNotification> = [];
+        const releaseCapture = this.captureTurnCompletions(params.threadId, (event) => {
+            capturedCompletions.push(event);
+        });
+
+        try {
+            const turnStarted = await this.turnStart(params);
+            const earlyCompletion = capturedCompletions.find(event => event.turn.id === turnStarted.turn.id);
+            releaseCapture();
+            if (earlyCompletion) {
+                return earlyCompletion;
+            }
+            // Wait for turn completion
+            // If turnInterrupt() was called, Codex will send turn/completed event with status "interrupted"
+            return await this.awaitTurnCompleted(params.threadId, turnStarted.turn.id);
+        } finally {
+            releaseCapture();
+        }
+    }
+
     async turnInterrupt(params: TurnInterruptParams): Promise<TurnInterruptResponse> {
         return await this.sendRequest({ method: "turn/interrupt", params: params });
     }
@@ -224,11 +250,10 @@ export class CodexAppServerClient {
     }
 
     //TODO create type-safe helper
-    async awaitTurnCompleted(): Promise<TurnCompletedNotification> {
+    async awaitTurnCompleted(threadId: string, turnId: string): Promise<TurnCompletedNotification> {
         return await new Promise((resolve) => {
-            this.connection.onNotification("turn/completed", (event: TurnCompletedNotification) => {
-                resolve(event);
-            });
+            const threadResolvers = this.getOrCreatePendingTurnCompletionResolvers(threadId);
+            threadResolvers.set(turnId, resolve);
         });
     }
 
@@ -255,11 +280,67 @@ export class CodexAppServerClient {
 
     private notificationHandlers = new Map<string, (event: ServerNotification) => void>();
     private notify(notification: ServerNotification) {
+        const threadId = extractThreadId(notification);
+        if (threadId !== null) {
+            const handler = this.notificationHandlers.get(threadId);
+            if (handler) {
+                handler(notification);
+            }
+            return;
+        }
         for (const notificationHandler of this.notificationHandlers.values()) {
             notificationHandler(notification);
         }
     }
 
+    private recordTurnCompleted(event: TurnCompletedNotification): void {
+        const threadResolvers = this.pendingTurnCompletionResolvers.get(event.threadId);
+        const resolve = threadResolvers?.get(event.turn.id);
+        if (resolve) {
+            threadResolvers!.delete(event.turn.id);
+            if (threadResolvers!.size === 0) {
+                this.pendingTurnCompletionResolvers.delete(event.threadId);
+            }
+            resolve(event);
+            return;
+        }
+
+        const captures = this.turnCompletionCaptures.get(event.threadId);
+        if (!captures) {
+            return;
+        }
+        for (const capture of captures) {
+            capture(event);
+        }
+    }
+
+    private getOrCreatePendingTurnCompletionResolvers(threadId: string): Map<string, (event: TurnCompletedNotification) => void> {
+        const existing = this.pendingTurnCompletionResolvers.get(threadId);
+        if (existing) {
+            return existing;
+        }
+        const created = new Map<string, (event: TurnCompletedNotification) => void>();
+        this.pendingTurnCompletionResolvers.set(threadId, created);
+        return created;
+    }
+
+    private captureTurnCompletions(threadId: string, capture: (event: TurnCompletedNotification) => void): () => void {
+        const captures = this.turnCompletionCaptures.get(threadId) ?? new Set<(event: TurnCompletedNotification) => void>();
+        captures.add(capture);
+        this.turnCompletionCaptures.set(threadId, captures);
+        let released = false;
+        return () => {
+            if (released) {
+                return;
+            }
+            released = true;
+            captures.delete(capture);
+            if (captures.size === 0) {
+                this.turnCompletionCaptures.delete(threadId);
+            }
+        };
+    }
+
     private resolveMcpServerStartupResolvers(): void {
         const pendingResolvers: Array<McpServerStartupResolver> = [];
         for (const resolver of this.mcpServerStartupResolvers) {
@@ -352,3 +433,18 @@ function isMcpServerStatusUpdatedNotification(notification: ServerNotification):
 } {
     return notification.method === "mcpServer/startupStatus/updated";
 }
+
+function isTurnCompletedNotification(notification: ServerNotification): notification is {
+    method: "turn/completed";
+    params: TurnCompletedNotification;
+} {
+    return notification.method === "turn/completed";
+}
+
+function extractThreadId(notification: ServerNotification): string | null {
+    const params = notification.params as { threadId?: unknown } | undefined;
+    if (params && typeof params.threadId === "string") {
+        return params.threadId;
+    }
+    return null;
+}

src/__tests__/CodexACPAgent/CodexAcpClient.test.ts

@@ -7,7 +7,7 @@ import {createTestFixture, createCodexMockTestFixture, createTestSessionState, t
 import type {ServerNotification} from "../../app-server";
 import type {SessionState} from "../../CodexAcpServer";
 import {AgentMode} from "../../AgentMode";
-import type {ListMcpServerStatusResponse, Model, SkillsListResponse} from "../../app-server/v2";
+import type {ListMcpServerStatusResponse, Model, SkillsListResponse, TurnStartParams} from "../../app-server/v2";
 import type {RateLimitsMap} from "../../RateLimitsMap";
 import {ModelId} from "../../ModelId";
 
@@ -407,6 +407,32 @@ describe('ACP server test', { timeout: 40_000 }, () => {
         return onServerNotification;
     }
 
+    function createTurn(id: string, status: "inProgress" | "completed") {
+        return {
+            id,
+            items: [],
+            status,
+            error: null,
+            startedAt: null,
+            completedAt: null,
+            durationMs: null,
+        };
+    }
+
+    function createTurnCompletedNotification(threadId: string, turnId: string): ServerNotification {
+        return {
+            method: "turn/completed",
+            params: {
+                threadId,
+                turn: createTurn(turnId, "completed"),
+            },
+        };
+    }
+
+    async function flushAsyncWork(): Promise<void> {
+        await new Promise((resolve) => setTimeout(resolve, 0));
+    }
+
     it('should map events from dump', async () => {
         fixture.getCodexAppServerClient().onServerNotification = loadNotifications();
 
@@ -462,9 +488,9 @@ describe('ACP server test', { timeout: 40_000 }, () => {
 
         // Trigger notifications after both prompts - should produce only 3 events, not 6
         const serverNotifications: ServerNotification[] = [
-            { method: "item/agentMessage/delta", params: { threadId: "string", turnId: "string", itemId: "string", delta: "He", }},
-            { method: "item/agentMessage/delta", params: { threadId: "string", turnId: "string", itemId: "string", delta: "ll", }},
-            { method: "item/agentMessage/delta", params: { threadId: "string", turnId: "string", itemId: "string", delta: "o!", }},
+            { method: "item/agentMessage/delta", params: { threadId: "id", turnId: "string", itemId: "string", delta: "He", }},
+            { method: "item/agentMessage/delta", params: { threadId: "id", turnId: "string", itemId: "string", delta: "ll", }},
+            { method: "item/agentMessage/delta", params: { threadId: "id", turnId: "string", itemId: "string", delta: "o!", }},
         ];
         for (const notification of serverNotifications) {
             mockFixture.sendServerNotification(notification);
@@ -486,10 +512,6 @@ describe('ACP server test', { timeout: 40_000 }, () => {
         mockFixture.getCodexAppServerClient().turnStart = vi.fn().mockResolvedValue({
             turn: { id: "turn-id", items: [], status: "inProgress", error: null }
         });
-        mockFixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockResolvedValue({
-            threadId: "id",
-            turn: { id: "turn-id", items: [], status: "completed", error: null }
-        });
 
         const sessionState1: SessionState = createTestSessionState({
             sessionId: "session-1",
@@ -506,15 +528,23 @@ describe('ACP server test', { timeout: 40_000 }, () => {
             return sessionId === "session-1" ? sessionState1 : sessionState2;
         });
 
+        // awaitTurnCompleted is per-turn; resolve the matching thread and turn.
+        mockFixture.getCodexAppServerClient().awaitTurnCompleted = vi.fn().mockImplementation((threadId: string, turnId: string) => Promise.resolve({
+            threadId,
+            turn: createTurn(turnId, "completed")
+        }));
+
         // Start prompts for two different sessions
         await codexAcpAgent.prompt({ sessionId: "session-1", prompt: [{type: "text", text: "Message to session 1"}] });
         await codexAcpAgent.prompt({ sessionId: "session-2", prompt: [{type: "text", text: "Message to session 2"}] });
 
         mockFixture.clearAcpConnectionDump();
 
-        // Trigger notifications - both session handlers should receive them
+        // Each notification carries the threadId of the session it belongs to,
+        // and must only be dispatched to that session.
         const serverNotifications: ServerNotification[] = [
-            { method: "item/agentMessage/delta", params: { threadId: "string", turnId: "string", itemId: "string", delta: "Hello", }},
+            { method: "item/agentMessage/delta", params: { threadId: "session-1", turnId: "string", itemId: "string", delta: "Hello-1", }},
+            { method: "item/agentMessage/delta", params: { threadId: "session-2", turnId: "string", itemId: "string", delta: "Hello-2", }},
         ];
         for (const notification of serverNotifications) {
             mockFixture.sendServerNotification(notification);
@@ -523,18 +553,100 @@ describe('ACP server test', { timeout: 40_000 }, () => {
         // Wait for async handlers to complete
         await vi.waitFor(() => {
             const dump = mockFixture.getAcpConnectionDump([]);
-            expect(dump.length).toBeGreaterThan(0);
+            expect(dump.length).toBeGreaterThanOrEqual(2);
         });
 
-        // Should have 2 events - one for each session's handler
+        // Should have exactly 2 events - the session-1 delta only on session-1, and
+        // the session-2 delta only on session-2 (no cross-session pollution).
         await expect(mockFixture.getAcpConnectionDump([])).toMatchFileSnapshot("data/multiple-sessions.json");
     });
 
+    it('should complete concurrent prompts by matching thread and turn id', async () => {
+        const mockFixture = createCodexMockTestFixture();
+        const codexAcpAgent = mockFixture.getCodexAcpAgent();
+
+        const turnIds = new Map([
+            ["session-1", "turn-1"],
+            ["session-2", "turn-2"],
+        ]);
+        const turnStart = vi.fn().mockImplementation((params: TurnStartParams) => Promise.resolve({
+            turn: createTurn(turnIds.get(params.threadId) ?? "unknown-turn", "inProgress"),
+        }));
+        mockFixture.getCodexAppServerClient().turnStart = turnStart;
+
+        const sessionState1: SessionState = createTestSessionState({
+            sessionId: "session-1",
+            currentModelId: "model-id[effort]",
+            agentMode: AgentMode.DEFAULT_AGENT_MODE
+        });
+        const sessionState2: SessionState = createTestSessionState({
+            sessionId: "session-2",
+            currentModelId: "model-id[effort]",
+            agentMode: AgentMode.DEFAULT_AGENT_MODE
+        });
+        vi.spyOn(codexAcpAgent, "getSessionState").mockImplementation((sessionId: string) => {
+            return sessionId === "session-1" ? sessionState1 : sessionState2;
+        });
+
+        const prompt1 = codexAcpAgent.prompt({ sessionId: "session-1", prompt: [{type: "text", text: "Message to session 1"}] });
+        const prompt2 = codexAcpAgent.prompt({ sessionId: "session-2", prompt: [{type: "text", text: "Message to session 2"}] });
+
+        await vi.waitFor(() => {
+            expect(turnStart).toHaveBeenCalledTimes(2);
+        });
+
+        let prompt1Settled = false;
+        void prompt1.then(() => {
+            prompt1Settled = true;
+        }, () => {
+            prompt1Settled = true;
+        });
+
+        mockFixture.sendServerNotification(createTurnCompletedNotification("session-1", "old-turn"));
+        await flushAsyncWork();
+        expect(prompt1Settled).toBe(false);
+
+        mockFixture.sendServerNotification(createTurnCompletedNotification("session-2", "turn-2"));
+        await expect(prompt2).resolves.toMatchObject({stopReason: "end_turn"});
+        expect(prompt1Settled).toBe(false);
+
+        mockFixture.sendServerNotification(createTurnCompletedNotification("session-1", "turn-1"));
+        await expect(prompt1).resolves.toMatchObject({stopReason: "end_turn"});
+    });
+
+    it('should handle a turn completion that arrives before awaitTurnCompleted is called', async () => {
+        const mockFixture = createCodexMockTestFixture();
+        const codexAcpAgent = mockFixture.getCodexAcpAgent();
+
+        mockFixture.getCodexAppServerClient().turnStart = vi.fn().mockImplementation((params: TurnStartParams) => {
+            mockFixture.sendServerNotification(createTurnCompletedNotification(params.threadId, "fast-turn"));
+            return Promise.resolve({
+                turn: createTurn("fast-turn", "inProgress"),
+            });
+        });
+
+        vi.spyOn(codexAcpAgent, "getSessionState").mockReturnValue(createTestSessionState({
+            sessionId: "fast-session",
+            currentModelId: "model-id[effort]",
+            agentMode: AgentMode.DEFAULT_AGENT_MODE
+        }));
+
+        await expect(codexAcpAgent.prompt({
+            sessionId: "fast-session",
+            prompt: [{type: "text", text: "Fast completion"}],
+        })).resolves.toMatchObject({stopReason: "end_turn"});
+    });
+
     it('should send attachments as prompt items', async () => {
         const mockFixture = createCodexMockTestFixture();
         const codexAcpAgent = mockFixture.getCodexAcpAgent();
         const codexAppServerClient = mockFixture.getCodexAppServerClient();
 
+        const realTurnStart = codexAppServerClient.turnStart.bind(codexAppServerClient);
+        vi.spyOn(codexAppServerClient, "turnStart").mockImplementation(async (params) => {
+            await realTurnStart(params);
+            return {turn: createTurn("turn-id", "inProgress")};
+        });
         vi.spyOn(codexAppServerClient, "awaitTurnCompleted").mockResolvedValue({
             threadId: "session-id",
             turn: {