fix(rpc): respect stdout backpressure

closes #4897

packages/coding-agent/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Fixed
 
+- Fixed RPC mode to respect stdout backpressure while streaming events, avoiding `ENOBUFS` crashes when clients drain stdout slowly ([#4897](https://github.com/earendil-works/pi/issues/4897)).
 - 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)).
 
 ## [0.75.5] - 2026-05-23

packages/coding-agent/docs/sdk.md

@@ -81,7 +81,7 @@ interface AgentSession {
   followUp(text: string): Promise<void>;
 
   // Subscribe to events (returns unsubscribe function)
-  subscribe(listener: (event: AgentSessionEvent) => void): () => void;
+  subscribe(listener: (event: AgentSessionEvent) => void | Promise<void>): () => void;
 
   // Session info
   sessionFile: string | undefined;
@@ -191,7 +191,7 @@ interface PromptOptions {
   images?: ImageContent[];
   streamingBehavior?: "steer" | "followUp";
   source?: InputSource;
-  preflightResult?: (success: boolean) => void;
+  preflightResult?: (success: boolean) => void | Promise<void>;
 }
 ```
 

packages/coding-agent/src/core/agent-session.ts

@@ -147,7 +147,7 @@ export type AgentSessionEvent =
 	| { type: "auto_retry_end"; success: boolean; attempt: number; finalError?: string };
 
 /** Listener function for agent session events */
-export type AgentSessionEventListener = (event: AgentSessionEvent) => void;
+export type AgentSessionEventListener = (event: AgentSessionEvent) => void | Promise<void>;
 
 // ============================================================================
 // Types
@@ -202,7 +202,7 @@ export interface PromptOptions {
 	/** Source of input for extension input event handlers. Defaults to "interactive". */
 	source?: InputSource;
 	/** Internal hook used by RPC mode to observe prompt preflight acceptance or rejection. */
-	preflightResult?: (success: boolean) => void;
+	preflightResult?: (success: boolean) => void | Promise<void>;
 }
 
 /** Result from cycleModel() */
@@ -448,14 +448,18 @@ export class AgentSession {
 	// =========================================================================
 
 	/** Emit an event to all listeners */
-	private _emit(event: AgentSessionEvent): void {
+	private async _emit(event: AgentSessionEvent): Promise<void> {
 		for (const l of this._eventListeners) {
-			l(event);
+			await l(event);
 		}
 	}
 
-	private _emitQueueUpdate(): void {
+	private _emitDetached(event: AgentSessionEvent): void {
-		this._emit({
+		void this._emit(event);
+	}
+
+	private async _emitQueueUpdate(): Promise<void> {
+		await this._emit({
 			type: "queue_update",
 			steering: [...this._steeringMessages],
 			followUp: [...this._followUpMessages],
@@ -477,13 +481,13 @@ export class AgentSession {
 				const steeringIndex = this._steeringMessages.indexOf(messageText);
 				if (steeringIndex !== -1) {
 					this._steeringMessages.splice(steeringIndex, 1);
-					this._emitQueueUpdate();
+					await this._emitQueueUpdate();
 				} else {
 					// Check follow-up queue
 					const followUpIndex = this._followUpMessages.indexOf(messageText);
 					if (followUpIndex !== -1) {
 						this._followUpMessages.splice(followUpIndex, 1);
-						this._emitQueueUpdate();
+						await this._emitQueueUpdate();
 					}
 				}
 			}
@@ -493,7 +497,9 @@ export class AgentSession {
 		await this._emitExtensionEvent(event);
 
 		// Notify all listeners
-		this._emit(event.type === "agent_end" ? { ...event, willRetry: this._willRetryAfterAgentEnd(event) } : event);
+		await this._emit(
+			event.type === "agent_end" ? { ...event, willRetry: this._willRetryAfterAgentEnd(event) } : event,
+		);
 
 		// Handle session persistence
 		if (event.type === "message_end") {
@@ -528,7 +534,7 @@ export class AgentSession {
 				// Reset retry counter immediately on successful assistant response
 				// This prevents accumulation across multiple LLM calls within a turn
 				if (assistantMsg.stopReason !== "error" && this._retryAttempt > 0) {
-					this._emit({
+					await this._emit({
 						type: "auto_retry_end",
 						success: true,
 						attempt: this._retryAttempt,
@@ -938,7 +944,7 @@ export class AgentSession {
 		}
 
 		if (msg.stopReason === "error" && this._retryAttempt > 0) {
-			this._emit({
+			await this._emit({
 				type: "auto_retry_end",
 				success: false,
 				attempt: this._retryAttempt,
@@ -971,7 +977,7 @@ export class AgentSession {
 				const handled = await this._tryExecuteExtensionCommand(text);
 				if (handled) {
 					// Extension command executed, no prompt to send
-					preflightResult?.(true);
+					await preflightResult?.(true);
 					return;
 				}
 			}
@@ -986,7 +992,7 @@ export class AgentSession {
 					options?.source ?? "interactive",
 				);
 				if (inputResult.action === "handled") {
-					preflightResult?.(true);
+					await preflightResult?.(true);
 					return;
 				}
 				if (inputResult.action === "transform") {
@@ -1014,7 +1020,7 @@ export class AgentSession {
 				} else {
 					await this._queueSteer(expandedText, currentImages);
 				}
-				preflightResult?.(true);
+				await preflightResult?.(true);
 				return;
 			}
 
@@ -1099,7 +1105,7 @@ export class AgentSession {
 				this.agent.state.systemPrompt = this._baseSystemPrompt;
 			}
 		} catch (error) {
-			preflightResult?.(false);
+			await preflightResult?.(false);
 			throw error;
 		}
 
@@ -1107,7 +1113,7 @@ export class AgentSession {
 			return;
 		}
 
-		preflightResult?.(true);
+		await preflightResult?.(true);
 		await this._runAgentPrompt(messages);
 	}
 
@@ -1217,7 +1223,7 @@ export class AgentSession {
 	 */
 	private async _queueSteer(text: string, images?: ImageContent[]): Promise<void> {
 		this._steeringMessages.push(text);
-		this._emitQueueUpdate();
+		await this._emitQueueUpdate();
 		const content: (TextContent | ImageContent)[] = [{ type: "text", text }];
 		if (images) {
 			content.push(...images);
@@ -1234,7 +1240,7 @@ export class AgentSession {
 	 */
 	private async _queueFollowUp(text: string, images?: ImageContent[]): Promise<void> {
 		this._followUpMessages.push(text);
-		this._emitQueueUpdate();
+		await this._emitQueueUpdate();
 		const content: (TextContent | ImageContent)[] = [{ type: "text", text }];
 		if (images) {
 			content.push(...images);
@@ -1303,8 +1309,8 @@ export class AgentSession {
 				message.display,
 				message.details,
 			);
-			this._emit({ type: "message_start", message: appMessage });
+			await this._emit({ type: "message_start", message: appMessage });
-			this._emit({ type: "message_end", message: appMessage });
+			await this._emit({ type: "message_end", message: appMessage });
 		}
 	}
 
@@ -1359,7 +1365,7 @@ export class AgentSession {
 		this._steeringMessages = [];
 		this._followUpMessages = [];
 		this.agent.clearAllQueues();
-		this._emitQueueUpdate();
+		void this._emitQueueUpdate();
 		return { steering, followUp };
 	}
 
@@ -1522,7 +1528,7 @@ export class AgentSession {
 			if (this.supportsThinking() || effectiveLevel !== "off") {
 				this.settingsManager.setDefaultThinkingLevel(effectiveLevel);
 			}
-			this._emit({ type: "thinking_level_changed", level: effectiveLevel });
+			this._emitDetached({ type: "thinking_level_changed", level: effectiveLevel });
 			void this._extensionRunner.emit({
 				type: "thinking_level_select",
 				level: effectiveLevel,
@@ -1612,7 +1618,7 @@ export class AgentSession {
 		this._disconnectFromAgent();
 		await this.abort();
 		this._compactionAbortController = new AbortController();
-		this._emit({ type: "compaction_start", reason: "manual" });
+		await this._emit({ type: "compaction_start", reason: "manual" });
 
 		try {
 			if (!this.model) {
@@ -1713,7 +1719,7 @@ export class AgentSession {
 				tokensBefore,
 				details,
 			};
-			this._emit({
+			await this._emit({
 				type: "compaction_end",
 				reason: "manual",
 				result: compactionResult,
@@ -1724,7 +1730,7 @@ export class AgentSession {
 		} catch (error) {
 			const message = error instanceof Error ? error.message : String(error);
 			const aborted = message === "Compaction cancelled" || (error instanceof Error && error.name === "AbortError");
-			this._emit({
+			await this._emit({
 				type: "compaction_end",
 				reason: "manual",
 				result: undefined,
@@ -1794,7 +1800,7 @@ export class AgentSession {
 		// Case 1: Overflow - LLM returned context overflow error
 		if (sameModel && isContextOverflow(assistantMessage, contextWindow)) {
 			if (this._overflowRecoveryAttempted) {
-				this._emit({
+				await this._emit({
 					type: "compaction_end",
 					reason: "overflow",
 					result: undefined,
@@ -1851,12 +1857,12 @@ export class AgentSession {
 	private async _runAutoCompaction(reason: "overflow" | "threshold", willRetry: boolean): Promise<boolean> {
 		const settings = this.settingsManager.getCompactionSettings();
 
-		this._emit({ type: "compaction_start", reason });
+		await this._emit({ type: "compaction_start", reason });
 		this._autoCompactionAbortController = new AbortController();
 
 		try {
 			if (!this.model) {
-				this._emit({
+				await this._emit({
 					type: "compaction_end",
 					reason,
 					result: undefined,
@@ -1871,7 +1877,7 @@ export class AgentSession {
 			if (this.agent.streamFn === streamSimple) {
 				const authResult = await this._modelRegistry.getApiKeyAndHeaders(this.model);
 				if (!authResult.ok || !authResult.apiKey) {
-					this._emit({
+					await this._emit({
 						type: "compaction_end",
 						reason,
 						result: undefined,
@@ -1890,7 +1896,7 @@ export class AgentSession {
 
 			const preparation = prepareCompaction(pathEntries, settings);
 			if (!preparation) {
-				this._emit({
+				await this._emit({
 					type: "compaction_end",
 					reason,
 					result: undefined,
@@ -1913,7 +1919,7 @@ export class AgentSession {
 				})) as SessionBeforeCompactResult | undefined;
 
 				if (extensionResult?.cancel) {
-					this._emit({
+					await this._emit({
 						type: "compaction_end",
 						reason,
 						result: undefined,
@@ -1959,7 +1965,7 @@ export class AgentSession {
 			}
 
 			if (this._autoCompactionAbortController.signal.aborted) {
-				this._emit({
+				await this._emit({
 					type: "compaction_end",
 					reason,
 					result: undefined,
@@ -1993,7 +1999,7 @@ export class AgentSession {
 				tokensBefore,
 				details,
 			};
-			this._emit({ type: "compaction_end", reason, result, aborted: false, willRetry });
+			await this._emit({ type: "compaction_end", reason, result, aborted: false, willRetry });
 
 			if (willRetry) {
 				const messages = this.agent.state.messages;
@@ -2009,7 +2015,7 @@ export class AgentSession {
 			return this.agent.hasQueuedMessages();
 		} catch (error) {
 			const errorMessage = error instanceof Error ? error.message : "compaction failed";
-			this._emit({
+			await this._emit({
 				type: "compaction_end",
 				reason,
 				result: undefined,
@@ -2460,7 +2466,7 @@ export class AgentSession {
 
 		const delayMs = settings.baseDelayMs * 2 ** (this._retryAttempt - 1);
 
-		this._emit({
+		await this._emit({
 			type: "auto_retry_start",
 			attempt: this._retryAttempt,
 			maxAttempts: settings.maxRetries,
@@ -2482,7 +2488,7 @@ export class AgentSession {
 			// Aborted during sleep - emit end event so UI can clean up
 			const attempt = this._retryAttempt;
 			this._retryAttempt = 0;
-			this._emit({
+			await this._emit({
 				type: "auto_retry_end",
 				success: false,
 				attempt,
@@ -2636,7 +2642,7 @@ export class AgentSession {
 	 */
 	setSessionName(name: string): void {
 		this.sessionManager.appendSessionInfo(name);
-		this._emit({ type: "session_info_changed", name: this.sessionManager.getSessionName() });
+		this._emitDetached({ type: "session_info_changed", name: this.sessionManager.getSessionName() });
 	}
 
 	// =========================================================================

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

@@ -1,3 +1,5 @@
+import { once } from "node:events";
+
 interface StdoutTakeoverState {
 	rawStdoutWrite: (chunk: string, callback?: (error?: Error | null) => void) => boolean;
 	rawStderrWrite: (chunk: string, callback?: (error?: Error | null) => void) => boolean;
@@ -46,12 +48,11 @@ export function isStdoutTakenOver(): boolean {
 	return stdoutTakeoverState !== undefined;
 }
 
-export function writeRawStdout(text: string): void {
+export async function writeRawStdout(text: string): Promise<void> {
-	if (stdoutTakeoverState) {
+	const canContinue = stdoutTakeoverState ? stdoutTakeoverState.rawStdoutWrite(text) : process.stdout.write(text);
-		stdoutTakeoverState.rawStdoutWrite(text);
+	if (!canContinue) {
-		return;
+		await once(process.stdout, "drain");
 	}
-	process.stdout.write(text);
 }
 
 export async function flushRawStdout(): Promise<void> {

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

@@ -100,9 +100,9 @@ export async function runPrintMode(runtimeHost: AgentSessionRuntime, options: Pr
 		});
 
 		unsubscribe?.();
-		unsubscribe = session.subscribe((event) => {
+		unsubscribe = session.subscribe(async (event) => {
 			if (mode === "json") {
-				writeRawStdout(`${JSON.stringify(event)}\n`);
+				await writeRawStdout(`${JSON.stringify(event)}\n`);
 			}
 		});
 	};
@@ -111,7 +111,7 @@ export async function runPrintMode(runtimeHost: AgentSessionRuntime, options: Pr
 		if (mode === "json") {
 			const header = session.sessionManager.getHeader();
 			if (header) {
-				writeRawStdout(`${JSON.stringify(header)}\n`);
+				await writeRawStdout(`${JSON.stringify(header)}\n`);
 			}
 		}
 
@@ -137,7 +137,7 @@ export async function runPrintMode(runtimeHost: AgentSessionRuntime, options: Pr
 				} else {
 					for (const content of assistantMsg.content) {
 						if (content.type === "text") {
-							writeRawStdout(`${content.text}\n`);
+							await writeRawStdout(`${content.text}\n`);
 						}
 					}
 				}

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

@@ -50,8 +50,14 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 	let session = runtimeHost.session;
 	let unsubscribe: (() => void) | undefined;
 
-	const output = (obj: RpcResponse | RpcExtensionUIRequest | object) => {
+	const output = async (obj: RpcResponse | RpcExtensionUIRequest | object): Promise<void> => {
-		writeRawStdout(serializeJsonLine(obj));
+		await writeRawStdout(serializeJsonLine(obj));
+	};
+
+	const outputDetached = (obj: RpcResponse | RpcExtensionUIRequest | object): void => {
+		void output(obj).catch((err: unknown) => {
+			process.stderr.write(`RPC output failed: ${err instanceof Error ? err.message : String(err)}\n`);
+		});
 	};
 
 	const success = <T extends RpcCommand["type"]>(
@@ -81,28 +87,30 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 	const signalCleanupHandlers: Array<() => void> = [];
 
 	/** Helper for dialog methods with signal/timeout support */
-	function createDialogPromise<T>(
+	async function createDialogPromise<T>(
 		opts: ExtensionUIDialogOptions | undefined,
 		defaultValue: T,
 		request: Record<string, unknown>,
 		parseResponse: (response: RpcExtensionUIResponse) => T,
 	): Promise<T> {
-		if (opts?.signal?.aborted) return Promise.resolve(defaultValue);
+		if (opts?.signal?.aborted) return defaultValue;
 
 		const id = crypto.randomUUID();
-		return new Promise((resolve, reject) => {
+		let cleanup = () => {};
+		const responsePromise = new Promise<T>((resolve, reject) => {
 			let timeoutId: ReturnType<typeof setTimeout> | undefined;
 
-			const cleanup = () => {
-				if (timeoutId) clearTimeout(timeoutId);
-				opts?.signal?.removeEventListener("abort", onAbort);
-				pendingExtensionRequests.delete(id);
-			};
-
 			const onAbort = () => {
 				cleanup();
 				resolve(defaultValue);
 			};
+
+			cleanup = () => {
+				if (timeoutId) clearTimeout(timeoutId);
+				opts?.signal?.removeEventListener("abort", onAbort);
+				pendingExtensionRequests.delete(id);
+			};
+
 			opts?.signal?.addEventListener("abort", onAbort, { once: true });
 
 			if (opts?.timeout) {
@@ -117,10 +125,20 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 					cleanup();
 					resolve(parseResponse(response));
 				},
-				reject,
+				reject: (error) => {
+					cleanup();
+					reject(error);
+				},
 			});
-			output({ type: "extension_ui_request", id, ...request } as RpcExtensionUIRequest);
 		});
+
+		try {
+			await output({ type: "extension_ui_request", id, ...request } as RpcExtensionUIRequest);
+		} catch (err) {
+			cleanup();
+			throw err;
+		}
+		return await responsePromise;
 	}
 
 	/**
@@ -144,7 +162,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 
 		notify(message: string, type?: "info" | "warning" | "error"): void {
 			// Fire and forget - no response needed
-			output({
+			outputDetached({
 				type: "extension_ui_request",
 				id: crypto.randomUUID(),
 				method: "notify",
@@ -160,7 +178,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 
 		setStatus(key: string, text: string | undefined): void {
 			// Fire and forget - no response needed
-			output({
+			outputDetached({
 				type: "extension_ui_request",
 				id: crypto.randomUUID(),
 				method: "setStatus",
@@ -188,7 +206,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 		setWidget(key: string, content: unknown, options?: ExtensionWidgetOptions): void {
 			// Only support string arrays in RPC mode - factory functions are ignored
 			if (content === undefined || Array.isArray(content)) {
-				output({
+				outputDetached({
 					type: "extension_ui_request",
 					id: crypto.randomUUID(),
 					method: "setWidget",
@@ -210,7 +228,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 
 		setTitle(title: string): void {
 			// Fire and forget - host can implement terminal title control
-			output({
+			outputDetached({
 				type: "extension_ui_request",
 				id: crypto.randomUUID(),
 				method: "setTitle",
@@ -230,7 +248,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 
 		setEditorText(text: string): void {
 			// Fire and forget - host can implement editor control
-			output({
+			outputDetached({
 				type: "extension_ui_request",
 				id: crypto.randomUUID(),
 				method: "set_editor_text",
@@ -246,9 +264,14 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 
 		async editor(title: string, prefill?: string): Promise<string | undefined> {
 			const id = crypto.randomUUID();
-			return new Promise((resolve, reject) => {
+			let cleanup = () => {};
+			const responsePromise = new Promise<string | undefined>((resolve, reject) => {
+				cleanup = () => {
+					pendingExtensionRequests.delete(id);
+				};
 				pendingExtensionRequests.set(id, {
 					resolve: (response: RpcExtensionUIResponse) => {
+						cleanup();
 						if ("cancelled" in response && response.cancelled) {
 							resolve(undefined);
 						} else if ("value" in response) {
@@ -257,10 +280,25 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 							resolve(undefined);
 						}
 					},
-					reject,
+					reject: (error) => {
+						cleanup();
+						reject(error);
+					},
 				});
-				output({ type: "extension_ui_request", id, method: "editor", title, prefill } as RpcExtensionUIRequest);
 			});
+			try {
+				await output({
+					type: "extension_ui_request",
+					id,
+					method: "editor",
+					title,
+					prefill,
+				} as RpcExtensionUIRequest);
+			} catch (err) {
+				cleanup();
+				throw err;
+			}
+			return await responsePromise;
 		},
 
 		addAutocompleteProvider(): void {
@@ -338,13 +376,18 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 				shutdownRequested = true;
 			},
 			onError: (err) => {
-				output({ type: "extension_error", extensionPath: err.extensionPath, event: err.event, error: err.error });
+				outputDetached({
+					type: "extension_error",
+					extensionPath: err.extensionPath,
+					event: err.event,
+					error: err.error,
+				});
 			},
 		});
 
 		unsubscribe?.();
-		unsubscribe = session.subscribe((event) => {
+		unsubscribe = session.subscribe(async (event) => {
-			output(event);
+			await output(event);
 		});
 	};
 
@@ -385,16 +428,17 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 						images: command.images,
 						streamingBehavior: command.streamingBehavior,
 						source: "rpc",
-						preflightResult: (didSucceed) => {
+						preflightResult: async (didSucceed) => {
 							if (didSucceed) {
+								await output(success(id, "prompt"));
 								preflightSucceeded = true;
-								output(success(id, "prompt"));
 							}
 						},
 					})
-					.catch((e) => {
+					.catch((err: unknown) => {
 						if (!preflightSucceeded) {
-							output(error(id, "prompt", e.message));
+							const message = err instanceof Error ? err.message : String(err);
+							outputDetached(error(id, "prompt", message));
 						}
 					});
 				return undefined;
@@ -690,7 +734,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 		try {
 			parsed = JSON.parse(line);
 		} catch (parseError: unknown) {
-			output(
+			await output(
 				error(
 					undefined,
 					"parse",
@@ -720,11 +764,11 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 		try {
 			const response = await handleCommand(command);
 			if (response) {
-				output(response);
+				await output(response);
 			}
 			await checkShutdownRequested();
 		} catch (commandError: unknown) {
-			output(
+			await output(
 				error(
 					command.id,
 					command.type,
@@ -741,7 +785,9 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 
 	detachInput = (() => {
 		const detachJsonl = attachJsonlLineReader(process.stdin, (line) => {
-			void handleInputLine(line);
+			void handleInputLine(line).catch((err: unknown) => {
+				process.stderr.write(`RPC command handling failed: ${err instanceof Error ? err.message : String(err)}\n`);
+			});
 		});
 		return () => {
 			detachJsonl();