diff --git a/.changeset/local-process-stdin-error-listener.md b/.changeset/local-process-stdin-error-listener.md new file mode 100644 index 0000000000..b6bd0a49d3 --- /dev/null +++ b/.changeset/local-process-stdin-error-listener.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-sandbox-local-process': patch +--- + +fix: stop an uncaught EPIPE when a write goes to a child that closed its stdin diff --git a/packages/ai-sandbox-local-process/src/handle.ts b/packages/ai-sandbox-local-process/src/handle.ts index 7c0106d209..cb066394be 100644 --- a/packages/ai-sandbox-local-process/src/handle.ts +++ b/packages/ai-sandbox-local-process/src/handle.ts @@ -891,6 +891,11 @@ export class LocalProcessHandle implements SandboxHandle { detached: spawnDetached, }) this.track(child) + // Node reports a failed write twice: to the `write` callback below, and + // as an `error` event on this socket. An unhandled `error` event throws + // and stops the host process. `stdin.write` still rejects with the same + // error. + child.stdin.on('error', () => {}) if (opts?.signal) { opts.signal.addEventListener( 'abort', diff --git a/packages/ai-sandbox-local-process/tests/local-process.test.ts b/packages/ai-sandbox-local-process/tests/local-process.test.ts index dcbcc39dd2..38503a81e2 100644 --- a/packages/ai-sandbox-local-process/tests/local-process.test.ts +++ b/packages/ai-sandbox-local-process/tests/local-process.test.ts @@ -228,6 +228,52 @@ describe('local-process killTree — POSIX child.kill(signal) branch', () => { ) }) +describe('local-process stdin errors', () => { + posixOnly( + 'a write to a child that closed its stdin rejects without an uncaught error (skipped on Windows: named pipes do not report EPIPE the same way)', + async () => { + const sbx = await fresh() + // `finally`: the `sleep 30` below outlives a failed assertion otherwise. + try { + // `exec 0<&-` makes the child close its own stdin while it keeps + // running: the write end stays open, and no process holds the read end. + const proc = await sbx.process.spawn( + 'exec 0<&- ; echo closed ; sleep 30', + ) + // The shell echoes after it closes fd 0. An earlier write only fills the + // pipe buffer and resolves, which would prove nothing. + let closed = false + for await (const chunk of proc.stdout) { + if (chunk.includes('closed')) { + closed = true + break + } + } + expect(closed).toBe(true) + + const uncaught: Array = [] + const onUncaught = (error: Error): void => { + uncaught.push(error) + } + process.on('uncaughtException', onUncaught) + try { + await expect(proc.stdin.write('probe\n')).rejects.toThrow(/EPIPE/) + // Node emits the socket's `error` event on a `nextTick` after the + // callback, and that queue always drains before a `setImmediate`. + await new Promise((resolve) => setImmediate(resolve)) + } finally { + process.off('uncaughtException', onUncaught) + } + + expect(uncaught).toEqual([]) + } finally { + await sbx.destroy() + } + }, + 30_000, + ) +}) + describe('local-process + spawnNdjson (real agent-CLI streaming)', () => { it('streams NDJSON events emitted by a spawned process', async () => { const sbx = await fresh()