Fix stale file read after patch causing empty diff content (#178)

SteffenDE · web-flow · commit feb979728f65 · 2026-06-08T17:28:56.000+03:00
* Fix stale file read after patch causing empty diff content

Also ensure events are always handled in order.

* queue per session, drain
diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts
@@ -49,6 +49,7 @@ export class CodexAcpClient {
     private gatewayConfig: GatewayConfig | null;
     private pendingLoginCompleted: Promise<AccountLoginCompletedNotification> | null = null;
     private pendingAccountUpdated: Promise<AccountUpdatedNotification> | null = null;
+    private readonly sessionNotificationQueues = new Map<string, Promise<void>>();
 
 
     constructor(codexClient: CodexAppServerClient, codexConfig?: JsonObject, modelProvider?: string) {
@@ -384,13 +385,56 @@ export class CodexAcpClient {
 
     async subscribeToSessionEvents(
         sessionId: string,
-        eventHandler: (result: ServerNotification) => void,
+        eventHandler: (result: ServerNotification) => void | Promise<void>,
         approvalHandler: ApprovalHandler,
         elicitationHandler: ElicitationHandler
     ) {
-        this.codexClient.onServerNotification(sessionId, eventHandler);
-        this.codexClient.onApprovalRequest(sessionId, approvalHandler);
-        this.codexClient.onElicitationRequest(sessionId, elicitationHandler);
+        this.codexClient.onServerNotification(sessionId, (event) => {
+            this.enqueueSessionNotification(sessionId, () => eventHandler(event));
+        });
+        this.codexClient.onApprovalRequest(sessionId, {
+            handleCommandExecution: async (params) => {
+                await this.waitForSessionNotifications(sessionId);
+                return await approvalHandler.handleCommandExecution(params);
+            },
+            handleFileChange: async (params) => {
+                await this.waitForSessionNotifications(sessionId);
+                return await approvalHandler.handleFileChange(params);
+            },
+        });
+        this.codexClient.onElicitationRequest(sessionId, {
+            handleElicitation: async (params) => {
+                await this.waitForSessionNotifications(sessionId);
+                return await elicitationHandler.handleElicitation(params);
+            },
+        });
+    }
+
+    async waitForSessionNotifications(sessionId: string): Promise<void> {
+        while (true) {
+            const queue = this.sessionNotificationQueues.get(sessionId);
+            if (!queue) return;
+            await queue;
+        }
+    }
+
+    private enqueueSessionNotification(sessionId: string, operation: () => void | Promise<void>): void {
+        const run = async () => {
+            try {
+                await operation();
+            } catch (error) {
+                logger.error("Error handling Codex session notification", error);
+            }
+        };
+
+        const previous = this.sessionNotificationQueues.get(sessionId);
+        const next = previous ? previous.then(run, run) : run();
+        this.sessionNotificationQueues.set(sessionId, next);
+        void next.finally(() => {
+            if (this.sessionNotificationQueues.get(sessionId) === next) {
+                this.sessionNotificationQueues.delete(sessionId);
+            }
+        });
     }
 
     async sendPrompt(
diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts
@@ -1222,6 +1222,8 @@ export class CodexAcpServer implements acp.Agent {
                 };
             }
 
+            await this.codexAcpClient.waitForSessionNotifications(params.sessionId);
+
             // Check if turn was interrupted (cancelled)
             if (turnCompleted.turn.status === "interrupted") {
                 if (!this.sessionIsClosing(params.sessionId) && this.sessions.has(params.sessionId)) {
diff --git a/src/CodexToolCallMapper.ts b/src/CodexToolCallMapper.ts
@@ -294,7 +294,18 @@ async function createUpdateFileContent(change: FileUpdateChange): Promise<ToolCa
     const oldContent = await readFileContent(change.path);
     if (oldContent !== null) {
         const patchedContent = applyPatch(oldContent, unifiedDiff);
-        if (patchedContent === false) return null;
+        if (patchedContent === false) {
+            // If Codex runs in full access mode, the file might already be patched.
+            // we can verify this by checking if the reverted patch applies.
+            const revertedPatch = revertPatch(unifiedDiff);
+            if (revertedPatch) {
+                const revertedContent = applyPatch(oldContent, revertedPatch);
+                if (revertedContent !== false) {
+                    return createUpdateDiffContent(change.path, revertedContent, oldContent);
+                }
+            }
+            return null;
+        }
         return createUpdateDiffContent(movePath ?? change.path, oldContent, patchedContent);
     }
 
diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts
@@ -1172,6 +1172,7 @@ describe('ACP server test', { timeout: 40_000 }, () => {
                 }
             }
         });
+        await mockFixture.getCodexAcpClient().waitForSessionNotifications(sessionId);
 
         expect(sessionState.rateLimits).not.toBeNull();
         expect(sessionState.rateLimits!.size).toBe(2);
diff --git a/src/__tests__/CodexACPAgent/data/mcp-tool-completed-with-logs.json b/src/__tests__/CodexACPAgent/data/mcp-tool-completed-with-logs.json
@@ -4,12 +4,23 @@
     {
       "sessionId": "test-session-id",
       "update": {
-        "sessionUpdate": "tool_call_update",
+        "sessionUpdate": "tool_call",
         "toolCallId": "call-id",
-        "_meta": {
-          "mcp_output_delta": {
-            "data": "File /Users/aleksandr.slapoguzov/Projects/ultimate/.ai/local.md doesn't exist or can't be opened"
+        "kind": "execute",
+        "title": "mcp.ijproxy.read_file",
+        "status": "in_progress",
+        "rawInput": {
+          "server": "ijproxy",
+          "tool": "read_file",
+          "arguments": {
+            "file_path": ".ai/local.md",
+            "mode": "slice",
+            "start_line": 1,
+            "max_lines": 200
           }
+        },
+        "_meta": {
+          "is_mcp_tool_call": true
         }
       }
     }
@@ -23,21 +34,9 @@
       "update": {
         "sessionUpdate": "tool_call_update",
         "toolCallId": "call-id",
-        "status": "failed",
-        "rawInput": {
-          "server": "ijproxy",
-          "tool": "read_file",
-          "arguments": {
-            "file_path": ".ai/local.md",
-            "mode": "slice",
-            "start_line": 1,
-            "max_lines": 200
-          }
-        },
-        "rawOutput": {
-          "result": null,
-          "error": {
-            "message": "File /Users/aleksandr.slapoguzov/Projects/ultimate/.ai/local.md doesn't exist or can't be opened"
+        "_meta": {
+          "mcp_output_delta": {
+            "data": "File /Users/aleksandr.slapoguzov/Projects/ultimate/.ai/local.md doesn't exist or can't be opened"
           }
         }
       }
@@ -50,11 +49,9 @@
     {
       "sessionId": "test-session-id",
       "update": {
-        "sessionUpdate": "tool_call",
+        "sessionUpdate": "tool_call_update",
         "toolCallId": "call-id",
-        "kind": "execute",
-        "title": "mcp.ijproxy.read_file",
-        "status": "in_progress",
+        "status": "failed",
         "rawInput": {
           "server": "ijproxy",
           "tool": "read_file",
@@ -65,8 +62,11 @@
             "max_lines": 200
           }
         },
-        "_meta": {
-          "is_mcp_tool_call": true
+        "rawOutput": {
+          "result": null,
+          "error": {
+            "message": "File /Users/aleksandr.slapoguzov/Projects/ultimate/.ai/local.md doesn't exist or can't be opened"
+          }
         }
       }
     }
diff --git a/src/__tests__/CodexACPAgent/data/mcp-tool-repeated-progress.json b/src/__tests__/CodexACPAgent/data/mcp-tool-repeated-progress.json
@@ -4,12 +4,20 @@
     {
       "sessionId": "test-session-id",
       "update": {
-        "sessionUpdate": "tool_call_update",
+        "sessionUpdate": "tool_call",
         "toolCallId": "call-id",
-        "_meta": {
-          "mcp_output_delta": {
-            "data": "Polling for status"
+        "kind": "execute",
+        "title": "mcp.server-name.tool-name",
+        "status": "in_progress",
+        "rawInput": {
+          "server": "server-name",
+          "tool": "tool-name",
+          "arguments": {
+            "argument": "example"
           }
+        },
+        "_meta": {
+          "is_mcp_tool_call": true
         }
       }
     }
@@ -40,18 +48,9 @@
       "update": {
         "sessionUpdate": "tool_call_update",
         "toolCallId": "call-id",
-        "status": "failed",
-        "rawInput": {
-          "server": "server-name",
-          "tool": "tool-name",
-          "arguments": {
-            "argument": "example"
-          }
-        },
-        "rawOutput": {
-          "result": null,
-          "error": {
-            "message": "Polling for status"
+        "_meta": {
+          "mcp_output_delta": {
+            "data": "Polling for status"
           }
         }
       }
@@ -64,20 +63,21 @@
     {
       "sessionId": "test-session-id",
       "update": {
-        "sessionUpdate": "tool_call",
+        "sessionUpdate": "tool_call_update",
         "toolCallId": "call-id",
-        "kind": "execute",
-        "title": "mcp.server-name.tool-name",
-        "status": "in_progress",
+        "status": "failed",
         "rawInput": {
           "server": "server-name",
           "tool": "tool-name",
           "arguments": {
             "argument": "example"
           }
         },
-        "_meta": {
-          "is_mcp_tool_call": true
+        "rawOutput": {
+          "result": null,
+          "error": {
+            "message": "Polling for status"
+          }
         }
       }
     }
diff --git a/src/__tests__/CodexACPAgent/data/terminal-full-flow.json b/src/__tests__/CodexACPAgent/data/terminal-full-flow.json
@@ -4,11 +4,24 @@
     {
       "sessionId": "test-session-id",
       "update": {
-        "sessionUpdate": "tool_call_update",
+        "sessionUpdate": "tool_call",
         "toolCallId": "command-flow",
+        "kind": "execute",
+        "title": "echo hello",
+        "status": "in_progress",
+        "content": [
+          {
+            "type": "terminal",
+            "terminalId": "command-flow"
+          }
+        ],
+        "rawInput": {
+          "command": "echo hello",
+          "cwd": "/test/project"
+        },
         "_meta": {
-          "terminal_output_delta": {
-            "data": "hello\n",
+          "terminal_info": {
+            "cwd": "/test/project",
             "terminal_id": "command-flow"
           }
         }
@@ -24,15 +37,9 @@
       "update": {
         "sessionUpdate": "tool_call_update",
         "toolCallId": "command-flow",
-        "status": "completed",
-        "rawOutput": {
-          "formatted_output": "hello\n",
-          "exit_code": 0
-        },
         "_meta": {
-          "terminal_exit": {
-            "exit_code": 0,
-            "signal": null,
+          "terminal_output_delta": {
+            "data": "hello\n",
             "terminal_id": "command-flow"
           }
         }
@@ -46,24 +53,17 @@
     {
       "sessionId": "test-session-id",
       "update": {
-        "sessionUpdate": "tool_call",
+        "sessionUpdate": "tool_call_update",
         "toolCallId": "command-flow",
-        "kind": "execute",
-        "title": "echo hello",
-        "status": "in_progress",
-        "content": [
-          {
-            "type": "terminal",
-            "terminalId": "command-flow"
-          }
-        ],
-        "rawInput": {
-          "command": "echo hello",
-          "cwd": "/test/project"
+        "status": "completed",
+        "rawOutput": {
+          "formatted_output": "hello\n",
+          "exit_code": 0
         },
         "_meta": {
-          "terminal_info": {
-            "cwd": "/test/project",
+          "terminal_exit": {
+            "exit_code": 0,
+            "signal": null,
             "terminal_id": "command-flow"
           }
         }
diff --git a/src/__tests__/CodexACPAgent/elicitation-events.test.ts b/src/__tests__/CodexACPAgent/elicitation-events.test.ts
@@ -288,6 +288,7 @@ describe('Elicitation Events', () => {
 
             fixture.sendServerNotification(startedNotification);
             fixture.sendServerNotification(completedNotification);
+            await fixture.getCodexAcpClient().waitForSessionNotifications(sessionId);
             fixture.clearAcpConnectionDump();
 
             const params: McpServerElicitationRequestParams = {
@@ -340,6 +341,7 @@ describe('Elicitation Events', () => {
 
             fixture.sendServerNotification(startedNotification);
             fixture.sendServerNotification(resolvedNotification);
+            await fixture.getCodexAcpClient().waitForSessionNotifications(sessionId);
             fixture.clearAcpConnectionDump();
 
             const params: McpServerElicitationRequestParams = {
diff --git a/src/__tests__/CodexACPAgent/file-change-events.test.ts b/src/__tests__/CodexACPAgent/file-change-events.test.ts
diff --git a/src/__tests__/acp-test-utils.ts b/src/__tests__/acp-test-utils.ts

Original file line number	Diff line number	Diff line change
`@@ -1222,6 +1222,8 @@ export class CodexAcpServer implements acp.Agent {`
`1222`	`1222`	`};`
`1223`	`1223`	`}`
`1224`	`1224`
	`1225`	`+ await this.codexAcpClient.waitForSessionNotifications(params.sessionId);`
	`1226`	`+`
`1225`	`1227`	`// Check if turn was interrupted (cancelled)`
`1226`	`1228`	`if (turnCompleted.turn.status === "interrupted") {`
`1227`	`1229`	`if (!this.sessionIsClosing(params.sessionId) && this.sessions.has(params.sessionId)) {`
Original file line number	Diff line number	Diff line change
`@@ -1172,6 +1172,7 @@ describe('ACP server test', { timeout: 40_000 }, () => {`
`1172`	`1172`	`}`
`1173`	`1173`	`}`
`1174`	`1174`	`});`
	`1175`	`+ await mockFixture.getCodexAcpClient().waitForSessionNotifications(sessionId);`
`1175`	`1176`
`1176`	`1177`	`expect(sessionState.rateLimits).not.toBeNull();`
`1177`	`1178`	`expect(sessionState.rateLimits!.size).toBe(2);`