diff --git a/.changeset/cap-foreground-command-output.md b/.changeset/cap-foreground-command-output.md new file mode 100644 index 0000000000..42e8e40cec --- /dev/null +++ b/.changeset/cap-foreground-command-output.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/agent-core": patch +--- + +Cap the output a single foreground shell command may stream so a runaway command can no longer crash the process. A command that produces a very large or unbounded amount of output (e.g. `b3sum --length 18446744073709551615`) previously grew the live-output buffer until Node aborted with a JavaScript heap out-of-memory error; it is now gracefully terminated once its output exceeds 16 MiB, and the result explains how to redirect large output to a file instead. The per-task output ring buffer is also maintained in O(1) per chunk rather than O(n²). Background tasks are unaffected. diff --git a/packages/agent-core/src/agent/background/index.ts b/packages/agent-core/src/agent/background/index.ts index 04b5837d35..5ef3fac694 100644 --- a/packages/agent-core/src/agent/background/index.ts +++ b/packages/agent-core/src/agent/background/index.ts @@ -58,8 +58,20 @@ interface ManagedTask { readonly taskId: string; readonly task: BackgroundTask; readonly outputChunks: string[]; + /** + * Running total of characters currently held in `outputChunks`, maintained + * incrementally so the ring-buffer cap stays O(1) per chunk instead of + * re-summing every chunk (which was O(n²) over a command's lifetime). + */ + outputRingChars: number; /** Total UTF-8 bytes observed, including chunks dropped from the live ring buffer. */ outputSizeBytes: number; + /** + * True once a foreground command has crossed `MAX_FOREGROUND_OUTPUT_BYTES` + * and termination has been requested. One-shot guard so the ceiling fires + * exactly once. + */ + outputLimitTripped: boolean; status: BackgroundTaskStatus; /** Normalized registration options. Current mutable state stays on ManagedTask. */ readonly options: RegisterBackgroundTaskOptions; @@ -110,6 +122,28 @@ interface ManagedTask { const MAX_OUTPUT_BYTES = 1024 * 1024; // 1 MiB const NOTIFICATION_FALLBACK_PREVIEW_BYTES = 3_000; +/** + * Hard ceiling on the combined output a single *foreground* command may stream + * before it is force-terminated (SIGTERM → grace → SIGKILL). It guards the + * live-forward path — which has no memory bound of its own — from a runaway + * command (e.g. `b3sum --length `) whose output would otherwise grow the + * process heap until Node aborts with an out-of-memory crash. + * + * Detached (background) tasks are exempt: their output is ring-buffered and + * spilled to disk, so it never accumulates unbounded in memory. + */ +const MAX_FOREGROUND_OUTPUT_BYTES = 16 * 1024 * 1024; // 16 MiB + +/** Terminal `stopReason` recorded when a foreground command trips the output ceiling. */ +function foregroundOutputLimitReason(): string { + const mib = Math.floor(MAX_FOREGROUND_OUTPUT_BYTES / (1024 * 1024)); + return ( + `Output limit exceeded: the command produced more than ${mib} MiB and was ` + + 'terminated. Redirect large output to a file (e.g. `command > out.txt`) and ' + + 'inspect it in slices instead.' + ); +} + const SIGTERM_GRACE_MS = 5_000; const USER_INTERRUPT_REASON = 'Interrupted by user'; @@ -281,7 +315,9 @@ export class BackgroundManager { taskId, task, outputChunks: [], + outputRingChars: 0, outputSizeBytes: 0, + outputLimitTripped: false, status: 'running', options: entryOptions, startedAt: Date.now(), @@ -594,14 +630,39 @@ export class BackgroundManager { private appendOutput(entry: ManagedTask, chunk: string): void { entry.outputSizeBytes += Buffer.byteLength(chunk, 'utf-8'); entry.outputChunks.push(chunk); - // Enforce output cap: drop oldest chunks when over budget. - let total = entry.outputChunks.reduce((s, c) => s + c.length, 0); - while (total > MAX_OUTPUT_BYTES && entry.outputChunks.length > 1) { + entry.outputRingChars += chunk.length; + // Enforce the ring-buffer cap: drop oldest chunks when over budget. The + // running total keeps this O(1) amortized per chunk; re-summing the whole + // buffer on every chunk was O(n²) over a command's lifetime and could + // starve the event loop (and so the foreground timeout) under a high-rate + // output stream. + while (entry.outputRingChars > MAX_OUTPUT_BYTES && entry.outputChunks.length > 1) { const removed = entry.outputChunks.shift(); if (removed === undefined) break; - total -= removed.length; + entry.outputRingChars -= removed.length; } + // Foreground output ceiling: a single non-detached command must not grow + // the (unbounded) live-forward buffer until the process runs out of memory. + // Trip once, then request graceful termination through the shared stop path + // (SIGTERM → grace → SIGKILL). Detached tasks are exempt — their output is + // ring-buffered and spilled to disk, so it never accumulates in memory. + if ( + !entry.outputLimitTripped && + !this.isDetached(entry) && + entry.outputSizeBytes > MAX_FOREGROUND_OUTPUT_BYTES + ) { + entry.outputLimitTripped = true; + void this.stop(entry.taskId, foregroundOutputLimitReason()); + } + + // Once the cap has tripped the task is being terminated: keep only the + // bounded in-memory ring buffer above and stop feeding the (unbounded) disk + // write chain. A producer that ignores SIGTERM could otherwise keep the + // chain — and the chunk strings each pending write retains — growing through + // the grace window until SIGKILL, re-introducing the OOM this cap prevents. + if (entry.outputLimitTripped) return; + if (this.persistence === undefined) return; // Foreground tasks keep their full output in memory and only touch disk diff --git a/packages/agent-core/src/agent/background/process-task.ts b/packages/agent-core/src/agent/background/process-task.ts index 3fd85ea7f9..22e96a744a 100644 --- a/packages/agent-core/src/agent/background/process-task.ts +++ b/packages/agent-core/src/agent/background/process-task.ts @@ -139,6 +139,12 @@ function observeProcessStream( const onData = (chunk: string): void => { if (chunk.length === 0) return; sink.appendOutput(chunk); + // Once the manager has begun terminating the task — an output-limit trip + // (see MAX_FOREGROUND_OUTPUT_BYTES), a user interrupt, or a timeout — + // `appendOutput` above may synchronously abort the signal. Stop forwarding + // live output from that point so the unbounded forward buffer cannot keep + // growing while the process is being killed. + if (sink.signal.aborted) return; onOutput?.(kind, chunk); }; stream.on('data', onData); diff --git a/packages/agent-core/test/agent/background/output-limit.test.ts b/packages/agent-core/test/agent/background/output-limit.test.ts new file mode 100644 index 0000000000..70fcd3fac2 --- /dev/null +++ b/packages/agent-core/test/agent/background/output-limit.test.ts @@ -0,0 +1,172 @@ +/** + * Foreground output ceiling. + * + * A single non-detached command that streams more output than the per-command + * limit must be force-terminated instead of growing the (unbounded) + * live-forward buffer until the process runs out of memory. Detached + * (background) tasks are exempt: their output is ring-buffered and spilled to + * disk, so it never accumulates in memory. + */ + +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { Readable, type Writable } from 'node:stream'; + +import type { KaosProcess } from '@moonshot-ai/kaos'; +import { join } from 'pathe'; +import { describe, expect, it, vi } from 'vitest'; + +import { ProcessBackgroundTask } from '../../../src/agent/background'; +import { createBackgroundManager, waitForTerminal } from './helpers'; + +const MiB = 1024 * 1024; +const LIMIT_BYTES = 16 * MiB; + +/** + * A process that streams `chunks` of stdout, then exits 0 on its own — unless + * it is killed first, in which case `wait()` resolves with the signal's exit + * code and the stream is destroyed (simulating the child dying on SIGTERM). + */ +function streamingProcess(chunks: string[]): { + proc: KaosProcess; + kill: ReturnType; +} { + const stdout = Readable.from(chunks); + const stderr = Readable.from([]); + let resolveWait!: (code: number) => void; + const waitP = new Promise((resolve) => { + resolveWait = resolve; + }); + stdout.on('end', () => { + resolveWait(0); + }); + const kill = vi.fn(async (signal: string) => { + stdout.destroy(); + resolveWait(signal === 'SIGKILL' ? 137 : 143); + }); + const proc = { + stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable, + stdout, + stderr, + pid: 4242, + exitCode: null, + wait: () => waitP, + kill, + dispose: vi.fn().mockResolvedValue(undefined), + } as unknown as KaosProcess; + return { proc, kill }; +} + +/** + * A process that keeps streaming all of `chunks` regardless of SIGTERM (only + * SIGKILL stops it) — simulating a producer that ignores the graceful stop and + * keeps writing through the SIGTERM grace window. + */ +function sigtermIgnoringProcess(chunks: string[]): { proc: KaosProcess; kill: ReturnType } { + const stdout = Readable.from(chunks); + const stderr = Readable.from([]); + let resolveWait!: (code: number) => void; + const waitP = new Promise((resolve) => { + resolveWait = resolve; + }); + stdout.on('end', () => { + resolveWait(0); + }); + const kill = vi.fn(async (signal: string) => { + if (signal === 'SIGKILL') { + stdout.destroy(); + resolveWait(137); + } + // SIGTERM is intentionally ignored. + }); + const proc = { + stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable, + stdout, + stderr, + pid: 4243, + exitCode: null, + wait: () => waitP, + kill, + dispose: vi.fn().mockResolvedValue(undefined), + } as unknown as KaosProcess; + return { proc, kill }; +} + +describe('BackgroundManager — foreground output ceiling', () => { + it('terminates a foreground command that exceeds the output limit and stops forwarding', async () => { + const { manager } = createBackgroundManager(); + // 20 MiB total, well past the 16 MiB ceiling. + const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB)); + const { proc, kill } = streamingProcess(chunks); + + let forwardedChars = 0; + const onOutput = vi.fn((_kind: 'stdout' | 'stderr', text: string) => { + forwardedChars += text.length; + }); + + const taskId = manager.registerTask( + new ProcessBackgroundTask(proc, 'b3sum --length 18446744073709551615', 'hash', onOutput), + { detached: false, signal: new AbortController().signal, timeoutMs: 60_000 }, + ); + + const info = await waitForTerminal(manager, taskId); + + expect(info?.status).toBe('killed'); + expect(info?.stopReason ?? '').toMatch(/output limit/i); + expect(kill).toHaveBeenCalledWith('SIGTERM'); + // The live-forward path is capped at the ceiling rather than draining the + // full 20 MiB into the (unbounded) transcript/stderr buffer. + expect(forwardedChars).toBeLessThanOrEqual(LIMIT_BYTES); + }); + + it('does not terminate a detached (background) task for the same output', async () => { + const { manager } = createBackgroundManager(); + const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB)); + const { proc, kill } = streamingProcess(chunks); + + const taskId = manager.registerTask(new ProcessBackgroundTask(proc, 'producer', 'bg'), { + detached: true, + timeoutMs: 60_000, + }); + + const info = await waitForTerminal(manager, taskId); + + expect(info?.status).toBe('completed'); + expect(kill).not.toHaveBeenCalledWith('SIGTERM'); + }); + + it('stops enqueuing output to disk once the foreground cap trips', async () => { + const sessionDir = mkdtempSync(join(tmpdir(), 'bpm-limit-')); + try { + const { manager, persistence } = createBackgroundManager({ sessionDir }); + let persistedChars = 0; + const spy = vi + .spyOn(persistence!, 'appendTaskOutput') + .mockImplementation(async (_id: string, chunk: string) => { + persistedChars += chunk.length; + }); + + // 20 MiB, and the producer ignores SIGTERM so it keeps writing through + // the whole grace window. + const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB)); + const { proc } = sigtermIgnoringProcess(chunks); + + const taskId = manager.registerTask( + new ProcessBackgroundTask(proc, 'runaway', 'hash', () => {}), + { detached: false, signal: new AbortController().signal, timeoutMs: 60_000 }, + ); + + const info = await waitForTerminal(manager, taskId); + + expect(info?.status).toBe('killed'); + // Before the fix every chunk of the 20 MiB is enqueued into the disk + // write chain (retaining each string until its write drains); afterwards + // enqueuing stops at the ceiling so the chain cannot grow unbounded. + expect(persistedChars).toBeLessThanOrEqual(17 * MiB); + + spy.mockRestore(); + } finally { + rmSync(sessionDir, { recursive: true, force: true }); + } + }); +});