From 3d5647890e1bfcd48643c0dfbd0ba0ab3021fd67 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Wed, 2 Sep 2026 14:14:17 +0200 Subject: [PATCH 1/2] fix(runner): bound the per-agent PTY output buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-agent PTY buffer grew without limit for the whole life of an agent step, so a chatty agent's entire transcript stayed resident in the orchestrator's heap. AgentWorkforce/cloud#1967, dev run 71cc4995: a 34-step DAG with 9 agents OOM-killed its orchestrator partway through — `bun exited with signal SIGKILL` about nine minutes in, while the parent node process survived and kept logging. That asymmetry is the cgroup OOM-killer signature (largest RSS dies, siblings live) rather than a sandbox stop or reclaim. This is not a tuning problem. Daytona caps sandboxes at 8GB and the run was already at the ceiling, so the only remaining lever is holding less. Retaining a tail is safe: - every consumer already clips to the last 2000-4000 characters - the complete, unclipped transcript is still written to the per-agent PTY log file on disk, so nothing is lost Also bounds a quadratic: when Slack human-assistance is enabled the listener calls buffer.join('') on EVERY chunk, which re-joins the whole transcript each time. Capping the buffer caps that too. The size counter is tracked alongside the buffer rather than recomputed, and is kept in sync at all four lifecycle points — clear, initial set, cleanup delete, and the agent rekey path where the broker renames an agent and the buffer moves between keys. Session-Id: 4991ccea-310c-48a6-a7b1-ba71cee738a6 --- .../src/__tests__/pty-buffer-bounded.test.ts | 120 ++++++++++++++++++ packages/core/src/runner.ts | 55 +++++++- 2 files changed, 173 insertions(+), 2 deletions(-) create mode 100644 packages/core/src/__tests__/pty-buffer-bounded.test.ts 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..f6e53f3 --- /dev/null +++ b/packages/core/src/__tests__/pty-buffer-bounded.test.ts @@ -0,0 +1,120 @@ +/** + * 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 { 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(); + }); +}); diff --git a/packages/core/src/runner.ts b/packages/core/src/runner.ts index 80092d9..02b5d21 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`); @@ -11546,6 +11571,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 From d33eb20df8afe06972c0b46520ab0af3347ccde1 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Wed, 2 Sep 2026 14:32:25 +0200 Subject: [PATCH 2/2] fix(runner): bound the rekeyed PTY listener too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses cubic P1 on #49, which was correct. rekeyPtyStreams installs a replacement listener when the broker assigns an agent a different name than requested, and registers it under BOTH names without ever swapping it out. It therefore handles every remaining chunk for the rest of that agent's life — and it was still doing a bare `buffer.push(stripped)`. So the previous commit bounded only the pre-rekey listener. Any rekeyed agent stayed unbounded for its whole life, which is precisely the OOM this change exists to prevent, and ptyOutputBufferSizes drifted from the real buffer contents because those pushes never updated it. The cap looked present and did nothing for the agents most likely to be long-lived. Route it through appendBoundedPtyChunk, keyed on newName where the buffer now lives. This also bounds the `buffer.join('')` that the rekeyed listener runs on every chunk under Slack human-assistance, same quadratic as the original path. Adds a regression test asserting the rekeyed listener contains no bare push. Session-Id: 4991ccea-310c-48a6-a7b1-ba71cee738a6 --- .../src/__tests__/pty-buffer-bounded.test.ts | 29 +++++++++++++++++++ packages/core/src/runner.ts | 9 +++++- 2 files changed, 37 insertions(+), 1 deletion(-) diff --git a/packages/core/src/__tests__/pty-buffer-bounded.test.ts b/packages/core/src/__tests__/pty-buffer-bounded.test.ts index f6e53f3..71d3ca5 100644 --- a/packages/core/src/__tests__/pty-buffer-bounded.test.ts +++ b/packages/core/src/__tests__/pty-buffer-bounded.test.ts @@ -9,6 +9,8 @@ * 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'; @@ -118,3 +120,30 @@ describe('PTY output buffer is bounded', () => { 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 02b5d21..510d701 100644 --- a/packages/core/src/runner.ts +++ b/packages/core/src/runner.ts @@ -8924,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({