fix: respect output stream backpressure

CODEBASE.md

@@ -60,7 +60,8 @@ src/
 ├─ index.ts               # Root barrel (the `.` export)
 ├─ errors.ts              # CliError<TCode> + CliErrorCode aggregator + getErrorMessage
 ├─ config.ts              # XDG config I/O; CoreConfig / UpdateChannel / ConfigErrorCode
-├─ json.ts                # formatJson / formatNdjson (throw on non-serializable)
+├─ json.ts                # formatJson / formatNdjson / outputNdjson
+├─ stream.ts              # Internal backpressure-aware buffered line writer
 ├─ options.ts             # ViewOptions type + emitView (json/ndjson/human dispatch)
 ├─ empty.ts               # printEmpty (machine-aware empty-state output)
 ├─ global-args.ts         # parseGlobalArgs + spinner/accessible gate factories + stripUserFlag

src/auth/account.test.ts

@@ -3,7 +3,11 @@ import { describe, expect, it, vi } from 'vitest'
 
 import { CliError } from '../errors.js'
 import { formatJson, formatNdjson } from '../json.js'
-import { buildProgram, installCapturedConsole } from '../test-support/cli-harness.js'
+import {
+    buildProgram,
+    installCapturedConsole,
+    installCapturedStream,
+} from '../test-support/cli-harness.js'
 import {
     type TestAccount as Account,
     alanGrant,
@@ -77,6 +81,7 @@ function buildRemove(
 
 describe('attachAccountListCommand', () => {
     const logSpy = installCapturedConsole()
+    const stdoutSpy = installCapturedStream()
 
     it('renders default human lines with a (default) marker only on the default entry', async () => {
         const { program } = buildList()
@@ -168,11 +173,12 @@ describe('attachAccountListCommand', () => {
 
         await program.parseAsync(['node', 'cli', 'account', 'list', '--ndjson'])
 
-        const emitted = logSpy().mock.calls.map((call: unknown[]) => call[0])
-        expect(emitted).toEqual([
-            formatNdjson([{ account: alanGrant, isDefault: true }]),
-            formatNdjson([{ account: ellieSattler, isDefault: false }]),
-        ])
+        expect(stdoutSpy()).toHaveBeenCalledWith(
+            `${formatNdjson([
+                { account: alanGrant, isDefault: true },
+                { account: ellieSattler, isDefault: false },
+            ])}\n`,
+        )
     })
 
     it('shapes each --ndjson line via renderJson, matching the --json accounts entries', async () => {
@@ -196,11 +202,12 @@ describe('attachAccountListCommand', () => {
             isDefault: false,
             flags: {},
         })
-        const emitted = logSpy().mock.calls.map((call: unknown[]) => call[0])
-        expect(emitted).toEqual([
-            formatNdjson([{ name: 'Alan Grant', isDefault: true }]),
-            formatNdjson([{ name: 'Ellie Sattler', isDefault: false }]),
-        ])
+        expect(stdoutSpy()).toHaveBeenCalledWith(
+            `${formatNdjson([
+                { name: 'Alan Grant', isDefault: true },
+                { name: 'Ellie Sattler', isDefault: false },
+            ])}\n`,
+        )
     })
 
     it('prefers --json over --ndjson when both flags are passed', async () => {
@@ -426,6 +433,7 @@ describe('attachAccountUseCommand', () => {
 
 describe('attachAccountCurrentCommand', () => {
     const logSpy = installCapturedConsole()
+    const stdoutSpy = installCapturedStream()
 
     it('renders the default human line with a (default) marker for the active account', async () => {
         const { program } = buildCurrent()
@@ -491,8 +499,8 @@ describe('attachAccountCurrentCommand', () => {
 
         await program.parseAsync(['node', 'cli', 'account', 'current', '--ndjson'])
 
-        expect(logSpy()).toHaveBeenCalledWith(
-            formatNdjson([{ account: alanGrant, isDefault: true }]),
+        expect(stdoutSpy()).toHaveBeenCalledWith(
+            `${formatNdjson([{ account: alanGrant, isDefault: true }])}\n`,
         )
     })
 

src/auth/account.ts

@@ -1,6 +1,6 @@
 import type { Command } from 'commander'
 import { CliError } from '../errors.js'
-import { formatJson, formatNdjson } from '../json.js'
+import { formatJson, outputNdjson } from '../json.js'
 import { type ViewOptions, emitView } from '../options.js'
 import type {
     AccountRef,
@@ -172,15 +172,15 @@ export function attachAccountListCommand<TAccount extends AuthAccount = AuthAcco
             // produced rather than buffering a joined string. `--json` wins when
             // both flags are set. Empty list → no lines (EOF-as-end-of-stream).
             if (view.ndjson && !view.json) {
-                for (const entry of accounts) console.log(formatNdjson([toPayload(entry)]))
+                await outputNdjson(accounts.map(toPayload))
                 return
             }
             // `renderJson` is machine-mode only, so build the payload lazily —
             // emitView ignores it in human mode where the thunk runs instead.
             const payload = view.json
                 ? { accounts: accounts.map(toPayload), default: defaultRef }
                 : {}
-            emitView(view, payload, () => {
+            await emitView(view, payload, () => {
                 const ctx: AttachAccountListContext<TAccount> = {
                     accounts,
                     default: defaultRef,
@@ -224,7 +224,7 @@ export function attachAccountUseCommand<TAccount extends AuthAccount = AuthAccou
                     ? ((await options.store.list()).find((entry) => entry.isDefault)?.account.id ??
                       ref)
                     : ref
-                emitView(view, { ok: true, default: resolvedDefault }, () => [
+                await emitView(view, { ok: true, default: resolvedDefault }, () => [
                     `✓ Default account set to ${ref}`,
                 ])
             }
@@ -307,7 +307,11 @@ export function attachAccountCurrentCommand<TAccount extends AuthAccount = AuthA
                     { account: ctx.account, isDefault: ctx.isDefault, flags },
                     options.renderJson,
                 )
-                console.log(view.json ? formatJson(payload) : formatNdjson([payload]))
+                if (view.json) {
+                    console.log(formatJson(payload))
+                } else {
+                    await outputNdjson([payload])
+                }
                 return
             }
             const text = options.renderText
@@ -389,7 +393,7 @@ export function attachAccountRemoveCommand<TAccount extends AuthAccount = AuthAc
             // action convention); human runs the thunk. The guard skips the
             // silent case so `emitView`'s own `--ndjson` branch never fires.
             if (view.json || !view.ndjson) {
-                emitView(view, { ok: true, removed: cleared.account.id }, () => {
+                await emitView(view, { ok: true, removed: cleared.account.id }, () => {
                     const name = cleared.account.label ?? cleared.account.id
                     const removedLine = `✓ Removed ${name}${ctx.wasDefault ? ' (default)' : ''}`
                     const text = options.renderText ? options.renderText(ctx) : removedLine

src/auth/status.test.ts

@@ -3,7 +3,11 @@ import { describe, expect, it, vi } from 'vitest'
 
 import { CliError } from '../errors.js'
 import { formatJson, formatNdjson } from '../json.js'
-import { buildProgram, installCapturedConsole } from '../test-support/cli-harness.js'
+import {
+    buildProgram,
+    installCapturedConsole,
+    installCapturedStream,
+} from '../test-support/cli-harness.js'
 import {
     type TestAccount as Account,
     type TokenStoreHarness,
@@ -44,6 +48,7 @@ function build(
 
 describe('attachStatusCommand', () => {
     const logSpy = installCapturedConsole()
+    const stdoutSpy = installCapturedStream()
 
     it('emits renderText output in plain mode', async () => {
         const { program, renderText } = build()
@@ -97,7 +102,7 @@ describe('attachStatusCommand', () => {
 
         await program.parseAsync(['node', 'cli', 'auth', 'status', '--ndjson'])
 
-        expect(logSpy()).toHaveBeenCalledWith(formatNdjson([account]))
+        expect(stdoutSpy()).toHaveBeenCalledWith(`${formatNdjson([account])}\n`)
     })
 
     it('does not invoke renderJson in human mode', async () => {

src/auth/status.ts

@@ -1,6 +1,6 @@
 import type { Command } from 'commander'
 import { CliError } from '../errors.js'
-import { formatJson, formatNdjson } from '../json.js'
+import { formatJson, outputNdjson } from '../json.js'
 import type { ViewOptions } from '../options.js'
 import type {
     AccountRef,
@@ -149,7 +149,7 @@ export function attachStatusCommand<TAccount extends AuthAccount = AuthAccount>(
         }
         if (view.ndjson) {
             const payload = options.renderJson ? options.renderJson({ account, flags }) : account
-            console.log(formatNdjson([payload]))
+            await outputNdjson([payload])
             return
         }
         const text = options.renderText({ account, view, flags })

src/commands/update.test.ts

@@ -79,10 +79,14 @@ function mockSpawnError(error: Error) {
 }
 
 let consoleSpy: ReturnType<typeof vi.spyOn>
+let stdoutSpy: ReturnType<typeof vi.spyOn>
 
 beforeEach(() => {
     chalk.level = 0
     consoleSpy = vi.spyOn(console, 'log').mockImplementation(() => {})
+    stdoutSpy = vi
+        .spyOn(process.stdout, 'write')
+        .mockImplementation((() => true) as typeof process.stdout.write)
     mockReadConfigOrThrow.mockReset().mockResolvedValue({})
     mockUpdateConfigOrThrow.mockReset().mockResolvedValue(undefined)
     mockSpawn.mockClear()
@@ -187,7 +191,11 @@ describe('update --check', () => {
     ])('emits machine envelope under %s', async (flag, parse) => {
         mockFetchOk('99.99.99')
         await createProgram().parseAsync(['node', 'td', 'update', '--check', flag])
-        expect(parse(consoleSpy.mock.calls[0][0] as string)).toEqual({
+        const output =
+            flag === '--ndjson'
+                ? stdoutSpy.mock.calls.map((call: unknown[]) => String(call[0])).join('')
+                : String(consoleSpy.mock.calls[0][0])
+        expect(parse(output)).toEqual({
             currentVersion: '1.0.0',
             latestVersion: '99.99.99',
             channel: 'stable',

src/commands/update.ts

@@ -274,7 +274,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions):
     const channel = await getConfiguredUpdateChannel(options.configPath)
 
     if (cmd.channel) {
-        emitView(view, { channel }, () => [`Update channel: ${formatChannel(channel)}`])
+        await emitView(view, { channel }, () => [`Update channel: ${formatChannel(channel)}`])
         return
     }
 
@@ -314,7 +314,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions):
     const updateAvailable = !upToDate && isNewer(currentVersion, latestVersion)
 
     if (cmd.check) {
-        emitView(view, { currentVersion, latestVersion, channel, updateAvailable }, () => {
+        await emitView(view, { currentVersion, latestVersion, channel, updateAvailable }, () => {
             const channelLine = `  Channel: ${formatChannel(channel)}`
             const headline = upToDate
                 ? `${chalk.green('✓')} Already up to date (v${currentVersion})`
@@ -327,7 +327,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions):
     }
 
     if (upToDate) {
-        emitView(view, { currentVersion, latestVersion, channel, installed: false }, () => [
+        await emitView(view, { currentVersion, latestVersion, channel, installed: false }, () => [
             `${chalk.green('✓')} Already up to date${label} (v${currentVersion})`,
         ])
         return
@@ -405,7 +405,7 @@ async function runUpdate(options: UpdateCommandOptions, cmd: UpdateCmdOptions):
         via: brew ? ('brew' as const) : pm,
         ...(brew && installedVersion ? { installedVersion } : {}),
     }
-    emitView(view, summary, () => {
+    await emitView(view, summary, () => {
         const lines = [
             `${chalk.green('✓')} ${brew ? 'brew upgrade complete' : `Updated to v${latestVersion}`}${label}`,
         ]
@@ -442,7 +442,7 @@ async function runSwitch(
 
     await updateConfigOrThrow<CoreConfig>(options.configPath, { update_channel: channel })
 
-    emitView(view, { channel }, () => {
+    await emitView(view, { channel }, () => {
         if (channel === 'pre-release') {
             return [
                 `${chalk.green('✓')} Update channel set to ${formatChannel(channel)}`,

src/ids.test.ts

@@ -16,27 +16,27 @@ describe('outputIds', () => {
         vi.restoreAllMocks()
     })
 
-    it('writes IDs to stdout in one block', () => {
-        const log = vi.spyOn(console, 'log').mockImplementation(() => {})
+    it('writes IDs to stdout in one block', async () => {
+        const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true)
 
-        outputIds([{ id: 'a' }, { id: 'b' }], (item) => item.id)
+        await outputIds([{ id: 'a' }, { id: 'b' }], (item) => item.id)
 
-        expect(log).toHaveBeenCalledOnce()
-        expect(log).toHaveBeenCalledWith('a\nb')
+        expect(write).toHaveBeenCalledOnce()
+        expect(write).toHaveBeenCalledWith('a\nb\n')
     })
 
-    it('writes nothing to stdout for no results', () => {
-        const log = vi.spyOn(console, 'log').mockImplementation(() => {})
+    it('writes nothing to stdout for no results', async () => {
+        const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true)
 
-        outputIds([], (item: { id: string }) => item.id)
+        await outputIds([], (item: { id: string }) => item.id)
 
-        expect(log).not.toHaveBeenCalled()
+        expect(write).not.toHaveBeenCalled()
     })
 
-    it('writes pagination notices to stderr', () => {
+    it('writes pagination notices to stderr', async () => {
         const error = vi.spyOn(console, 'error').mockImplementation(() => {})
 
-        outputIds([], (item: { id: string }) => item.id, 'More results exist.')
+        await outputIds([], (item: { id: string }) => item.id, 'More results exist.')
 
         expect(error).toHaveBeenCalledWith('More results exist.')
     })

src/ids.ts

@@ -1,18 +1,19 @@
+import { writeLines } from './stream.js'
+
 /** Format stable IDs as one value per line, with no trailing newline. */
 export function formatIds<T>(items: readonly T[], getId: (item: T) => string | number): string {
     return items.map((item) => String(getId(item))).join('\n')
 }
 
 /**
- * Write stable IDs to stdout and an optional pagination notice to stderr.
- * Empty results write nothing to stdout.
+ * Write stable IDs to stdout in bounded chunks, waiting when the stream applies
+ * backpressure. Empty results write nothing; pagination notices go to stderr.
  */
-export function outputIds<T>(
-    items: readonly T[],
+export async function outputIds<T>(
+    items: Iterable<T>,
     getId: (item: T) => string | number,
     paginationNotice = '',
-): void {
-    const output = formatIds(items, getId)
-    if (output) console.log(output)
+): Promise<void> {
+    await writeLines(items, (item) => String(getId(item)))
     if (paginationNotice) console.error(paginationNotice)
 }

src/index.ts

@@ -33,7 +33,7 @@ export type {
     GlobalArgsStore,
     SpinnerGateOptions,
 } from './global-args.js'
-export { formatJson, formatNdjson } from './json.js'
+export { formatJson, formatNdjson, outputNdjson } from './json.js'
 export { formatIds, outputIds } from './ids.js'
 export { emitView, OUTPUT_MODES, resolveOutputMode } from './options.js'
 export type { ListViewOptions, OutputMode, ViewOptions } from './options.js'

src/json.test.ts

@@ -1,6 +1,6 @@
-import { describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 
-import { formatJson, formatNdjson } from './json.js'
+import { formatJson, formatNdjson, outputNdjson } from './json.js'
 
 describe('formatJson', () => {
     it('pretty-prints objects with 2-space indentation', () => {
@@ -69,3 +69,34 @@ describe('formatNdjson', () => {
         expect(() => formatNdjson([Symbol('x')])).toThrow(/index 0/)
     })
 })
+
+describe('outputNdjson', () => {
+    afterEach(() => {
+        vi.restoreAllMocks()
+    })
+
+    it('writes NDJSON to stdout in one block', async () => {
+        const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true)
+
+        await outputNdjson([{ a: 1 }, { a: 2 }])
+
+        expect(write).toHaveBeenCalledOnce()
+        expect(write).toHaveBeenCalledWith('{"a":1}\n{"a":2}\n')
+    })
+
+    it('writes nothing for an empty iterable', async () => {
+        const write = vi.spyOn(process.stdout, 'write').mockReturnValue(true)
+
+        await outputNdjson([])
+
+        expect(write).not.toHaveBeenCalled()
+    })
+
+    it('reports the index of a non-serializable item', async () => {
+        vi.spyOn(process.stdout, 'write').mockReturnValue(true)
+
+        await expect(outputNdjson([1, undefined, 2])).rejects.toThrow(
+            /index 1.*not JSON-serializable/,
+        )
+    })
+})