fix(coding-agent): retry RPC stdout backpressure

closes #4897

packages/coding-agent/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Fixed
 
 - 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)).
 
 ## [0.75.5] - 2026-05-23
 

packages/coding-agent/src/core/output-guard.ts

@@ -6,6 +6,42 @@ interface StdoutTakeoverState {
 
 let stdoutTakeoverState: StdoutTakeoverState | undefined;
 
+const RAW_STDOUT_RETRY_DELAY_MS = 10;
+
+let rawStdoutWriteTail: Promise<void> = Promise.resolve();
+
+function getRawStdoutWrite(): StdoutTakeoverState["rawStdoutWrite"] {
+	if (stdoutTakeoverState) {
+		return stdoutTakeoverState.rawStdoutWrite;
+	}
+	return process.stdout.write.bind(process.stdout) as StdoutTakeoverState["rawStdoutWrite"];
+}
+
+async function writeRawStdoutChunk(text: string): Promise<void> {
+	while (true) {
+		try {
+			await new Promise<void>((resolve, reject) => {
+				try {
+					getRawStdoutWrite()(text, (error) => {
+						if (error) reject(error);
+						else resolve();
+					});
+				} catch (error) {
+					reject(error instanceof Error ? error : new Error(String(error)));
+				}
+			});
+			return;
+		} catch (error) {
+			const writeError = error instanceof Error ? error : new Error(String(error));
+			const code = (writeError as Error & { code?: unknown }).code;
+			if (code !== "ENOBUFS" && code !== "EAGAIN" && code !== "EWOULDBLOCK") {
+				throw writeError;
+			}
+			await new Promise<void>((resolve) => setTimeout(resolve, RAW_STDOUT_RETRY_DELAY_MS));
+		}
+	}
+}
+
 export function takeOverStdout(): void {
 	if (stdoutTakeoverState) {
 		return;
@@ -47,28 +83,26 @@ export function isStdoutTakenOver(): boolean {
 }
 
 export function writeRawStdout(text: string): void {
-	if (stdoutTakeoverState) {
-		stdoutTakeoverState.rawStdoutWrite(text);
+	if (text.length === 0) {
 		return;
 	}
-	process.stdout.write(text);
+	rawStdoutWriteTail = rawStdoutWriteTail.then(() => writeRawStdoutChunk(text));
+	void rawStdoutWriteTail.catch(() => {
+		process.exit(1);
+	});
+}
+
+export async function waitForRawStdoutBackpressure(): Promise<void> {
+	while (true) {
+		const tail = rawStdoutWriteTail;
+		await tail;
+		if (tail === rawStdoutWriteTail) {
+			return;
+		}
+	}
 }
 
 export async function flushRawStdout(): Promise<void> {
-	if (stdoutTakeoverState) {
-		await new Promise<void>((resolve, reject) => {
-			stdoutTakeoverState?.rawStdoutWrite("", (err) => {
-				if (err) reject(err);
-				else resolve();
-			});
-		});
-		return;
-	}
-
-	await new Promise<void>((resolve, reject) => {
-		process.stdout.write("", (err) => {
-			if (err) reject(err);
-			else resolve();
-		});
-	});
+	await waitForRawStdoutBackpressure();
+	await writeRawStdoutChunk("");
 }

packages/coding-agent/src/modes/rpc/rpc-mode.ts

@@ -19,7 +19,12 @@ import type {
 	ExtensionWidgetOptions,
 	WorkingIndicatorOptions,
 } from "../../core/extensions/index.ts";
-import { takeOverStdout, writeRawStdout } from "../../core/output-guard.ts";
+import {
+	flushRawStdout,
+	takeOverStdout,
+	waitForRawStdoutBackpressure,
+	writeRawStdout,
+} from "../../core/output-guard.ts";
 import { killTrackedDetachedChildren } from "../../utils/shell.ts";
 import { type Theme, theme } from "../interactive/theme/theme.ts";
 import { attachJsonlLineReader, serializeJsonLine } from "./jsonl.ts";
@@ -49,6 +54,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 	takeOverStdout();
 	let session = runtimeHost.session;
 	let unsubscribe: (() => void) | undefined;
+	let unsubscribeBackpressure: (() => void) | undefined;
 
 	const output = (obj: RpcResponse | RpcExtensionUIRequest | object) => {
 		writeRawStdout(serializeJsonLine(obj));
@@ -343,9 +349,13 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 		});
 
 		unsubscribe?.();
+		unsubscribeBackpressure?.();
 		unsubscribe = session.subscribe((event) => {
 			output(event);
 		});
+		unsubscribeBackpressure = session.agent.subscribe(async () => {
+			await waitForRawStdoutBackpressure();
+		});
 	};
 
 	const registerSignalHandlers = (): void => {
@@ -357,7 +367,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 		for (const signal of signals) {
 			const handler = () => {
 				killTrackedDetachedChildren();
-				void shutdown(signal === "SIGHUP" ? 129 : 143);
+				void shutdown(signal === "SIGHUP" ? 129 : 143, signal);
 			};
 			process.on(signal, handler);
 			signalCleanupHandlers.push(() => process.off(signal, handler));
@@ -665,7 +675,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 	 */
 	let detachInput = () => {};
 
-	async function shutdown(exitCode = 0): Promise<never> {
+	async function shutdown(exitCode = 0, signal?: NodeJS.Signals): Promise<never> {
 		if (shuttingDown) {
 			process.exit(exitCode);
 		}
@@ -674,9 +684,13 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 			cleanup();
 		}
 		unsubscribe?.();
+		unsubscribeBackpressure?.();
 		await runtimeHost.dispose();
 		detachInput();
 		process.stdin.pause();
+		if (signal !== "SIGTERM") {
+			await flushRawStdout();
+		}
 		process.exit(exitCode);
 	}
 
@@ -697,6 +711,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 					`Failed to parse command: ${parseError instanceof Error ? parseError.message : String(parseError)}`,
 				),
 			);
+			await waitForRawStdoutBackpressure();
 			return;
 		}
 
@@ -721,6 +736,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 			const response = await handleCommand(command);
 			if (response) {
 				output(response);
+				await waitForRawStdoutBackpressure();
 			}
 			await checkShutdownRequested();
 		} catch (commandError: unknown) {
@@ -731,6 +747,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 					commandError instanceof Error ? commandError.message : String(commandError),
 				),
 			);
+			await waitForRawStdoutBackpressure();
 		}
 	};
 

packages/coding-agent/test/rpc-prompt-response-semantics.test.ts

@@ -25,7 +25,9 @@ const rpcIo = vi.hoisted(() => ({
 }));
 
 vi.mock("../src/core/output-guard.js", () => ({
+	flushRawStdout: vi.fn(async () => {}),
 	takeOverStdout: vi.fn(),
+	waitForRawStdoutBackpressure: vi.fn(async () => {}),
 	writeRawStdout: (line: string) => {
 		rpcIo.outputLines.push(line);
 	},