fix(bg): stream session logs with bounded memory (#1762)

* fix(bg): stream session logs with bounded memory

* fix(bg): handle log follow cleanup edge cases

* fix(bg): surface non-follow log read errors

* test(bg): isolate log streaming temp dirs

src/cli/bg.test.ts

@@ -1,12 +1,91 @@
+import { EventEmitter } from 'node:events'
+import { mkdtemp, rm, unlink, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
 import { describe, expect, it } from 'bun:test'
 import {
   buildBackgroundSessionLaunch,
   buildBackgroundChildProcessConfig,
+  followLogFile,
+  printExistingLog,
   terminateBackgroundProcessTree,
+  LOG_STREAM_CHUNK_SIZE,
   parseBackgroundInvocation,
   parseLogsInvocation,
 } from './bg.js'
 
+class TestOutput extends EventEmitter {
+  chunks: Buffer[] = []
+  destroyed = false
+  writableDestroyed = false
+  writeResults: boolean[] = []
+  writeError: unknown
+
+  write(chunk: Uint8Array): boolean {
+    if (this.writeError) throw this.writeError
+    if (this.destroyed || this.writableDestroyed) {
+      throw Object.assign(new Error('stdout closed'), { code: 'EPIPE' })
+    }
+    this.chunks.push(Buffer.from(chunk))
+    return this.writeResults.shift() ?? true
+  }
+
+  bytes(): Buffer {
+    return Buffer.concat(this.chunks)
+  }
+}
+
+function createManualScheduler() {
+  let intervalCallback: (() => void) | undefined
+  let cleared = false
+
+  return {
+    setInterval(callback: () => void): ReturnType<typeof setInterval> {
+      intervalCallback = callback
+      return 1 as unknown as ReturnType<typeof setInterval>
+    },
+    clearInterval(): void {
+      cleared = true
+    },
+    tick(): void {
+      intervalCallback?.()
+    },
+    get cleared(): boolean {
+      return cleared
+    },
+  }
+}
+
+function deferred<T = void>() {
+  let resolve!: (value: T | PromiseLike<T>) => void
+  let reject!: (reason?: unknown) => void
+  const promise = new Promise<T>((resolvePromise, rejectPromise) => {
+    resolve = resolvePromise
+    reject = rejectPromise
+  })
+  return { promise, resolve, reject }
+}
+
+async function waitFor(condition: () => boolean): Promise<void> {
+  for (let i = 0; i < 50; i++) {
+    if (condition()) return
+    await new Promise(resolve => setTimeout(resolve, 0))
+  }
+  throw new Error('condition was not met')
+}
+
+async function withTempFile<T>(
+  name: string,
+  run: (path: string) => Promise<T>,
+): Promise<T> {
+  const dir = await mkdtemp(join(tmpdir(), 'openclaude-bg-test-'))
+  try {
+    return await run(join(dir, name))
+  } finally {
+    await rm(dir, { recursive: true, force: true })
+  }
+}
+
 describe('background session CLI parsing', () => {
   it('builds a print-mode child command and preserves provider/model flags', () => {
     const parsed = parseBackgroundInvocation([
@@ -254,6 +333,16 @@ describe('background session CLI parsing', () => {
       follow: false,
       stream: 'stderr',
     })
+    expect(parseLogsInvocation(['auth-refactor', '--stdout', '-f'])).toEqual({
+      target: 'auth-refactor',
+      follow: true,
+      stream: 'stdout',
+    })
+    expect(parseLogsInvocation(['auth-refactor', '-f', '--stderr'])).toEqual({
+      target: 'auth-refactor',
+      follow: true,
+      stream: 'stderr',
+    })
   })
 
   it('preserves Node exec flags and lets the launcher manage heap relaunch state', () => {
@@ -306,3 +395,292 @@ describe('background session CLI parsing', () => {
     expect(signals).toEqual(['SIGTERM', 'SIGKILL'])
   })
 })
+
+describe('background session log streaming', () => {
+  it('emits a multi-megabyte existing log exactly with bounded allocations', async () => {
+    await withTempFile('stdout.log', async path => {
+      const chunkSize = LOG_STREAM_CHUNK_SIZE
+      const contents = Buffer.alloc(chunkSize * 32 + 123)
+      for (let i = 0; i < contents.length; i++) contents[i] = i % 251
+      await writeFile(path, contents)
+
+      const allocations: number[] = []
+      const output = new TestOutput()
+      const offset = await printExistingLog(path, {
+        output,
+        chunkSize,
+        createBuffer: size => {
+          allocations.push(size)
+          return Buffer.alloc(size)
+        },
+      })
+
+      expect(offset).toBe(contents.length)
+      expect(output.bytes()).toEqual(contents)
+      expect(Math.max(...allocations)).toBeLessThanOrEqual(chunkSize)
+      expect(allocations.length).toBeGreaterThan(1)
+    })
+  })
+
+  it('follow mode emits existing and appended content exactly once in order', async () => {
+    await withTempFile('stdout.log', async path => {
+      const output = new TestOutput()
+      await writeFile(path, Buffer.from('existing-'))
+
+      const offset = await printExistingLog(path, { output, chunkSize: 4 })
+      const scheduler = createManualScheduler()
+      const abort = new AbortController()
+      const following = followLogFile(path, offset, {
+        output,
+        chunkSize: 4,
+        signal: abort.signal,
+        setInterval: scheduler.setInterval,
+        clearInterval: scheduler.clearInterval,
+      })
+
+      await writeFile(path, Buffer.from('existing-appended'))
+      scheduler.tick()
+      await waitFor(() => output.bytes().toString() === 'existing-appended')
+      abort.abort()
+      await following
+
+      expect(output.bytes().toString()).toBe('existing-appended')
+    })
+  })
+
+  it('splits a large appended range into bounded chunks', async () => {
+    await withTempFile('stdout.log', async path => {
+      const chunkSize = 16
+      await writeFile(path, Buffer.from('seed'))
+      const appended = Buffer.alloc(chunkSize * 2 + 5, 7)
+      await writeFile(path, Buffer.concat([Buffer.from('seed'), appended]))
+
+      const output = new TestOutput()
+      const scheduler = createManualScheduler()
+      const abort = new AbortController()
+      const following = followLogFile(path, 4, {
+        output,
+        chunkSize,
+        signal: abort.signal,
+        setInterval: scheduler.setInterval,
+        clearInterval: scheduler.clearInterval,
+      })
+
+      scheduler.tick()
+      await waitFor(() => output.bytes().length === appended.length)
+      abort.abort()
+      await following
+
+      expect(output.bytes()).toEqual(appended)
+      expect(output.chunks.map(chunk => chunk.length)).toEqual([16, 16, 5])
+    })
+  })
+
+  it('waits for drain before reading or writing more when stdout applies backpressure', async () => {
+    await withTempFile('stdout.log', async path => {
+      await writeFile(path, Buffer.from('abcdef'))
+      const output = new TestOutput()
+      output.writeResults.push(false)
+
+      let settled = false
+      const printing = printExistingLog(path, { output, chunkSize: 3 }).then(
+        offset => {
+          settled = true
+          return offset
+        },
+      )
+
+      await waitFor(() => output.chunks.length === 1)
+      expect(output.bytes().toString()).toBe('abc')
+      expect(settled).toBe(false)
+
+      output.emit('drain')
+      await expect(printing).resolves.toBe(6)
+      expect(output.bytes().toString()).toBe('abcdef')
+      expect(output.chunks.map(chunk => chunk.length)).toEqual([3, 3])
+    })
+  })
+
+  it('resets the follow read position when the log is truncated', async () => {
+    await withTempFile('stdout.log', async path => {
+      await writeFile(path, Buffer.from('abcdef'))
+      const output = new TestOutput()
+      const scheduler = createManualScheduler()
+      const abort = new AbortController()
+      const following = followLogFile(path, 6, {
+        output,
+        chunkSize: 8,
+        signal: abort.signal,
+        setInterval: scheduler.setInterval,
+        clearInterval: scheduler.clearInterval,
+      })
+
+      await writeFile(path, Buffer.from('xy'))
+      scheduler.tick()
+      await waitFor(() => output.bytes().toString() === 'xy')
+      abort.abort()
+      await following
+
+      expect(output.bytes().toString()).toBe('xy')
+    })
+  })
+
+  it('tolerates temporary file disappearance while following', async () => {
+    await withTempFile('stdout.log', async path => {
+      await writeFile(path, Buffer.from('seed'))
+      const output = new TestOutput()
+      const scheduler = createManualScheduler()
+      const abort = new AbortController()
+      const following = followLogFile(path, 4, {
+        output,
+        chunkSize: 8,
+        signal: abort.signal,
+        setInterval: scheduler.setInterval,
+        clearInterval: scheduler.clearInterval,
+      })
+
+      await unlink(path)
+      scheduler.tick()
+      await new Promise(resolve => setTimeout(resolve, 0))
+      expect(output.bytes().length).toBe(0)
+
+      await writeFile(path, Buffer.from('new'))
+      scheduler.tick()
+      await waitFor(() => output.bytes().toString() === 'new')
+      abort.abort()
+      await following
+
+      expect(output.bytes().toString()).toBe('new')
+    })
+  })
+
+  it('prevents writes after signal cleanup during an in-flight poll', async () => {
+    const output = new TestOutput()
+    const scheduler = createManualScheduler()
+    const abort = new AbortController()
+    const readStarted = deferred()
+    const releaseRead = deferred()
+    let closed = false
+
+    const handle = {
+      stat: async () => ({ size: 4 }),
+      read: async (buffer: Buffer) => {
+        readStarted.resolve()
+        await releaseRead.promise
+        buffer.write('late')
+        return { bytesRead: 4, buffer }
+      },
+      close: async () => {
+        closed = true
+      },
+    }
+
+    let resolved = false
+    const following = followLogFile('/tmp/stdout.log', 0, {
+      output,
+      chunkSize: 4,
+      signal: abort.signal,
+      setInterval: scheduler.setInterval,
+      clearInterval: scheduler.clearInterval,
+      openFile: async () => handle,
+    }).then(() => {
+      resolved = true
+    })
+
+    scheduler.tick()
+    await readStarted.promise
+    abort.abort()
+    await new Promise(resolve => setTimeout(resolve, 0))
+    expect(resolved).toBe(false)
+    releaseRead.resolve()
+    await following
+    scheduler.tick()
+    await new Promise(resolve => setTimeout(resolve, 0))
+
+    expect(scheduler.cleared).toBe(true)
+    expect(closed).toBe(true)
+    expect(output.bytes().length).toBe(0)
+  })
+
+  it('preserves streamed progress when a later close failure occurs', async () => {
+    const output = new TestOutput()
+    const scheduler = createManualScheduler()
+    const abort = new AbortController()
+    let openCount = 0
+
+    const following = followLogFile('/tmp/stdout.log', 0, {
+      output,
+      chunkSize: 4,
+      signal: abort.signal,
+      setInterval: scheduler.setInterval,
+      clearInterval: scheduler.clearInterval,
+      openFile: async () => {
+        openCount++
+        return {
+          stat: async () => ({ size: 4 }),
+          read: async (buffer: Buffer) => {
+            buffer.write('once')
+            return { bytesRead: 4 }
+          },
+          close: async () => {
+            if (openCount === 1) throw new Error('close failed')
+          },
+        }
+      },
+    })
+
+    scheduler.tick()
+    await waitFor(() => output.bytes().toString() === 'once')
+    scheduler.tick()
+    await waitFor(() => openCount === 2)
+    await new Promise(resolve => setTimeout(resolve, 0))
+    abort.abort()
+    await following
+
+    expect(output.bytes().toString()).toBe('once')
+  })
+
+  it('surfaces non-follow file read failures', async () => {
+    let closed = false
+    const readError = Object.assign(new Error('read failed'), {
+      code: 'EIO',
+    })
+
+    await expect(
+      printExistingLog('/tmp/stdout.log', {
+        chunkSize: 4,
+        openFile: async () => ({
+          stat: async () => ({ size: 4 }),
+          read: async () => {
+            throw readError
+          },
+          close: async () => {
+            closed = true
+          },
+        }),
+      }),
+    ).rejects.toThrow('read failed')
+    expect(closed).toBe(true)
+  })
+
+  it('handles EPIPE and destroyed stdout without throwing', async () => {
+    await withTempFile('stdout.log', async path => {
+      await writeFile(path, Buffer.from('closed-pipe'))
+
+      const epipeOutput = new TestOutput()
+      epipeOutput.writeError = Object.assign(new Error('broken pipe'), {
+        code: 'EPIPE',
+      })
+      await expect(
+        printExistingLog(path, { output: epipeOutput, chunkSize: 4 }),
+      ).resolves.toBe(0)
+
+      const destroyedOutput = new TestOutput()
+      destroyedOutput.destroyed = true
+      await expect(
+        printExistingLog(path, { output: destroyedOutput, chunkSize: 4 }),
+      ).resolves.toBe(0)
+      expect(destroyedOutput.bytes().length).toBe(0)
+    })
+  })
+})

src/cli/bg.ts

@@ -1,7 +1,7 @@
 import { spawn } from 'node:child_process'
 import { randomUUID } from 'node:crypto'
 import { closeSync, openSync } from 'node:fs'
-import { open, readFile, unlink } from 'node:fs/promises'
+import { open, unlink } from 'node:fs/promises'
 import { basename } from 'node:path'
 import treeKill from 'tree-kill'
 import { argsBeforeDelimiter } from '../utils/cliArgs.js'
@@ -58,6 +58,50 @@ const HEAP_RELAUNCHED_ENV = 'OPENCLAUDE_HEAP_RELAUNCHED'
 const DEFAULT_TERM_GRACE_MS = 2_000
 const DEFAULT_KILL_GRACE_MS = 2_000
 const DEFAULT_KILL_POLL_INTERVAL_MS = 100
+// Each background-log read buffer is capped at 64 KiB to avoid whole-log allocations.
+export const LOG_STREAM_CHUNK_SIZE = 64 * 1024
+const LOG_FOLLOW_POLL_INTERVAL_MS = 500
+
+type LogOutput = {
+  destroyed?: boolean
+  writableDestroyed?: boolean
+  write(chunk: Uint8Array): boolean
+  once(event: string, listener: (...args: unknown[]) => void): unknown
+  off(event: string, listener: (...args: unknown[]) => void): unknown
+}
+
+type LogFileHandle = {
+  close(): Promise<void>
+  stat(): Promise<{ size: number }>
+  read(
+    buffer: Buffer,
+    offset: number,
+    length: number,
+    position: number,
+  ): Promise<{ bytesRead: number }>
+}
+
+type LogFollowTimer = ReturnType<typeof setInterval> | number
+
+type StreamLogOptions = {
+  output?: LogOutput
+  chunkSize?: number
+  createBuffer?: (size: number) => Buffer
+  signal?: AbortSignal
+  openFile?: (path: string, flags: 'r') => Promise<LogFileHandle>
+  continueOnFileError?: boolean
+}
+
+type FollowLogOptions = StreamLogOptions & {
+  pollIntervalMs?: number
+  setInterval?: (callback: () => void, ms: number) => LogFollowTimer
+  clearInterval?: (timer: LogFollowTimer) => void
+}
+
+type StreamLogResult = {
+  position: number
+  outputOpen: boolean
+}
 
 // This must stay in sync with value-consuming CLI flags in main.tsx and related
 // handlers. If the CLI flag definitions become centralized, move this parser
@@ -410,56 +454,216 @@ function printSessionTable(
   }
 }
 
-async function printExistingLog(path: string): Promise<number> {
+function isOutputClosed(output: LogOutput): boolean {
+  return output.destroyed === true || output.writableDestroyed === true
+}
+
+function normalizeChunkSize(chunkSize: number | undefined): number {
+  if (!Number.isFinite(chunkSize) || !chunkSize || chunkSize < 1) {
+    return LOG_STREAM_CHUNK_SIZE
+  }
+  return Math.floor(chunkSize)
+}
+
+async function waitForDrain(
+  output: LogOutput,
+  signal: AbortSignal | undefined,
+): Promise<boolean> {
+  if (signal?.aborted || isOutputClosed(output)) return false
+
+  return await new Promise<boolean>(resolve => {
+    let settled = false
+
+    const cleanup = () => {
+      output.off('drain', onDrain)
+      output.off('error', onError)
+      output.off('close', onClose)
+      signal?.removeEventListener('abort', onAbort)
+    }
+    const finish = (open: boolean) => {
+      if (settled) return
+      settled = true
+      cleanup()
+      resolve(open)
+    }
+
+    const onDrain = () => finish(!isOutputClosed(output))
+    const onError = () => finish(false)
+    const onClose = () => finish(false)
+    const onAbort = () => finish(false)
+
+    output.once('drain', onDrain)
+    output.once('error', onError)
+    output.once('close', onClose)
+    signal?.addEventListener('abort', onAbort, { once: true })
+  })
+}
+
+async function writeLogBuffer(
+  output: LogOutput,
+  buffer: Buffer,
+  signal: AbortSignal | undefined,
+): Promise<boolean> {
+  if (buffer.length === 0) return true
+  if (signal?.aborted || isOutputClosed(output)) return false
+
   try {
-    const contents = await readFile(path)
-    if (contents.length > 0) process.stdout.write(contents)
-    return contents.length
+    if (output.write(buffer)) return !isOutputClosed(output)
   } catch {
-    return 0
+    return false
+  }
+
+  return await waitForDrain(output, signal)
+}
+
+async function streamLogRange(
+  handle: LogFileHandle,
+  start: number,
+  endExclusive: number,
+  options: StreamLogOptions,
+): Promise<StreamLogResult> {
+  const output = options.output ?? process.stdout
+  const chunkSize = normalizeChunkSize(options.chunkSize)
+  const createBuffer = options.createBuffer ?? Buffer.allocUnsafe
+  let position = start
+
+  while (position < endExclusive) {
+    if (options.signal?.aborted) return { position, outputOpen: false }
+    const bytesToRead = Math.min(chunkSize, endExclusive - position)
+    const buffer = createBuffer(bytesToRead)
+    if (buffer.length < bytesToRead) {
+      throw new Error('Log stream buffer factory returned a short buffer')
+    }
+
+    let bytesRead: number
+    try {
+      const readResult = await handle.read(buffer, 0, bytesToRead, position)
+      bytesRead = readResult.bytesRead
+    } catch (error) {
+      if (!options.continueOnFileError) throw error
+      return { position, outputOpen: true }
+    }
+    if (bytesRead <= 0) break
+    if (options.signal?.aborted) return { position, outputOpen: false }
+
+    const chunk =
+      bytesRead === buffer.length ? buffer : buffer.subarray(0, bytesRead)
+    if (!(await writeLogBuffer(output, chunk, options.signal))) {
+      return { position, outputOpen: false }
+    }
+    position += bytesRead
+  }
+
+  return { position, outputOpen: true }
+}
+
+async function streamLogSnapshot(
+  path: string,
+  offset: number,
+  options: StreamLogOptions,
+): Promise<StreamLogResult> {
+  let handle: LogFileHandle
+  try {
+    handle = await (options.openFile ?? open)(path, 'r')
+  } catch (error) {
+    if (!options.continueOnFileError) throw error
+    // Keep following; the child may create or rotate the file later.
+    return { position: offset, outputOpen: true }
+  }
+
+  let result: StreamLogResult = { position: offset, outputOpen: true }
+  try {
+    const { size } = await handle.stat()
+    const start = size < offset ? 0 : offset
+    result =
+      size <= start
+        ? { position: start, outputOpen: true }
+        : await streamLogRange(handle, start, size, options)
+    return result
+  } catch (error) {
+    if (!options.continueOnFileError) throw error
+    return result
+  } finally {
+    if (options.continueOnFileError) {
+      await handle.close().catch(() => undefined)
+    } else {
+      await handle.close()
+    }
   }
 }
 
-async function followLogFile(path: string, offset: number): Promise<void> {
+export async function printExistingLog(
+  path: string,
+  options: StreamLogOptions = {},
+): Promise<number> {
+  const result = await streamLogSnapshot(path, 0, options)
+  return result.position
+}
+
+export async function followLogFile(
+  path: string,
+  offset: number,
+  options: FollowLogOptions = {},
+): Promise<void> {
+  const output = options.output ?? process.stdout
+  const cleanupController = new AbortController()
   let position = offset
   let reading = false
+  let stopped = false
+  let timer: LogFollowTimer | undefined
+  let activePoll: Promise<void> | undefined
 
   await new Promise<void>(resolve => {
     const cleanup = () => {
-      clearInterval(timer)
+      if (stopped) return
+      stopped = true
+      if (timer) (options.clearInterval ?? clearInterval)(timer)
+      cleanupController.abort()
       process.off('SIGINT', cleanup)
       process.off('SIGTERM', cleanup)
-      resolve()
+      options.signal?.removeEventListener('abort', cleanup)
+      const pendingPoll = activePoll
+      if (pendingPoll) {
+        void pendingPoll.finally(resolve)
+      } else {
+        resolve()
+      }
     }
 
-    const timer = setInterval(() => {
-      if (reading) return
+    const poll = () => {
+      if (stopped || reading) return
       reading = true
-      void (async () => {
+      const pollPromise = (async () => {
         try {
-          const handle = await open(path, 'r')
-          try {
-            const { size } = await handle.stat()
-            if (size < position) position = 0
-            if (size > position) {
-              const buffer = Buffer.alloc(size - position)
-              await handle.read(buffer, 0, buffer.length, position)
-              position = size
-              process.stdout.write(buffer)
-            }
-          } finally {
-            await handle.close()
-          }
-        } catch {
-          // Keep following; the child may create or rotate the file later.
+          const result = await streamLogSnapshot(path, position, {
+            ...options,
+            output,
+            signal: cleanupController.signal,
+            continueOnFileError: true,
+          })
+          position = result.position
+          if (!result.outputOpen) cleanup()
         } finally {
           reading = false
         }
       })()
-    }, 500)
+      activePoll = pollPromise
+      void pollPromise.finally(() => {
+        if (activePoll === pollPromise) activePoll = undefined
+      })
+    }
 
     process.once('SIGINT', cleanup)
     process.once('SIGTERM', cleanup)
+    if (options.signal?.aborted) {
+      cleanup()
+      return
+    }
+    options.signal?.addEventListener('abort', cleanup, { once: true })
+    timer = (options.setInterval ?? setInterval)(
+      poll,
+      options.pollIntervalMs ?? LOG_FOLLOW_POLL_INTERVAL_MS,
+    )
   })
 }
 
@@ -571,7 +775,14 @@ export async function logsHandler(
     fail(`Log file does not exist: ${logPath}`)
   }
 
-  const offset = await printExistingLog(logPath)
+  let offset: number
+  try {
+    offset = await printExistingLog(logPath, {
+      continueOnFileError: parsed.follow,
+    })
+  } catch (error) {
+    fail(`Failed to read log file: ${errorMessage(error)}`)
+  }
   if (parsed.follow) {
     await followLogFile(logPath, offset)
   }