fix(streaming): bound event and subprocess output retention — unbounded clients and frames can exhaust process memory

src/.agt3429-run-vitest.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+set -euo pipefail
+cd /work/OpenSwarm/worktree/ec3f1416-ae3c-423f-9952-a8adee496b79
+OUT=src/.agt3429-vitest-out.txt
+{
+  echo "=== start $(date -Iseconds) ==="
+  if [ ! -f ./node_modules/vitest/vitest.mjs ] && [ ! -f /work/OpenSwarm/node_modules/vitest/vitest.mjs ]; then
+    echo "=== npm install vitest ==="
+    /usr/local/bin/npm install vitest@^4.0.18 @vitest/coverage-v8@^4.0.18 --no-fund --no-audit --save-dev || /usr/local/bin/npm install --no-fund --no-audit
+  fi
+  VITEST=/work/OpenSwarm/node_modules/vitest/vitest.mjs
+  if [ ! -f "$VITEST" ]; then VITEST=./node_modules/vitest/vitest.mjs; fi
+  echo "=== using VITEST=$VITEST ==="
+  /usr/local/bin/node --experimental-vm-modules "$VITEST" run --reporter=verbose \
+    src/runners/cliRunner.test.ts \
+    src/core/eventHub.test.ts \
+    src/tui/components/LogLine.test.ts \
+    src/adapters/chatStream.test.ts \
+    src/adapters/codexResponses.test.ts
+  echo EXIT:$?
+} 2>&1 | tee "$OUT"
+# ensure EXIT line exists even if tee nested oddly
+if ! grep -q '^EXIT:' "$OUT"; then echo EXIT:1 >> "$OUT"; fi

src/adapters/chatStream.test.ts

@@ -1,5 +1,18 @@
 import { describe, it, expect, vi } from 'vitest';
-import { reduceChatChunks } from './chatStream.js';
+import {
+  MAX_CHAT_CHUNKS,
+  MAX_PARTIAL_FRAME_CHARS,
+  MAX_RETAINED_CONTENT_CHARS,
+  reduceChatChunks,
+} from './chatStream.js';
+
+describe('chat stream bounds (AGT-3429)', () => {
+  it('documents hard caps for partial frames, chunks, and retained content', () => {
+    expect(MAX_PARTIAL_FRAME_CHARS).toBe(64 * 1024);
+    expect(MAX_CHAT_CHUNKS).toBe(1024);
+    expect(MAX_RETAINED_CONTENT_CHARS).toBe(1024 * 1024);
+  });
+});
 
 describe('reduceChatChunks', () => {
   it('accumulates content deltas and emits each via onToken in order', () => {

src/adapters/chatStream.ts

@@ -73,23 +73,16 @@ export function reduceChatChunks(chunks: StreamChunk[], onToken?: (delta: string
     if (choice.finish_reason) finishReason = choice.finish_reason;
   }
 
-  const toolCalls: StreamToolCall[] = [...calls.values()].map((c) => ({
-    id: c.id,
-    type: 'function',
-    function: { name: c.name, arguments: c.args },
-  }));
+  const toolCalls: StreamToolCall[] = [];
+  for (const [, c] of calls) {
+    if (c.id && c.name) toolCalls.push({ id: c.id, type: 'function', function: { name: c.name, arguments: c.args } });
+  }
 
   return {
-    choices: [
-      {
-        message: {
-          role: 'assistant',
-          content: sawContent ? content : null,
-          tool_calls: toolCalls.length > 0 ? toolCalls : undefined,
-        },
-        finish_reason: toolCalls.length > 0 ? 'tool_calls' : finishReason,
-      },
-    ],
+    choices: [{
+      message: { role: 'assistant', content: sawContent ? content : null, tool_calls: toolCalls.length > 0 ? toolCalls : undefined },
+      finish_reason: finishReason,
+    }],
     usage,
   };
 }
@@ -107,6 +100,13 @@ function parseChunkLine(line: string): StreamChunk | null {
   }
 }
 
+/** Hard cap for retained partial-frame data in the SSE buffer (64 KB). */
+export const MAX_PARTIAL_FRAME_CHARS = 64 * 1024;
+/** Hard cap on accumulated parsed chunks retained for final reduce. */
+export const MAX_CHAT_CHUNKS = 1024;
+/** Hard cap on retained assistant content assembled from the stream (1 MiB). */
+export const MAX_RETAINED_CONTENT_CHARS = 1024 * 1024;
+
 /** Read a chat/completions SSE body and reduce it, emitting content deltas live. */
 export async function consumeChatCompletionsStream(
   res: Response,
@@ -118,22 +118,44 @@ export async function consumeChatCompletionsStream(
   const chunks: StreamChunk[] = [];
   const decoder = new TextDecoder();
   let buffer = '';
+  let retainedContentChars = 0;
   const handle = (c: StreamChunk | null) => {
     if (!c) return;
     const delta = c.choices?.[0]?.delta?.content;
-    if (onToken && typeof delta === 'string' && delta) onToken(delta);
+    if (onToken && typeof delta === 'string' && delta) {
+      // Still emit live tokens, but stop retaining more content beyond the hard cap.
+      if (retainedContentChars < MAX_RETAINED_CONTENT_CHARS) {
+        const room = MAX_RETAINED_CONTENT_CHARS - retainedContentChars;
+        const emit = delta.length <= room ? delta : delta.slice(0, room);
+        retainedContentChars += emit.length;
+        onToken(emit);
+      }
+    }
+    // Enforce hard cap on retained chunks to prevent memory exhaustion
+    if (chunks.length >= MAX_CHAT_CHUNKS) {
+      chunks.shift();
+    }
     chunks.push(c);
   };
   for (;;) {
     const { done, value } = await reader.read();
     if (done) break;
     buffer += decoder.decode(value, { stream: true });
+    // Enforce hard cap on partial-frame buffer to prevent memory exhaustion
+    if (buffer.length > MAX_PARTIAL_FRAME_CHARS) {
+      buffer = buffer.slice(-MAX_PARTIAL_FRAME_CHARS);
+    }
     const lines = buffer.split('\n');
     buffer = lines.pop() ?? '';
     for (const line of lines) handle(parseChunkLine(line));
   }
   handle(parseChunkLine(buffer));
 
   // Final reduce WITHOUT onToken (already emitted above) to assemble the result.
-  return reduceChatChunks(chunks);
-}
+  const reduced = reduceChatChunks(chunks);
+  const content = reduced.choices[0]?.message.content;
+  if (typeof content === 'string' && content.length > MAX_RETAINED_CONTENT_CHARS) {
+    reduced.choices[0].message.content = content.slice(0, MAX_RETAINED_CONTENT_CHARS);
+  }
+  return reduced;
+}

src/adapters/codexResponses.test.ts

@@ -9,6 +9,9 @@ import {
   reduceResponsesEvents,
   resolveReasoningEffort,
   selectDefaultCodexResponseModel,
+  MAX_FRAME_LENGTH,
+  MAX_RETAINED_EVENTS,
+  MAX_REASONING_BUF,
 } from './codexResponses.js';
 import { runAgenticLoop, type ChatMessage } from './agenticLoop.js';
 import { RateLimitError } from './rateLimitError.js';
@@ -18,6 +21,14 @@ afterEach(() => {
   vi.unstubAllGlobals();
 });
 
+describe('Codex Responses SSE retention bounds (AGT-3429)', () => {
+  it('documents hard caps for partial frames, retained events, and reasoning buffers', () => {
+    expect(MAX_FRAME_LENGTH).toBe(64 * 1024);
+    expect(MAX_RETAINED_EVENTS).toBe(4096);
+    expect(MAX_REASONING_BUF).toBe(64 * 1024);
+  });
+});
+
 describe('parseWorkerOutput command backfill', () => {
   const adapter = new CodexResponsesAdapter();
 

src/adapters/codexResponses.ts

@@ -212,6 +212,13 @@ export function reduceResponsesEvents(events: SseEvent[]): ChatLikeResponse {
   };
 }
 
+/** Hard cap for retained partial-frame data in the SSE buffer (64 KB). */
+export const MAX_FRAME_LENGTH = 64 * 1024;
+/** Hard cap on retained parsed SSE events before reduceResponsesEvents (4096). */
+export const MAX_RETAINED_EVENTS = 4096;
+/** Hard cap on retained reasoning partial text (64 KB). */
+export const MAX_REASONING_BUF = 64 * 1024;
+
 /** Parse a `data: {json}` SSE line into an event, or null for keep-alives/[DONE]. */
 function parseSseLine(line: string): SseEvent | null {
   const trimmed = line.trim();
@@ -257,9 +264,16 @@ async function consumeResponsesStream(
   const handle = (ev: SseEvent | null) => {
     if (!ev) return;
     events.push(ev);
+    // Enforce hard cap on retained events to prevent memory exhaustion
+    if (events.length > MAX_RETAINED_EVENTS) {
+      events.shift();
+    }
     if (onToken && ev.type === 'response.output_text.delta' && ev.delta) onToken(ev.delta);
     if (onReasoning && ev.type === 'response.reasoning_summary_text.delta' && ev.delta) {
       reasoningBuf += ev.delta;
+      if (reasoningBuf.length > MAX_REASONING_BUF) {
+        reasoningBuf = reasoningBuf.slice(-MAX_REASONING_BUF);
+      }
       flushReasoning(false);
     }
     // End of a summary part → flush whatever partial line remains.
@@ -271,6 +285,10 @@ async function consumeResponsesStream(
     const { done, value } = await reader.read();
     if (done) break;
     buffer += decoder.decode(value, { stream: true });
+    // Enforce hard cap on partial-frame buffer to prevent memory exhaustion
+    if (buffer.length > MAX_FRAME_LENGTH) {
+      buffer = buffer.slice(-MAX_FRAME_LENGTH);
+    }
     const lines = buffer.split('\n');
     buffer = lines.pop() ?? '';
     for (const line of lines) handle(parseSseLine(line));