diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index c7ae064a..3dd492c0 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/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 diff --git a/packages/coding-agent/src/core/output-guard.ts b/packages/coding-agent/src/core/output-guard.ts index c0783760..8781a51d 100644 --- a/packages/coding-agent/src/core/output-guard.ts +++ b/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 = 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 { + while (true) { + try { + await new Promise((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((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 { + while (true) { + const tail = rawStdoutWriteTail; + await tail; + if (tail === rawStdoutWriteTail) { + return; + } + } } export async function flushRawStdout(): Promise { - if (stdoutTakeoverState) { - await new Promise((resolve, reject) => { - stdoutTakeoverState?.rawStdoutWrite("", (err) => { - if (err) reject(err); - else resolve(); - }); - }); - return; - } - - await new Promise((resolve, reject) => { - process.stdout.write("", (err) => { - if (err) reject(err); - else resolve(); - }); - }); + await waitForRawStdoutBackpressure(); + await writeRawStdoutChunk(""); } diff --git a/packages/coding-agent/src/modes/rpc/rpc-mode.ts b/packages/coding-agent/src/modes/rpc/rpc-mode.ts index 828c9f67..513af70e 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-mode.ts +++ b/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 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 { output(event); }); + unsubscribeBackpressure = session.agent.subscribe(async () => { + await waitForRawStdoutBackpressure(); + }); }; const registerSignalHandlers = (): void => { @@ -357,7 +367,7 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise { 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 {}; - async function shutdown(exitCode = 0): Promise { + async function shutdown(exitCode = 0, signal?: NodeJS.Signals): Promise { if (shuttingDown) { process.exit(exitCode); } @@ -674,9 +684,13 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise ({ })); 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); },