From 74a377ff6ac0a82e339ff610be789148e4bffc59 Mon Sep 17 00:00:00 2001 From: ElliotWu <112760141+Elliotwu-7@users.noreply.github.com> Date: Mon, 31 Aug 2026 21:35:34 +0800 Subject: [PATCH] fix(server): keep daemon alive when JSONL RPC stdin hits EPIPE (#4048) * fix(server): keep daemon alive when JSONL RPC stdin hits EPIPE A dead Claude/OMP child currently turns stdin.write() EPIPE into uncaughtException and kills the daemon. Catch the write, listen for stdin errors, and close that transport instead. Closes #4047 Co-Authored-By: Claude Sonnet 5 * test(server): reproduce JSONL RPC stdin EPIPE * test(server): gate native EPIPE fixture on POSIX --------- Co-authored-by: Claude Sonnet 5 Co-authored-by: Mohamed Boudra --- .../agent/providers/jsonl-rpc-process.test.ts | 75 ++++++++++++++++++- .../agent/providers/jsonl-rpc-process.ts | 24 +++++- 2 files changed, 96 insertions(+), 3 deletions(-) diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts index 960d61c1f0..2b3e88659a 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.test.ts @@ -54,6 +54,14 @@ readline.createInterface({ input: process.stdin, crlfDelay: Infinity }).on("line }); `; +const CLOSED_STDIN_CHILD_SOURCE = String.raw` +const fs = require("node:fs"); + +fs.closeSync(0); +process.stdout.write(JSON.stringify({ type: "stdin_closed" }) + "\n"); +setInterval(() => {}, 1_000); +`; + interface InMemoryChildProcess extends ChildProcessWithoutNullStreams { stdin: PassThrough; stdout: PassThrough; @@ -63,6 +71,7 @@ interface InMemoryChildProcess extends ChildProcessWithoutNullStreams { interface StartProcessOptions { child?: ChildProcessWithoutNullStreams; defaultRequestTimeoutMs?: number; + source?: string; } function createInMemoryChildProcess(): InMemoryChildProcess { @@ -85,7 +94,7 @@ function startProcess(options: StartProcessOptions = {}): JsonlRpcProcess { return new JsonlRpcProcess({ launch: { command: process.execPath, - args: ["-e", CHILD_SOURCE, "--", "resolved-arg"], + args: ["-e", options.source ?? CHILD_SOURCE, "--", "resolved-arg"], cwd: process.cwd(), env: { JSONL_RPC_TEST_VALUE: "resolved-env" }, }, @@ -222,6 +231,70 @@ describe("JsonlRpcProcess", () => { await rejection; }); + test("closes the transport when a direct send synchronously throws EPIPE", async () => { + const child = createInMemoryChildProcess(); + const transport = startProcess({ child }); + + child.stdin.write = () => { + throw Object.assign(new Error("write EPIPE"), { code: "EPIPE" }); + }; + + expect(() => transport.send({ type: "notice" })).not.toThrow(); + await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow( + "JSONL RPC process is closed", + ); + }); + + // Closing fd 0 does not sever the inherited pipe on Windows, so this fixture cannot produce EPIPE. + test.skipIf(process.platform === "win32")( + "keeps the parent alive when a real child closes its stdin pipe", + async () => { + const transport = startProcess({ source: CLOSED_STDIN_CHILD_SOURCE }); + const stdinClosed = new Promise((resolve) => { + const unsubscribe = transport.onMessage((message) => { + if (message.type !== "stdin_closed") return; + unsubscribe(); + resolve(); + }); + }); + + await stdinClosed; + await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow( + /EPIPE|stdin is not writable/, + ); + await expect(transport.request({ type: "echo", value: "later" })).rejects.toThrow( + "JSONL RPC process is closed", + ); + }, + ); + + test("stdin error events close the transport instead of becoming uncaught exceptions", async () => { + const child = createInMemoryChildProcess(); + const transport = startProcess({ child }); + const request = transport.request({ type: "hang" }); + const err = Object.assign(new Error("write EPIPE"), { code: "EPIPE" }); + + child.stdin.emit("error", err); + + await expect(request).rejects.toThrow("write EPIPE"); + await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow( + "JSONL RPC process is closed", + ); + }); + + test("a non-writable stdin closes the transport instead of hanging the request", async () => { + const child = createInMemoryChildProcess(); + const transport = startProcess({ child }); + child.stdin.end(); + + await expect(transport.request({ type: "hang" })).rejects.toThrow( + "JSONL RPC stdin is not writable", + ); + await expect(transport.request({ type: "echo", value: "after" })).rejects.toThrow( + "JSONL RPC process is closed", + ); + }); + test("reassembles protocol v2 chunked responses into the logical frame", async () => { const child = createInMemoryChildProcess(); const transport = startProcess({ child }); diff --git a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts index 0d1c8a436a..791df2500a 100644 --- a/packages/server/src/server/agent/providers/jsonl-rpc-process.ts +++ b/packages/server/src/server/agent/providers/jsonl-rpc-process.ts @@ -104,6 +104,9 @@ export class JsonlRpcProcess { this.stderrBuffer = this.stderrBuffer.slice(-STDERR_BUFFER_LIMIT); } }); + this.child.stdin.on("error", (error) => { + this.handleStdinError(error); + }); this.child.on("error", (error) => { this.failAll(error instanceof Error ? error : new Error(String(error))); }); @@ -173,10 +176,18 @@ export class JsonlRpcProcess { } send(message: Record): void { - if (this.disposed || this.child.stdin.destroyed || !this.child.stdin.writable) { + if (this.disposed) { return; } - this.child.stdin.write(`${JSON.stringify(message)}\n`); + if (this.child.stdin.destroyed || !this.child.stdin.writable) { + this.handleStdinError(new Error(`${this.diagnosticName} stdin is not writable`)); + return; + } + try { + this.child.stdin.write(`${JSON.stringify(message)}\n`); + } catch (error) { + this.handleStdinError(error); + } } async close(error = new Error(`${this.diagnosticName} process is closed`)): Promise { @@ -243,6 +254,15 @@ export class JsonlRpcProcess { pending.resolve(response.data); } + private handleStdinError(error: unknown): void { + if (this.disposed) { + return; + } + const err = error instanceof Error ? error : new Error(String(error)); + this.options.logger.warn({ err }, `${this.diagnosticName} stdin write failed`); + void this.close(err); + } + private failAll(error: Error): void { if (this.disposed) { return;