fix(cli): a broken pipe always wins and exits 0; deterministic EPIPE tests

The EPIPE test flaked once on the macOS arm64 lane (exit 1). The handler only knew
EPIPE, but a stream already torn down by the first failed write can follow it with
ERR_STREAM_DESTROYED or ECONNRESET, a write can throw instead of emitting, and the
error can surface as an uncaught exception. All of those now exit 0 at once (stream
error events, io.out/io.err writes, uncaughtException, the top-level catch), so no later
code can turn a closed pipe into a failure.

The tests no longer depend on timing: one closes the read end before the child has
started (its first write meets a closed pipe), one after the first chunk of a 5,000
record answer, both with a 60 s timeout. 50 runs under nice -n 19 with twice as many
busy loops as CPUs: 0 failures.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

packages/cli/src/context.ts

@@ -14,31 +14,55 @@ import { Assistant, Collaboration, Printing, missingQBitMessage, runProcess, typ
 import { openRealtimeStream } from '@qrun-io/qqq-ui-api-client'
 import { CLIENT_CHANNEL, authenticatedFetch, clientFor, currentToken, selectTarget, sessionCookieHeaderFor, target, type Target } from './session.js'
 
+/** A write that ends the process quietly (status 0) when it throws because the reader has gone. */
+function safeWrite(stream: NodeJS.WriteStream, text: string): void {
+  try {
+    stream.write(text)
+  } catch (error) {
+    if (isBrokenPipe(error)) process.exit(0)
+    throw error
+  }
+}
+
 /**
  * Where output goes. The CLI writes to the process streams; `runCli` (the TUI's `:` command line)
  * points these at a pane for the duration of one command.
  */
 export const io = {
-  out: (text: string): void => { process.stdout.write(text) },
-  err: (text: string): void => { process.stderr.write(text) },
+  out: (text: string): void => { safeWrite(process.stdout, text) },
+  err: (text: string): void => { safeWrite(process.stderr, text) },
   isTTY: (): boolean => Boolean(process.stdout.isTTY),
   /** The terminal width table output fits nested results in; undefined when not on a terminal. */
   width: (): number | undefined => (process.stdout.isTTY ? process.stdout.columns : undefined),
 }
 
+/** What a write to a pipe whose reader has gone ends with: EPIPE, or (macOS, a stream already torn down) the errors that follow it. */
+const BROKEN_PIPE_CODES = new Set(['EPIPE', 'ECONNRESET', 'ERR_STREAM_DESTROYED', 'ERR_STREAM_WRITE_AFTER_END'])
+
 /** Whether an error is a write to a pipe whose reader has gone (`qqq table query | head -1`). */
 export function isBrokenPipe(error: unknown): boolean {
-  return (error as NodeJS.ErrnoException | undefined)?.code === 'EPIPE'
+  return BROKEN_PIPE_CODES.has(String((error as NodeJS.ErrnoException | undefined)?.code))
 }
 
-/** Exits quietly with status 0 when stdout or stderr is a pipe nobody reads any more, as `yes | head` does. */
-export function exitOnBrokenPipe(streams: Array<NodeJS.WriteStream> = [process.stdout, process.stderr], exit: (code: number) => void = (code) => process.exit(code)): void {
+/**
+ * A pipe nobody reads any more ends the command at once with status 0, as `yes | head` does, whatever
+ * else was going on: the process exits from the stream's error event, from a write that throws, and
+ * from an uncaught error that is one, so no later code can turn it into a failure.
+ */
+export function exitOnBrokenPipe(streams: Array<NodeJS.WriteStream> = [process.stdout, process.stderr], exit: (code: number) => void = (code) => process.exit(code), on: (event: 'uncaughtException', listener: (error: Error) => void) => unknown = (event, listener) => process.on(event, listener)): void {
   for (const stream of streams) {
     stream.on('error', (error) => {
       if (isBrokenPipe(error)) exit(0)
       else throw error
     })
   }
+  on('uncaughtException', (error) => {
+    if (isBrokenPipe(error)) exit(0)
+    else {
+      process.stderr.write(`${error.stack ?? error.message}\n`)
+      process.exit(1)
+    }
+  })
 }
 
 /** Exit codes, as in the design doc (KofTwentyTwo ADR-0031 shape). */

packages/cli/src/main.ts

@@ -33,10 +33,7 @@ importLegacyCredentials().then(() => program.parseAsync()).catch((error: unknown
     return
   }
   // a stream that failed with EPIPE (an export piped into head): the reader left, nothing to report
-  if (isBrokenPipe(error)) {
-    process.exitCode = 0
-    return
-  }
+  if (isBrokenPipe(error)) process.exit(0)
   const { code, text } = describeError(error, Boolean(program.opts().trace))
   process.stderr.write(text + '\n')
   process.exitCode = code

packages/cli/src/test/epipe.test.ts

@@ -20,7 +20,7 @@ describe('broken pipes', () => {
     stub = await startStub([...BASE_ROUTES,
       [/^GET \/metaData$/, () => ({ tables: { big: { name: 'big', label: 'Big' } }, processes: {}, apps: {} })],
       [/^GET \/metaData\/table\/big$/, () => ({ name: 'big', label: 'Big', primaryKeyField: 'id', capabilities: ['TABLE_QUERY'], fields: { id: { name: 'id', label: 'Id', type: 'INTEGER' }, note: { name: 'note', label: 'Note', type: 'STRING' } } })],
-      [/^POST \/table\/big\/query$/, () => ({ records: Array.from({ length: 2000 }, (_, i) => ({ values: { id: i, note: 'x'.repeat(200) } })) })],
+      [/^POST \/table\/big\/query$/, () => ({ records: Array.from({ length: 5000 }, (_, i) => ({ values: { id: i, note: 'x'.repeat(200) } })) })],
     ])
   })
   after(async () => {
@@ -30,23 +30,40 @@ describe('broken pipes', () => {
 
   test('the helper exits 0 on EPIPE and rethrows anything else', () => {
     assert.ok(isBrokenPipe(Object.assign(new Error('write EPIPE'), { code: 'EPIPE' })))
+    assert.ok(isBrokenPipe(Object.assign(new Error('destroyed'), { code: 'ERR_STREAM_DESTROYED' })))
     assert.ok(!isBrokenPipe(new Error('other')))
     const listeners: Array<(e: Error) => void> = []
     const exits: number[] = []
-    exitOnBrokenPipe([{ on: (_: string, f: (e: Error) => void) => listeners.push(f) } as never], (code) => exits.push(code))
+    let uncaught: ((e: Error) => void) | undefined
+    exitOnBrokenPipe([{ on: (_: string, f: (e: Error) => void) => listeners.push(f) } as never], (code) => exits.push(code), (_, f) => { uncaught = f })
     listeners[0]!(Object.assign(new Error('write EPIPE'), { code: 'EPIPE' }))
-    assert.deepEqual(exits, [0])
+    uncaught!(Object.assign(new Error('write EPIPE'), { code: 'EPIPE' }))
+    assert.deepEqual(exits, [0, 0])
     assert.throws(() => listeners[0]!(new Error('disk full')), /disk full/)
   })
 
-  test('a closed stdout ends the command quietly with status 0 and no stack trace', async () => {
+  /** Runs a query whose output is far larger than a pipe buffer, closing the read end when `closeAt` says so. */
+  const query = async (closeAt: 'before-any-write' | 'first-chunk') => {
     const env = { ...process.env, ...box.env, QQQ_TOKEN: 't', NO_COLOR: '1' }
-    const child = spawn(process.execPath, [CLI, '--server', stub.url, 'table', 'query', 'big', '-O', 'json', '-l', '2000'], { env, stdio: ['ignore', 'pipe', 'pipe'] })
+    const child = spawn(process.execPath, [CLI, '--server', stub.url, 'table', 'query', 'big', '-O', 'json', '-l', '5000'], { env, stdio: ['ignore', 'pipe', 'pipe'] })
     let stderr = ''
     child.stderr.on('data', (d: Buffer) => (stderr += d.toString('utf8')))
-    child.stdout.once('data', () => child.stdout.destroy())
+    // before-any-write: the reader is gone before the child has even started, so its first write meets a closed pipe
+    if (closeAt === 'before-any-write') child.stdout.destroy()
+    else child.stdout.once('data', () => child.stdout.destroy())
     const code = await new Promise<number | null>((resolve) => child.on('close', resolve))
-    assert.equal(code, 0)
+    return { code, stderr }
+  }
+
+  test('a reader that is gone before the first write ends the command quietly with status 0', { timeout: 60_000 }, async () => {
+    const { code, stderr } = await query('before-any-write')
+    assert.equal(code, 0, stderr)
+    assert.doesNotMatch(stderr, /EPIPE|at .*\.js/)
+  })
+
+  test('a reader that leaves after the first chunk ends the command quietly with status 0', { timeout: 60_000 }, async () => {
+    const { code, stderr } = await query('first-chunk')
+    assert.equal(code, 0, stderr)
     assert.doesNotMatch(stderr, /EPIPE|at .*\.js/)
   })
 })