diff --git a/packages/core/src/__tests__/pty-buffer-bounded.test.ts b/packages/core/src/__tests__/pty-buffer-bounded.test.ts new file mode 100644 index 0000000..71d3ca5 --- /dev/null +++ b/packages/core/src/__tests__/pty-buffer-bounded.test.ts @@ -0,0 +1,149 @@ +/** + * The per-agent PTY buffer used to grow without limit for the whole life of an + * agent step, so a chatty agent's entire transcript stayed resident in the + * orchestrator's heap. A 34-step DAG with 9 agents OOM-killed its orchestrator + * in an 8 GB sandbox partway through (AgentWorkforce/cloud#1967, dev run + * 71cc4995 — `bun exited with signal SIGKILL` while the parent node process + * survived, which is the cgroup OOM-killer signature). + * + * Retaining a tail is safe: consumers clip to the last few thousand characters, + * and the complete transcript is still written to the PTY log file on disk. + */ +import { readFileSync } from 'node:fs'; + +import { describe, expect, it } from 'vitest'; + +import { WorkflowRunner } from '../runner.js'; + +type BoundedAppend = ( + agentName: string, + buffer: string[] | undefined, + chunk: string, +) => void; + +/** Reach the private helper without widening the public surface. */ +function boundedAppend(runner: WorkflowRunner): BoundedAppend { + const fn = (runner as unknown as { appendBoundedPtyChunk: BoundedAppend }) + .appendBoundedPtyChunk; + return fn.bind(runner) as BoundedAppend; +} + +function maxChars(): number { + return (WorkflowRunner as unknown as { MAX_PTY_BUFFER_CHARS: number }) + .MAX_PTY_BUFFER_CHARS; +} + +function newRunner(): WorkflowRunner { + return Object.create(WorkflowRunner.prototype) as WorkflowRunner; +} + +function withSizes(runner: WorkflowRunner): WorkflowRunner { + (runner as unknown as { ptyOutputBufferSizes: Map }) + .ptyOutputBufferSizes = new Map(); + return runner; +} + +describe('PTY output buffer is bounded', () => { + it('keeps a chatty agent from growing the buffer without limit', () => { + const runner = withSizes(newRunner()); + const append = boundedAppend(runner); + const buffer: string[] = []; + + // 40 MB of output through a 1 MB cap. + const chunk = 'x'.repeat(100_000); + for (let i = 0; i < 400; i += 1) append('scout', buffer, chunk); + + const retained = buffer.join('').length; + expect(retained).toBeLessThanOrEqual(maxChars()); + // Sanity: the old behaviour would have retained everything. + expect(retained).toBeLessThan(40_000_000); + }); + + it('retains the most recent output, not the oldest', () => { + const runner = withSizes(newRunner()); + const append = boundedAppend(runner); + const buffer: string[] = []; + + append('scout', buffer, 'FIRST'); + for (let i = 0; i < 30; i += 1) append('scout', buffer, 'y'.repeat(100_000)); + append('scout', buffer, 'LAST'); + + const retained = buffer.join(''); + // The tail is what every consumer reads, so it must survive. + expect(retained.endsWith('LAST')).toBe(true); + expect(retained).not.toContain('FIRST'); + }); + + it('leaves small transcripts completely untouched', () => { + const runner = withSizes(newRunner()); + const append = boundedAppend(runner); + const buffer: string[] = []; + + append('scout', buffer, 'hello '); + append('scout', buffer, 'world'); + + expect(buffer.join('')).toBe('hello world'); + }); + + it('keeps a single oversized write rather than dropping it entirely', () => { + const runner = withSizes(newRunner()); + const append = boundedAppend(runner); + const buffer: string[] = []; + + const huge = 'z'.repeat(maxChars() * 2); + append('scout', buffer, huge); + + // One chunk bigger than the cap is still visible — silently discarding it + // would lose the very output someone is debugging. + expect(buffer).toHaveLength(1); + expect(buffer[0]).toBe(huge); + }); + + it('tracks size per agent independently', () => { + const runner = withSizes(newRunner()); + const append = boundedAppend(runner); + const scout = ['seed-scout']; + const lead = ['seed-lead']; + + append('scout', scout, 'a'.repeat(10)); + append('lead', lead, 'b'.repeat(20)); + + const sizes = (runner as unknown as { ptyOutputBufferSizes: Map }) + .ptyOutputBufferSizes; + expect(sizes.get('scout')).toBe(10); + expect(sizes.get('lead')).toBe(20); + }); + + it('is a no-op when the buffer is missing', () => { + const runner = withSizes(newRunner()); + const append = boundedAppend(runner); + expect(() => append('gone', undefined, 'anything')).not.toThrow(); + }); +}); + +/** + * Regression cover for the rekey path (cubic P1 on relayflows#49). + * + * When the broker assigns an agent a different name than requested, + * `rekeyPtyStreams` installs a replacement listener under BOTH names and never + * swaps it out — so it handles every remaining chunk for the rest of that + * agent's life. The first version of this fix bounded the original listener but + * left the rekeyed one doing a bare `buffer.push(...)`, which meant any rekeyed + * agent stayed unbounded: the exact OOM the cap exists to prevent. + */ +describe('rekeyed PTY listener stays bounded', () => { + it('routes rekeyed chunks through the bounded append', () => { + const source = readFileSync( + new URL('../runner.ts', import.meta.url), + 'utf8', + ); + const rekeyBody = source.slice( + source.indexOf('const rekeyedListener = (chunk: string) => {'), + ); + const listener = rekeyBody.slice(0, rekeyBody.indexOf('};')); + + // The bare push is what made rekeyed agents unbounded. + expect(listener).not.toMatch(/buffer\.push\(stripped\)/); + expect(listener).toContain('appendBoundedPtyChunk'); + }); +}); diff --git a/packages/core/src/runner.ts b/packages/core/src/runner.ts index 80092d9..510d701 100644 --- a/packages/core/src/runner.ts +++ b/packages/core/src/runner.ts @@ -946,8 +946,28 @@ export class WorkflowRunner { /** Per-agent relayfile mounts keyed by logical agent definition name. */ private readonly agentMounts = new Map(); - // PTY-based output capture: accumulate terminal output per-agent + // PTY-based output capture: accumulate terminal output per-agent. + // + // Bounded on purpose. This buffer used to grow without limit for the whole + // life of an agent step, which put a chatty agent's entire transcript in the + // orchestrator's heap. A 34-step DAG with 9 agents OOM-killed its + // orchestrator in a 8 GB sandbox partway through (AgentWorkforce/cloud#1967, + // dev run 71cc4995: `bun exited with signal SIGKILL`, parent survived — the + // cgroup OOM-killer signature). + // + // Keeping a tail rather than everything is safe because every consumer either + // clips to the last few thousand characters for diagnostics, or is reading + // failure context where the end is what matters. The complete, unclipped + // transcript is still written to the per-agent PTY log file, so nothing is + // actually lost. private readonly ptyOutputBuffers = new Map(); + /** Retained tail per agent, in characters. Generous enough that ordinary + * steps are unaffected; small enough that a runaway agent cannot exhaust + * the orchestrator. */ + private static readonly MAX_PTY_BUFFER_CHARS = 1_000_000; + /** Live character count per agent buffer, so trimming does not have to + * re-measure the whole buffer on every chunk. */ + private readonly ptyOutputBufferSizes = new Map(); /** Snapshot of PTY output from the most recent failed attempt, keyed by step name. */ private readonly lastFailedStepOutput = new Map(); /** Most recent custom verification failure details, keyed by step name. */ @@ -4451,6 +4471,7 @@ export class WorkflowRunner { for (const stream of this.ptyLogStreams.values()) stream.end(); this.ptyLogStreams.clear(); this.ptyOutputBuffers.clear(); + this.ptyOutputBufferSizes.clear(); this.ptyListeners.clear(); this.unsubBrokerStderr?.(); @@ -7989,6 +8010,7 @@ export class WorkflowRunner { // Register PTY output listener before spawning so we capture everything this.readyRuntimeAgents.delete(agentName); this.ptyOutputBuffers.set(agentName, []); + this.ptyOutputBufferSizes.set(agentName, 0); // Open a log file so `agents:logs ` works for workflow-spawned agents const logsDir = this.getWorkerLogsDir(); @@ -7998,7 +8020,7 @@ export class WorkflowRunner { this.ptyListeners.set(agentName, (chunk: string) => { const stripped = WorkflowRunner.stripAnsi(chunk); const buffer = this.ptyOutputBuffers.get(agentName); - buffer?.push(stripped); + this.appendBoundedPtyChunk(agentName, buffer, stripped); // Write raw output (with ANSI codes) to log file so dashboard's // XTermLogViewer can render colors/formatting natively via xterm.js logStream.write(chunk); @@ -8259,6 +8281,7 @@ export class WorkflowRunner { stopHeartbeat?.(); this.activeAgentHandles.delete(agentName); this.ptyOutputBuffers.delete(agentName); + this.ptyOutputBufferSizes.delete(agentName); this.ptyListeners.delete(agentName); const stream = this.ptyLogStreams.get(agentName); if (stream) { @@ -8881,6 +8904,8 @@ export class WorkflowRunner { const buffer = this.ptyOutputBuffers.get(oldName) ?? []; this.ptyOutputBuffers.set(newName, buffer); this.ptyOutputBuffers.delete(oldName); + this.ptyOutputBufferSizes.set(newName, this.ptyOutputBufferSizes.get(oldName) ?? 0); + this.ptyOutputBufferSizes.delete(oldName); const oldLogPath = path.join(logsDir, `${oldName}.log`); const newLogPath = path.join(logsDir, `${newName}.log`); @@ -8899,7 +8924,14 @@ export class WorkflowRunner { if (this.ptyListeners.has(oldName)) { const rekeyedListener = (chunk: string) => { const stripped = WorkflowRunner.stripAnsi(chunk); - buffer.push(stripped); + // Must go through the bounded append, not a bare push. This listener is + // registered under BOTH names and never replaced, so after a rekey it + // handles every remaining chunk for the rest of the agent's life. A + // direct push here would leave rekeyed agents unbounded — the exact OOM + // this cap exists to prevent — and would drift ptyOutputBufferSizes + // from the real buffer contents. Keyed on newName, where the buffer now + // lives. + this.appendBoundedPtyChunk(newName, buffer, stripped); writeToLog(chunk); if (this.isSlackHumanAssistanceEnabled(humanAssistanceConfig)) { this.observeHumanAssistanceOutput({ @@ -11546,6 +11578,32 @@ export class WorkflowRunner { return stripAnsiFn(text); } + /** + * Append a PTY chunk, dropping oldest chunks once the retained tail exceeds + * {@link WorkflowRunner.MAX_PTY_BUFFER_CHARS}. + * + * Previously this was a bare `buffer.push(...)`, so a long-running or chatty + * agent kept its entire transcript resident in the orchestrator for the whole + * step. Consumers only ever read a tail, and the full transcript is on disk + * in the PTY log, so the head is the safe thing to discard. + */ + private appendBoundedPtyChunk( + agentName: string, + buffer: string[] | undefined, + chunk: string, + ): void { + if (!buffer) return; + buffer.push(chunk); + let size = (this.ptyOutputBufferSizes.get(agentName) ?? 0) + chunk.length; + // Drop whole chunks from the front until back under the cap. Keep at least + // the most recent chunk even if it alone exceeds the cap, so a single huge + // write is still visible rather than silently vanishing. + while (size > WorkflowRunner.MAX_PTY_BUFFER_CHARS && buffer.length > 1) { + size -= buffer.shift()?.length ?? 0; + } + this.ptyOutputBufferSizes.set(agentName, size); + } + /** * Strip TUI chrome from PTY-captured output before posting to a channel. * Removes: ANSI codes, unicode spinner/thinking characters, cursor-movement