fix(rpc): reject pending requests on child process exit

closes #4764
2026-05-25 00:40:07 +02:00
parent ce0e801d8e
commit e007fcd0d2
3 changed files with 104 additions and 8 deletions
@@ -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)).

@@ -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<RpcResponse> {
-		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);
+			}
 		});
 	}

@@ -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\)/);
+	});
+});