fix(runner): bound the per-agent PTY output buffer

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<string, number> })
+    .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<string, number> })
+      .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');
+  });
+});

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<string, MountHandle>();
 
-  // 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<string, string[]>();
+  /** 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<string, number>();
   /** Snapshot of PTY output from the most recent failed attempt, keyed by step name. */
   private readonly lastFailedStepOutput = new Map<string, string>();
   /** 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 <name>` 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