StreamProcessor folds long text and tool-call streams in quadratic memory and time

Aperta
  • Qualcosa non va?
#1,963 0 commenti 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@jherr ci sta già lavorando.

Dal 1/11/2026.

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

waiting-on: maintainer
Describe the bug

StreamProcessor gets slower and heavier with every delta of one message. Two places cause it, both still in @tanstack/ai 1.0.1 (packages/ai/src/activities/chat/stream/processor.ts).

1. Text deltas hold a copy of the whole message per delta (memory and time).

handleTextMessageContentEvent, line 2343:

const nextText = delta !== '' ? currentText + delta : currentText
const textDelta = nextText.slice(currentText.length)
state.currentSegmentText = nextText
state.totalTextContent += textDelta

In V8, currentText + delta builds a rope. slice on a rope first flattens it into one buffer. For a result of 13 characters or more it then returns a sliced string that points into that buffer, so the delta is not copied out. totalTextContent concatenates those slices, and each one keeps a full flattened copy of the message alive. One message of 10,000 deltas of 16 characters (160 KB of text) holds about 760 MB after a forced GC. 5,000 deltas of 200 characters run Node out of memory at the default 4 GB heap.

textDelta always equals delta here, so the fix is:

state.currentSegmentText = nextText
state.totalTextContent += delta

2. Tool-call arguments are re-parsed in full on every delta (time).

handleToolCallArgsEvent, line 2515:

existingToolCall.arguments += chunk.delta || ''
existingToolCall.parsedArguments = this.jsonParser.parse(existingToolCall.arguments)

The partial parse is O(length of the arguments so far), so a call whose arguments arrive in n deltas costs O(n²). A tool writing a large file streams thousands of TOOL_CALL_ARGS deltas: 4,000 deltas of 100 characters take about 2 s, 10,000 about 12 s. Within the processor only getState() reads parsedArguments before TOOL_CALL_END or the completion path sets the final value. One fix is to parse on read:

// getState()
for (const [id, tc] of state.toolCalls) {
  if (tc.state === 'awaiting-input' || tc.state === 'input-streaming') {
    tc.parsedArguments = this.jsonParser.parse(tc.arguments)
  }
  toolCalls.set(id, tc)
}
Reproduction
import { StreamProcessor } from '@tanstack/ai'
import { setFlagsFromString } from 'node:v8'
import { runInNewContext } from 'node:vm'

setFlagsFromString('--expose-gc')
const gc = runInNewContext('gc')

gc()
const before = process.memoryUsage().heapUsed
const p = new StreamProcessor()
p.processChunk({ type: 'RUN_STARTED', runId: 'r', threadId: 't' })
p.processChunk({ type: 'TEXT_MESSAGE_START', messageId: 'm', role: 'assistant' })
for (let i = 0; i < 10_000; i++)
  p.processChunk({ type: 'TEXT_MESSAGE_CONTENT', messageId: 'm', delta: 'x'.repeat(16) })
gc()
const held = Math.round((process.memoryUsage().heapUsed - before) / 1e6)
console.log(held, 'MB held for', p.getMessages()[0].parts[0].content.length, 'characters')
// 0.61.0, whose code here matches 1.0.1: ~800 MB. With the fix: ~1 MB.

let parsed = 0
const q = new StreamProcessor({ jsonParser: { parse: (s) => ((parsed += s.length), undefined) } })
q.processChunk({ type: 'RUN_STARTED', runId: 'r', threadId: 't' })
q.processChunk({ type: 'TEXT_MESSAGE_START', messageId: 'm', role: 'assistant' })
q.processChunk({ type: 'TOOL_CALL_START', toolCallId: 'c', toolCallName: 'Write', parentMessageId: 'm' })
for (let i = 0; i < 10_000; i++)
  q.processChunk({ type: 'TOOL_CALL_ARGS', toolCallId: 'c', delta: 'x'.repeat(10) })
console.log(parsed, 'characters parsed for 100,000 characters of arguments')
// 0.61.0: 500,050,000. With the fix: 0.
Expected behavior

Folding a message of n deltas takes time and memory linear in its length.

Environment

@tanstack/ai 0.61.0 and 1.0.1, Node 24.21. We run the two fixes above as a local patch on 0.61.0 and can open a PR if that helps.

Lingua principale
TypeScript
Stelle
3.3k
Fork
378
Merge medio
1g 2h
PR unite (30g)
219

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di TanStack/ai

Tutte le issue di TanStack/ai

Issue simili

Altre issue su TypeScript

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.