From f4553dd1ec27a9baa924bb4b80ee6bb0f2091a5b Mon Sep 17 00:00:00 2001 From: HuXiang <617953687@qq.com> Date: Sun, 6 Sep 2026 14:18:20 +0800 Subject: [PATCH] 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 --- cli/src/codex/codexAppServerClient.test.ts | 141 ++++++++++++++++++++- cli/src/codex/codexAppServerClient.ts | 42 +++++- 2 files changed, 176 insertions(+), 7 deletions(-) diff --git a/cli/src/codex/codexAppServerClient.test.ts b/cli/src/codex/codexAppServerClient.test.ts index bd9c0ad5..b7df81ba 100644 --- a/cli/src/codex/codexAppServerClient.test.ts +++ b/cli/src/codex/codexAppServerClient.test.ts @@ -36,12 +36,26 @@ function fakeStream(): EventEmitter & { setEncoding: ReturnType } 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() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((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(); + const handlerFinished = deferred(); + + 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(); + 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((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(); + const handlerFinished = deferred(); + + 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(); + }); }); diff --git a/cli/src/codex/codexAppServerClient.ts b/cli/src/codex/codexAppServerClient.ts index 801c32d5..822a5d36 100644 --- a/cli/src/codex/codexAppServerClient.ts +++ b/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 { + private async handleIncomingRequest( + request: { id: unknown; method: string; params?: unknown }, + sourceProcess: ChildProcessWithoutNullStreams | null + ): Promise { 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 {