fix(codex): ignore late app-server writes during disconnect (#1748)

* fix(codex): ignore late app-server writes during disconnect

* fix(codex): scope app-server responses to their process

---------

Co-authored-by: huxiang <huxiang@myai.tech>

cli/src/codex/codexAppServerClient.test.ts

@@ -36,12 +36,26 @@ function fakeStream(): EventEmitter & { setEncoding: ReturnType<typeof vi.fn> }
 
 function fakeChild() {
     return Object.assign(new EventEmitter(), {
-        stdin: { end: vi.fn(), write: vi.fn() },
+        stdin: Object.assign(new EventEmitter(), {
+            end: vi.fn(),
+            write: vi.fn(),
+            destroyed: false,
+            writable: true,
+            writableEnded: false
+        }),
         stdout: fakeStream(),
         stderr: fakeStream()
     });
 }
 
+function deferred<T>() {
+    let resolve!: (value: T | PromiseLike<T>) => void;
+    const promise = new Promise<T>((resolvePromise) => {
+        resolve = resolvePromise;
+    });
+    return { promise, resolve };
+}
+
 describe('CodexAppServerClient process cwd', () => {
     beforeEach(() => {
         execFileSyncMock.mockClear();
@@ -144,4 +158,129 @@ describe('CodexAppServerClient process cwd', () => {
         await expect(steer.completed).rejects.toThrow('stdin closed');
         await client.disconnect();
     });
+
+    it('drops an incoming response when stdin closes while its handler is pending', async () => {
+        const child = fakeChild();
+        spawnMock.mockReturnValue(child);
+        const client = new CodexAppServerClient({ cwd: '/neutral-home' });
+        const handlerStarted = deferred<void>();
+        const handlerFinished = deferred<void>();
+
+        client.registerRequestHandler('slow/request', async () => {
+            handlerStarted.resolve();
+            await handlerFinished.promise;
+            return { ok: true };
+        });
+
+        await client.connect();
+        child.stdout.emit('data', Buffer.from(JSON.stringify({
+            id: 1,
+            method: 'slow/request',
+            params: {}
+        }) + '\n'));
+        await handlerStarted.promise;
+
+        child.stdin.destroyed = true;
+        child.stdin.writable = false;
+        handlerFinished.resolve();
+        await new Promise((resolve) => setTimeout(resolve, 0));
+
+        expect(child.stdin.write).not.toHaveBeenCalled();
+        await client.disconnect();
+    });
+
+    it('does not leak late incoming handler failures as unhandled rejections', async () => {
+        const child = fakeChild();
+        child.stdin.write = vi.fn(() => {
+            throw new Error('Cannot call write after a stream was destroyed');
+        });
+        spawnMock.mockReturnValue(child);
+        const client = new CodexAppServerClient({ cwd: '/neutral-home' });
+        const handlerFinished = deferred<void>();
+        const unhandled: unknown[] = [];
+        const onUnhandled = (reason: unknown) => {
+            unhandled.push(reason);
+        };
+        process.on('unhandledRejection', onUnhandled);
+
+        client.registerRequestHandler('slow/request', async () => {
+            await handlerFinished.promise;
+            return { ok: true };
+        });
+
+        try {
+            await client.connect();
+            child.stdout.emit('data', Buffer.from(JSON.stringify({
+                id: 1,
+                method: 'slow/request',
+                params: {}
+            }) + '\n'));
+            await Promise.resolve();
+            handlerFinished.resolve();
+            await new Promise((resolve) => setTimeout(resolve, 0));
+
+            expect(unhandled).toEqual([]);
+        } finally {
+            process.off('unhandledRejection', onUnhandled);
+            await client.disconnect();
+        }
+    });
+
+    it('handles an asynchronous stdin error after a writable response', async () => {
+        const child = fakeChild();
+        spawnMock.mockReturnValue(child);
+        const client = new CodexAppServerClient({ cwd: '/neutral-home' });
+
+        client.registerRequestHandler('ping', () => ({ ok: true }));
+        await client.connect();
+        child.stdout.emit('data', Buffer.from(JSON.stringify({
+            id: 1,
+            method: 'ping',
+            params: {}
+        }) + '\n'));
+        await new Promise((resolve) => setTimeout(resolve, 0));
+        expect(child.stdin.write).toHaveBeenCalledTimes(1);
+
+        await new Promise<void>((resolve) => {
+            setTimeout(() => {
+                child.stdin.emit('error', new Error('EPIPE'));
+                resolve();
+            }, 0);
+        });
+
+        expect(client.isConnected()).toBe(true);
+        await client.disconnect();
+    });
+
+    it('drops a late response from an old app-server after reconnecting', async () => {
+        const oldChild = fakeChild();
+        const newChild = fakeChild();
+        spawnMock.mockReturnValueOnce(oldChild).mockReturnValueOnce(newChild);
+        const client = new CodexAppServerClient({ cwd: '/neutral-home' });
+        const handlerStarted = deferred<void>();
+        const handlerFinished = deferred<void>();
+
+        client.registerRequestHandler('slow/request', async () => {
+            handlerStarted.resolve();
+            await handlerFinished.promise;
+            return { ok: true };
+        });
+
+        await client.connect();
+        oldChild.stdout.emit('data', Buffer.from(JSON.stringify({
+            id: 1,
+            method: 'slow/request',
+            params: {}
+        }) + '\n'));
+        await handlerStarted.promise;
+
+        await client.disconnect();
+        await client.connect();
+        handlerFinished.resolve();
+        await new Promise((resolve) => setTimeout(resolve, 0));
+
+        expect(oldChild.stdin.write).not.toHaveBeenCalled();
+        expect(newChild.stdin.write).not.toHaveBeenCalled();
+        await client.disconnect();
+    });
 });

cli/src/codex/codexAppServerClient.ts

@@ -224,6 +224,11 @@ export class CodexAppServerClient extends JsonLineParser {
         });
         this.process = child;
 
+        child.stdin.on('error', (error) => {
+            if (this.process !== child) return;
+            logger.debug('[CodexAppServer] stdin error', error);
+        });
+
         child.stdout.setEncoding('utf8');
         child.stdout.on('data', (chunk) => {
             if (this.process === child) this.feed(chunk);
@@ -677,10 +682,13 @@ export class CodexAppServerClient extends JsonLineParser {
 
             if ('id' in message && message.id !== undefined) {
                 const requestId = message.id;
+                const sourceProcess = this.process;
                 void this.handleIncomingRequest({
                     id: requestId,
                     method,
                     params
+                }, sourceProcess).catch((error) => {
+                    logger.debug('[CodexAppServer] Error handling incoming request', error);
                 });
                 return;
             }
@@ -694,7 +702,10 @@ export class CodexAppServerClient extends JsonLineParser {
         }
     }
 
-    private async handleIncomingRequest(request: { id: unknown; method: string; params?: unknown }): Promise<void> {
+    private async handleIncomingRequest(
+        request: { id: unknown; method: string; params?: unknown },
+        sourceProcess: ChildProcessWithoutNullStreams | null
+    ): Promise<void> {
         const responseId = typeof request.id === 'number' || typeof request.id === 'string'
             ? request.id
             : null;
@@ -707,7 +718,7 @@ export class CodexAppServerClient extends JsonLineParser {
                     code: -32601,
                     message: `Method not found: ${request.method}`
                 }
-            } satisfies JsonRpcLiteResponse);
+            } satisfies JsonRpcLiteResponse, sourceProcess);
             return;
         }
 
@@ -716,7 +727,7 @@ export class CodexAppServerClient extends JsonLineParser {
             this.writePayload({
                 id: responseId,
                 result
-            } satisfies JsonRpcLiteResponse);
+            } satisfies JsonRpcLiteResponse, sourceProcess);
         } catch (error) {
             this.writePayload({
                 id: responseId,
@@ -724,7 +735,7 @@ export class CodexAppServerClient extends JsonLineParser {
                     code: -32603,
                     message: error instanceof Error ? error.message : 'Internal error'
                 }
-            } satisfies JsonRpcLiteResponse);
+            } satisfies JsonRpcLiteResponse, sourceProcess);
         }
     }
 
@@ -764,9 +775,28 @@ export class CodexAppServerClient extends JsonLineParser {
         return error;
     }
 
-    private writePayload(payload: JsonRpcLiteRequest | JsonRpcLiteNotification | JsonRpcLiteResponse): void {
+    private writePayload(
+        payload: JsonRpcLiteRequest | JsonRpcLiteNotification | JsonRpcLiteResponse,
+        targetProcess: ChildProcessWithoutNullStreams | null = this.process
+    ): void {
+        if (!targetProcess || targetProcess !== this.process) {
+            return;
+        }
+
+        const stdin = targetProcess.stdin;
+        if (!stdin || stdin.destroyed || stdin.writableEnded || stdin.writable === false) {
+            return;
+        }
+
         const serialized = JSON.stringify(payload);
-        this.process?.stdin.write(`${serialized}\n`);
+        try {
+            stdin.write(`${serialized}\n`);
+        } catch (error) {
+            // The app-server can close stdin while an async request handler is
+            // still completing. Dropping that late response is safe; the
+            // transport is already unavailable and must not crash the runner.
+            logger.debug('[CodexAppServer] Ignoring payload write after process shutdown', error);
+        }
     }
 
     private resetParserState(): void {