fix(langgraph): stop streamed chunks being dropped by prefix-based dedupe#751
Conversation
Root cause of intermittent mid-stream character loss (tables collapsing to raw pipes): accumulateContent's string-prefix dedupe guard drops any delta that coincidentally prefixes the accumulated message — every bare "|" token in a table. Proven by byte-identical replay of captured wire events. Fix: merge by declared event kind (tuple=delta → unconditional append; messages/partial=snapshot → prefix reconcile) with an identity-based canonical-id backstop replacing the text heuristic. TDD via the existing bridge spec harness + a live 3-layer Chrome MCP gate. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
5 tasks: red tests via the existing bridge spec harness, mode-aware merge (delta=append, snapshot=prefix reconcile) + canonical-id backstop, cross- suite gates, a REQUIRED live Chrome MCP gate (wire==accumulation across ≥6 table runs, temp recorder reverted), PR + merge. Spec refined: values-sync keeps snapshot semantics (lagging mid-run state must not rewind or mark canonical); finality is marked only by the canonical reasoning+text shape. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ing chunks mergeMessages/accumulateContent in stream-manager.bridge.ts merges messages-tuple chunks via string-prefix heuristics. The guard `existingText.startsWith(incomingText) -> keep existing` silently drops any delta that coincidentally prefixes the accumulated message, which corrupts streamed markdown tables (every bare "|" token vanishes). Pins the correct identity-based behavior with 6 cases: every delta chunk must be preserved (including prefix-coinciding ones), canonical reasoning+text arrays still replace + block stragglers, a new run's messages accumulate independently, and messages/partial + values-sync snapshot semantics are unaffected (regression baselines).
…t prefixes
Tuple events are deltas: append unconditionally. The old prefix-based dedupe
guard dropped any delta that coincidentally prefixed the accumulated message —
every bare pipe token in a streamed markdown table — silently corrupting
content mid-stream (wire had 48 pipes, accumulation 35; replaying captured
events through the guard reproduced the corruption byte-identically).
Snapshot payloads (messages/partial, values-sync) keep prefix reconciliation.
Stale post-final stragglers are now suppressed by canonical message id, not
by comparing text to text.
Also fixes a pre-existing bridge spec fixture ("merges message tuple events
into the existing transcript") that modeled tuple/messages events resending
the whole message-so-far ('hel' then 'hello') rather than genuine deltas —
that shape only made sense under the old prefix-reconciliation guard the
design explicitly removes from the delta path. Updated the fixture to emit
real incremental deltas ('hel' then 'lo'), preserving the test's original
intent (merge-by-id into one transcript entry).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Claude encountered an error after 4m 2s —— View job Code Review
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| incomingRaw['content'], | ||
| mode, | ||
| ); | ||
| if (targetId && isFinalCanonicalReasoningContent(incomingRaw['content'])) { |
There was a problem hiding this comment.
isFinalCanonicalReasoningContent(incomingRaw['content']) is evaluated here and also inside accumulateContent on line ~1213. Both calls parse the same array. Minor DRY concern — if the heuristic ever becomes more expensive or its definition changes, one of the two call sites is easy to miss. Consider computing it once before the accumulateContent call and threading the result in (e.g. a dedicated flag parameter or a small return-tagged struct).
| return { | ||
| type: 'messages', | ||
| data: [{ id, type: 'ai', content }, META], | ||
| messageMetadata: META, |
There was a problem hiding this comment.
function tupleEvent(id: string, content: unknown) {
return {
type: 'messages',
data: [{ id, type: 'ai', content }, META],
messageMetadata: META,
} as any;
}The rest of the test file uses satisfies StreamEvent for typed fixtures. as any works here (and normalizeMessages handles the data array path correctly), but switching to satisfies would keep the pattern consistent and catch future StreamEvent shape changes at compile time.
|
Claude encountered an error after 3m 40s —— View job Code Review
|
|
Claude encountered an error after 3m 55s —— View job Code Review
The fix is correct. Replacing text-comparison heuristics with declared-event-kind semantics is the right call — using string shapes to infer event semantics is fundamentally fragile, and this PR proves it with a byte-identical reproduction. The No blocking issues. Three minor things below (two already flagged by the prior review, still present). Orphaned JSDoc (minor)
Double
|
| incomingRaw['content'], | ||
| mode, | ||
| ); | ||
| if (targetId && isFinalCanonicalReasoningContent(incomingRaw['content'])) { |
There was a problem hiding this comment.
isFinalCanonicalReasoningContent(incomingRaw['content']) is computed twice: here (to decide whether to register the id) and again inside accumulateContent at line ~1213 (to decide whether to replace). Both traverse the same array. Not a correctness issue now, but if the heuristic ever gets more expensive or changes definition, having two call sites is fragile. Consider extracting the result once before the accumulateContent call and passing it in as a flag — or having accumulateContent return a tagged result that surfaces the detection. Fix this →
| return { | ||
| type: 'messages', | ||
| data: [{ id, type: 'ai', content }, META], | ||
| messageMetadata: META, |
There was a problem hiding this comment.
The rest of the test file uses satisfies StreamEvent for typed fixtures. as any here bypasses structural type-checking — if StreamEvent's shape changes (e.g., messageMetadata is renamed), this helper silently stays wrong while the other fixtures fail at compile time. Also worth noting: the helper uses data: [msg, META] (the SDK wire format) while the existing fixtures above use messages: [...] — both are handled by normalizeMessages, but the inconsistency is non-obvious. Switching to satisfies StreamEvent would surface if either format stops matching the type. Fix this →
| } | ||
|
|
||
| function accumulateContent(existing: unknown, incoming: unknown): string { | ||
| function accumulateContent(existing: unknown, incoming: unknown, mode: MergeMode = 'snapshot'): string { |
There was a problem hiding this comment.
The JSDoc block for this function (above) is orphaned — isFinalCanonicalReasoningContent with its own JSDoc was inserted between the accumulateContent doc and this function definition. IDE tooling attaches a JSDoc to the immediately following declaration, so the accumulateContent doc currently documents isFinalCanonicalReasoningContent instead. Move the accumulateContent JSDoc to immediately precede this line. Fix this →
|
Claude finished @blove's task in 3m 11s —— View job Code Review
The fix is correct and the latest two commits ( Three minor issues from previous reviews are still open (inline comments posted):
What's right
|
| } | ||
|
|
||
| function accumulateContent(existing: unknown, incoming: unknown): string { | ||
| function accumulateContent(existing: unknown, incoming: unknown, mode: MergeMode = 'snapshot'): string { |
There was a problem hiding this comment.
The JSDoc block for accumulateContent (lines 1155–1179) is still orphaned — isFinalCanonicalReasoningContent with its own JSDoc was inserted between it and this function definition. IDE tooling attaches a JSDoc to the immediately following declaration, so the accumulateContent doc currently documents isFinalCanonicalReasoningContent instead. Move the accumulateContent JSDoc to immediately precede this line. Fix this →
| incomingRaw['content'], | ||
| mode, | ||
| ); | ||
| if (targetId && isFinalCanonicalReasoningContent(incomingRaw['content'])) { |
There was a problem hiding this comment.
isFinalCanonicalReasoningContent(incomingRaw['content']) is evaluated here and also inside accumulateContent at line 1213 — same data, two traversals. Not harmful, but if the heuristic changes (e.g. adds a third block type to detect) one call site is easy to miss. Consider computing it once before accumulateContent and passing the result in as a boolean flag:
| if (targetId && isFinalCanonicalReasoningContent(incomingRaw['content'])) { | |
| const isCanonical = isFinalCanonicalReasoningContent(incomingRaw['content']); | |
| const accumulatedContent = accumulateContent( | |
| existing.content as unknown, | |
| incomingRaw['content'], | |
| mode, | |
| isCanonical, | |
| ); | |
| if (targetId && isCanonical) { |
| type: 'messages', | ||
| data: [{ id, type: 'ai', content }, META], | ||
| messageMetadata: META, | ||
| } as any; |
There was a problem hiding this comment.
The rest of the test file uses satisfies StreamEvent for typed fixtures. as any here bypasses structural type-checking — if StreamEvent's shape changes (e.g. messageMetadata is renamed), this helper stays wrong silently while typed fixtures fail at compile time. Also note: this helper uses the data: [msg, META] wire format while the existing fixtures use the already-processed messages: [...] property — both paths are covered by normalizeMessages, but the as any means the compiler won't catch if either format drifts from StreamEvent.
| } as any; | |
| } satisfies StreamEvent; |
|
Claude encountered an error after 2m 41s —— View job Code Review
|
|
Claude encountered an error after 1m 48s —— View job Code Review Fixes
|
|
Claude encountered an error after 2m 5s —— View job Code Review & Fix
|
|
Claude encountered an error after 1m 53s —— View job Code Review Fixes
|
|
Claude encountered an error after 2m 27s —— View job Code Review Fixes
|
1 participant
Problem
Streaming an assistant message over the LangGraph transport intermittently lost characters mid-stream. Most visible with markdown tables: the wire delivered
|--------------|--------|-------------|(48 pipes across the message) but the accumulated content rendered-----------------------------------|(35 pipes) — the table collapsed to raw text until the final message healed it.Root cause (proven)
accumulateContentmerged every incoming messages-tuple chunk with string-prefix heuristics. The guardexistingText.startsWith(incomingText) → keep existing(a stale-duplicate guess) drops any delta that coincidentally prefixes the accumulated message. A table message starts with|, so every bare"|"token the model streams is silently swallowed. Proof: replaying a corrupted run's captured wire events through this exact algorithm reproduced the observed accumulation byte-identically (13 dropped chunks, every one the single character|); the same replay without the guard yields wire-exact text. Intermittency = tokenization luck (|---fused tokens don't trip it; bare|tokens do).Fix — reconcile by identity and declared event kind, never by comparing text to text
event.messageMetadata) are deltas → append unconditionally. Both prefix guards removed from the delta path (the other guard,incoming.startsWith(existing) → replace, is the same bug class: a"| Gem"delta onto"|"must give"|| Gem").messages/partial+ values-sync keep snapshot semantics (mutual-prefix reconcile — correct for message-so-far payloads, including lagging mid-run values state).canonicalMessageIdsbackstop: the final-canonical reasoning+text replacement marks its message id; late deltas for a canonical id are ignored by id — the old guard's real purpose, now exact. Cleared per run and on thread switch.Verification
'hel'→'hello'cumulative resend) was corrected to a genuine delta — history shows it was synthetic from birth and encoded the buggy guard, not wire behavior.wire pipes == accumulated pipeson every run (48/48, 70/70, 48/48, 56/56, 35/35, 56/56), runs 3-6 byte-identical (accumulation === reconstructed wire text), zero table-collapse frames, zero divergences. Pre-fix baseline on the same harness: 48 vs 35.Closes out the streaming-table arc (partial-markdown 0.5.2/0.5.3, #743/#744): this was the last remaining cause of mid-stream table corruption.
🤖 Generated with Claude Code