From e007fcd0d2a1e924338c719f2a8ac9b561f15305 Mon Sep 17 00:00:00 2001 From: Armin Ronacher Date: Mon, 25 May 2026 00:40:07 +0200 Subject: [PATCH] fix(rpc): reject pending requests on child process exit closes #4764 --- packages/coding-agent/CHANGELOG.md | 1 + .../coding-agent/src/modes/rpc/rpc-client.ts | 73 +++++++++++++++++-- .../test/rpc-client-process-exit.test.ts | 38 ++++++++++ 3 files changed, 104 insertions(+), 8 deletions(-) create mode 100644 packages/coding-agent/test/rpc-client-process-exit.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 3dd492c0..d8b9c62f 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -4,6 +4,7 @@ ### Fixed +- Fixed `RpcClient` to reject pending requests and consume stdin pipe errors when the child process exits unexpectedly ([#4764](https://github.com/earendil-works/pi/issues/4764)). - Fixed managed npm extension updates to avoid package managers installing or resolving pi host packages as peer dependencies ([#4907](https://github.com/earendil-works/pi/issues/4907)). - Fixed RPC mode raw stdout writes to retry transient backpressure errors and flush queued protocol output during shutdown ([#4897](https://github.com/earendil-works/pi/issues/4897)). diff --git a/packages/coding-agent/src/modes/rpc/rpc-client.ts b/packages/coding-agent/src/modes/rpc/rpc-client.ts index 53b96e64..4c46feff 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-client.ts +++ b/packages/coding-agent/src/modes/rpc/rpc-client.ts @@ -59,6 +59,7 @@ export class RpcClient { new Map(); private requestId = 0; private stderr = ""; + private exitError: Error | null = null; private options: RpcClientOptions; constructor(options: RpcClientOptions = {}) { @@ -73,6 +74,8 @@ export class RpcClient { throw new Error("Client already started"); } + this.exitError = null; + const cliPath = this.options.cliPath ?? "dist/cli.js"; const args = ["--mode", "rpc"]; @@ -86,20 +89,41 @@ export class RpcClient { args.push(...this.options.args); } - this.process = spawn("node", [cliPath, ...args], { + const childProcess = spawn("node", [cliPath, ...args], { cwd: this.options.cwd, env: { ...process.env, ...this.options.env }, stdio: ["pipe", "pipe", "pipe"], }); + this.process = childProcess; // Collect stderr for debugging - this.process.stderr?.on("data", (data) => { + childProcess.stderr?.on("data", (data) => { this.stderr += data.toString(); process.stderr.write(data); }); + childProcess.once("exit", (code, signal) => { + if (this.process !== childProcess) return; + const error = this.createProcessExitError(code, signal); + this.exitError = error; + this.rejectPendingRequests(error); + }); + childProcess.once("error", (error) => { + if (this.process !== childProcess) return; + const processError = new Error(`Agent process error: ${error.message}. Stderr: ${this.stderr}`); + this.exitError = processError; + this.rejectPendingRequests(processError); + }); + childProcess.stdin?.on("error", (error) => { + if (this.process !== childProcess) return; + const stdinError = + this.exitError ?? new Error(`Agent process stdin error: ${error.message}. Stderr: ${this.stderr}`); + this.exitError = stdinError; + this.rejectPendingRequests(stdinError); + }); + // Set up strict JSONL reader for stdout. - this.stopReadingStdout = attachJsonlLineReader(this.process.stdout!, (line) => { + this.stopReadingStdout = attachJsonlLineReader(childProcess.stdout!, (line) => { this.handleLine(line); }); @@ -107,7 +131,9 @@ export class RpcClient { await new Promise((resolve) => setTimeout(resolve, 100)); if (this.process.exitCode !== null) { - throw new Error(`Agent process exited immediately with code ${this.process.exitCode}. Stderr: ${this.stderr}`); + const error = this.exitError ?? this.createProcessExitError(this.process.exitCode, this.process.signalCode); + this.exitError = error; + throw error; } } @@ -474,17 +500,41 @@ export class RpcClient { } } + private createProcessExitError(code: number | null, signal: NodeJS.Signals | null): Error { + return new Error(`Agent process exited (code=${code} signal=${signal}). Stderr: ${this.stderr}`); + } + + private rejectPendingRequests(error: Error): void { + for (const pending of this.pendingRequests.values()) { + pending.reject(error); + } + this.pendingRequests.clear(); + } + private async send(command: RpcCommandBody): Promise { - if (!this.process?.stdin) { + const childProcess = this.process; + const stdin = childProcess?.stdin; + if (!childProcess || !stdin) { throw new Error("Client not started"); } + if (this.exitError) { + throw this.exitError; + } + if (childProcess.exitCode !== null) { + const error = this.createProcessExitError(childProcess.exitCode, childProcess.signalCode); + this.exitError = error; + throw error; + } + if (stdin.destroyed || !stdin.writable) { + const error = new Error(`Agent process stdin is not writable. Stderr: ${this.stderr}`); + this.exitError = error; + throw error; + } const id = `req_${++this.requestId}`; const fullCommand = { ...command, id } as RpcCommand; return new Promise((resolve, reject) => { - this.pendingRequests.set(id, { resolve, reject }); - const timeout = setTimeout(() => { this.pendingRequests.delete(id); reject(new Error(`Timeout waiting for response to ${command.type}. Stderr: ${this.stderr}`)); @@ -501,7 +551,14 @@ export class RpcClient { }, }); - this.process!.stdin!.write(serializeJsonLine(fullCommand)); + try { + stdin.write(serializeJsonLine(fullCommand)); + } catch (error: unknown) { + const writeError = error instanceof Error ? error : new Error(String(error)); + const pending = this.pendingRequests.get(id); + this.pendingRequests.delete(id); + pending?.reject(writeError); + } }); } diff --git a/packages/coding-agent/test/rpc-client-process-exit.test.ts b/packages/coding-agent/test/rpc-client-process-exit.test.ts new file mode 100644 index 00000000..f023849e --- /dev/null +++ b/packages/coding-agent/test/rpc-client-process-exit.test.ts @@ -0,0 +1,38 @@ +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, test } from "vitest"; +import { RpcClient } from "../src/modes/rpc/rpc-client.ts"; + +const tempDirs: string[] = []; + +function writeChildScript(contents: string): string { + const dir = mkdtempSync(join(tmpdir(), "pi-rpc-client-exit-")); + tempDirs.push(dir); + const path = join(dir, "child.mjs"); + writeFileSync(path, contents); + return path; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +describe("RpcClient child process failures", () => { + test("rejects an in-flight request when the child process exits", async () => { + const client = new RpcClient({ + cliPath: writeChildScript(` +process.stdin.once("data", () => { + process.exit(43); +}); +process.stdin.resume(); +`), + }); + + await client.start(); + + await expect(client.getCommands()).rejects.toThrow(/Agent process exited \(code=43 signal=null\)/); + }); +});