fix(langgraph): stop streamed chunks being dropped by prefix-based dedupe

apps/website/content/docs/chat/api/api-docs.json

@@ -5382,7 +5382,21 @@
         "optional": false
       }
     ],
-    "methods": []
+    "methods": [
+      {
+        "name": "resolve",
+        "signature": "resolve(child: MarkdownNode): Type<unknown> | null",
+        "description": "",
+        "params": [
+          {
+            "name": "child",
+            "type": "MarkdownNode",
+            "description": "",
+            "optional": false
+          }
+        ]
+      }
+    ]
   },
   {
     "name": "MarkdownTextComponent",

examples/chat/angular/e2e/markdown-surfaces.spec.ts

@@ -1,5 +1,5 @@
 // SPDX-License-Identifier: MIT
-import { test, expect } from '@playwright/test';
+import { test, expect, type Locator } from '@playwright/test';
 import { sendPromptAndWait } from './test-helpers';
 
 test('heading: assistant bubble renders an <h1>', async ({ page }) => {
@@ -43,6 +43,7 @@ test('markdown checklist matrix: rich markdown renders with escaped html', async
   await expect(bubble.locator('table')).toBeVisible();
   await expect(bubble.locator('thead th')).toHaveText(['Name', 'Mental model', 'When to use']);
   await expect(bubble.locator('tbody tr')).toHaveCount(2);
+  await expect.poll(async () => tableColumnsAlign(bubble)).toBe(true);
   await expect(bubble.locator('blockquote')).toBeVisible();
   await expect(bubble).toContainText('This is a blockquote.');
   await expect(bubble.locator('a', { hasText: 'Angular' })).toHaveAttribute(
@@ -53,3 +54,21 @@ test('markdown checklist matrix: rich markdown renders with escaped html', async
   await expect(bubble.locator('script')).toHaveCount(0);
   await expect(bubble).toContainText("<script>alert('xss')</script>");
 });
+
+async function tableColumnsAlign(bubble: Locator): Promise<boolean> {
+  const table = bubble.locator('table').first();
+  return table.evaluate((el) => {
+    const headerCells = Array.from(el.querySelectorAll('thead th'));
+    const rows = Array.from(el.querySelectorAll('tbody tr'));
+    if (headerCells.length === 0 || rows.length === 0) return false;
+
+    const headerLefts = headerCells.map((cell) => cell.getBoundingClientRect().left);
+    return rows.every((row) => {
+      const cells = Array.from(row.querySelectorAll('td'));
+      if (cells.length !== headerLefts.length) return false;
+      return cells.every((cell, index) => (
+        Math.abs(cell.getBoundingClientRect().left - headerLefts[index]) <= 1
+      ));
+    });
+  });
+}

libs/chat/src/lib/markdown/views/markdown-table-row.component.spec.ts

@@ -4,7 +4,7 @@ import { describe, it, expect, beforeEach } from 'vitest';
 import { TestBed } from '@angular/core/testing';
 import { Component, signal } from '@angular/core';
 import { views } from '@threadplane/render';
-import type { MarkdownTableRowNode } from '@cacheplane/partial-markdown';
+import type { MarkdownTableCellNode, MarkdownTableRowNode } from '@cacheplane/partial-markdown';
 import { MarkdownTableRowComponent } from './markdown-table-row.component';
 import { MarkdownTableCellComponent } from './markdown-table-cell.component';
 import { MARKDOWN_VIEW_REGISTRY } from '../markdown-view-registry';
@@ -18,6 +18,15 @@ function makeRowNode(isHeader: boolean, children: MarkdownTableRowNode['children
   } as MarkdownTableRowNode;
 }
 
+function makeCellNode(id: number): MarkdownTableCellNode {
+  return {
+    id, type: 'table-cell', status: 'complete',
+    parent: null, index: null,
+    alignment: null,
+    children: [],
+  } as MarkdownTableCellNode;
+}
+
 @Component({
   standalone: true,
   imports: [MarkdownTableRowComponent],
@@ -64,4 +73,17 @@ describe('MarkdownTableRowComponent', () => {
     const tr = fixture.nativeElement.querySelector('tr');
     expect(tr.classList.contains('chat-md-table-row--header')).toBe(true);
   });
+
+  it('renders table-cell components directly under the table row', () => {
+    const fixture = TestBed.createComponent(HostComponent);
+    fixture.componentInstance.node.set(makeRowNode(false, [
+      makeCellNode(3),
+      makeCellNode(4),
+    ] as MarkdownTableRowNode['children']));
+    fixture.detectChanges();
+
+    const tr = fixture.nativeElement.querySelector('tr');
+    expect(tr.querySelector(':scope > chat-md-children')).toBeNull();
+    expect(tr.querySelectorAll(':scope > chat-md-table-cell').length).toBe(2);
+  });
 });

libs/chat/src/lib/markdown/views/markdown-table-row.component.ts

@@ -1,18 +1,25 @@
 // libs/chat/src/lib/markdown/views/markdown-table-row.component.ts
 // SPDX-License-Identifier: MIT
-import { Component, ChangeDetectionStrategy, input, computed, inject } from '@angular/core';
-import type { MarkdownTableRowNode } from '@cacheplane/partial-markdown';
-import { MarkdownChildrenComponent } from '../markdown-children.component';
+import { NgComponentOutlet } from '@angular/common';
+import { Component, ChangeDetectionStrategy, input, computed, inject, type Type } from '@angular/core';
+import type { ViewRegistry } from '@threadplane/render';
+import type { MarkdownNode, MarkdownTableRowNode } from '@cacheplane/partial-markdown';
+import { MARKDOWN_VIEW_REGISTRY } from '../markdown-view-registry';
 import { IS_HEADER_ROW } from '../markdown-table-row.token';
 
 @Component({
   selector: 'chat-md-table-row',
   standalone: true,
-  imports: [MarkdownChildrenComponent],
+  imports: [NgComponentOutlet],
   changeDetection: ChangeDetectionStrategy.OnPush,
   template: `
     <tr class="chat-md-table-row" [class.chat-md-table-row--header]="node().isHeader">
-      <chat-md-children [parent]="node()" />
+      @for (child of node().children; track $index) {
+        @let comp = resolve(child);
+        @if (comp) {
+          <ng-container *ngComponentOutlet="comp; inputs: { node: child }" />
+        }
+      }
     </tr>
   `,
   providers: [
@@ -27,4 +34,11 @@ import { IS_HEADER_ROW } from '../markdown-table-row.token';
 })
 export class MarkdownTableRowComponent {
   readonly node = input.required<MarkdownTableRowNode>();
+  private readonly registry = inject<ViewRegistry>(MARKDOWN_VIEW_REGISTRY);
+
+  protected resolve(child: MarkdownNode): Type<unknown> | null {
+    const entry = this.registry[child.type];
+    if (!entry) return null;
+    return typeof entry === 'function' ? entry : entry.component;
+  }
 }

libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts

@@ -994,14 +994,18 @@ describe('createStreamManagerBridge', () => {
     });
 
     bridge.submit({ messages: [{ type: 'human', content: 'hello' }] });
+    // Tuple/messages events with messageMetadata are declared deltas — each
+    // chunk carries only its incremental slice of text, never the
+    // message-so-far. 'hel' + 'lo' (not a resent 'hello') reflects real
+    // wire behavior and merges into the same transcript entry by id.
     transport.emit([{
       type: 'messages',
       messages: [{ id: 'ai-1', type: 'ai', content: 'hel' }],
       messageMetadata: { langgraph_node: 'model' },
     } satisfies StreamEvent]);
     transport.emit([{
       type: 'messages',
-      messages: [{ id: 'ai-1', type: 'ai', content: 'hello' }],
+      messages: [{ id: 'ai-1', type: 'ai', content: 'lo' }],
       messageMetadata: { langgraph_node: 'model' },
     } satisfies StreamEvent]);
     transport.close();
@@ -1451,3 +1455,152 @@ describe('stream-manager.bridge — captured streaming replay (Finding C)', () =
     expect(visible.length).toBeLessThanOrEqual(expected + 20);
   });
 });
+
+describe('identity-based delta merge (messages-tuple)', () => {
+  const META = { langgraph_node: 'chatbot' };
+
+  function setup() {
+    const transport = new MockAgentTransport();
+    const subjects = makeSubjects();
+    const destroy$ = new Subject<void>();
+    const bridge = createStreamManagerBridge({
+      options: { apiUrl: '', assistantId: 'test', transport },
+      subjects,
+      threadId$: of(null),
+      destroy$: destroy$.asObservable(),
+    });
+    return { transport, subjects, destroy$, bridge };
+  }
+
+  function tupleEvent(id: string, content: unknown) {
+    return {
+      type: 'messages',
+      data: [{ id, type: 'ai', content }, META],
+      messageMetadata: META,
+    } as any;
+  }
+
+  function lastAiContent(subjects: ReturnType<typeof makeSubjects>): string {
+    // Select the last AI message specifically (not just the last array
+    // entry): mergeMessages/preserveIds do not guarantee human messages
+    // sort before AI messages once a values-sync event appends an
+    // out-of-order human entry after an already-accumulating AI entry.
+    const msgs = subjects.messages$.value as Array<{ type?: string; content?: unknown }>;
+    const lastAi = [...msgs].reverse().find(m => m?.type === 'ai');
+    return typeof lastAi?.content === 'string' ? lastAi.content : JSON.stringify(lastAi?.content);
+  }
+
+  it('preserves every delta chunk, including ones that prefix the message (table pipes)', async () => {
+    const { transport, subjects, destroy$, bridge } = setup();
+    bridge.submit({});
+    const deltas = ['|', ' Gem', ' |', ' Color', ' |', '\n', '|', '---', '|', '---', '|', '\n', '|', ' Ruby', ' |', ' red', ' |'];
+    for (const d of deltas) transport.emit([tupleEvent('ai-1', d)]);
+    transport.close();
+    await new Promise(r => setTimeout(r, 10));
+    expect(lastAiContent(subjects)).toBe(deltas.join(''));
+    destroy$.next();
+  });
+
+  it('appends a multi-char delta that begins with the accumulated text', async () => {
+    const { transport, subjects, destroy$, bridge } = setup();
+    bridge.submit({});
+    transport.emit([tupleEvent('ai-1', '|')]);
+    transport.emit([tupleEvent('ai-1', '| Gem')]); // delta, NOT a superset echo
+    transport.close();
+    await new Promise(r => setTimeout(r, 10));
+    expect(lastAiContent(subjects)).toBe('|| Gem');
+    destroy$.next();
+  });
+
+  it('final canonical reasoning+text array replaces the accumulation and blocks late deltas', async () => {
+    const { transport, subjects, destroy$, bridge } = setup();
+    bridge.submit({});
+    transport.emit([tupleEvent('ai-1', '| a |')]);
+    transport.emit([tupleEvent('ai-1', ' | b |')]);
+    transport.emit([tupleEvent('ai-1', [
+      { type: 'reasoning', text: 'thought' },
+      { type: 'text', text: '| a | | b | done' },
+    ])]);
+    transport.emit([tupleEvent('ai-1', '|')]); // straggler after canonical — must be ignored
+    transport.close();
+    await new Promise(r => setTimeout(r, 10));
+    expect(lastAiContent(subjects)).toBe('| a | | b | done');
+    destroy$.next();
+  });
+
+  it('a new run resets canonical marking (fresh-id deltas accumulate again)', async () => {
+    // NOTE: MockAgentTransport.closed is permanent once close() is called —
+    // its stream() generator checks `!this.closed` on every fresh invocation,
+    // so a second submit() through the SAME instance would fall straight
+    // through to an immediate drain-and-return, never seeing events emitted
+    // after the second submit. Use a second transport/bridge instance for the
+    // second run, same pattern as the existing 'clears custom$ on a new
+    // submit' test above, which shares `subjects` but not the transport.
+    const { transport, subjects, destroy$, bridge } = setup();
+    bridge.submit({});
+    transport.emit([tupleEvent('ai-1', [
+      { type: 'reasoning', text: 'r' },
+      { type: 'text', text: 'final one' },
+    ])]);
+    transport.close();
+    await new Promise(r => setTimeout(r, 10));
+
+    const transport2 = new MockAgentTransport();
+    const destroy2$ = new Subject<void>();
+    const bridge2 = createStreamManagerBridge({
+      options: { apiUrl: '', assistantId: 'test', transport: transport2 },
+      subjects,
+      threadId$: of(null),
+      destroy$: destroy2$.asObservable(),
+    });
+    // Real "new run" submits carry a human message, which runStream()
+    // optimistically injects into messages$ before streaming starts. That
+    // human message is what gives mergeMessages' trailing-AI fallback a
+    // boundary to stop at (see mergeMessages: it walks backward from the
+    // tail and stops at the first human/tool/system message) — without it,
+    // a fresh id with no content overlap would keep accumulating onto the
+    // previous run's AI slot since nothing marks the run boundary.
+    bridge2.submit({ messages: [{ type: 'human', content: 'second question' }] });
+    transport2.emit([tupleEvent('ai-2', 'fresh')]);
+    transport2.emit([tupleEvent('ai-2', ' text')]);
+    transport2.close();
+    await new Promise(r => setTimeout(r, 10));
+    expect(lastAiContent(subjects)).toBe('fresh text');
+    destroy$.next();
+    destroy2$.next();
+  });
+
+  it('messages/partial snapshots still reconcile by prefix (regression)', async () => {
+    const { transport, subjects, destroy$, bridge } = setup();
+    bridge.submit({});
+    const partial = (content: string) => ({
+      type: 'messages/partial',
+      data: [{ id: 'ai-1', type: 'ai', content }],
+    } as any);
+    transport.emit([partial('| Gem')]);
+    transport.emit([partial('| Gem | Color |')]); // superset → replace
+    transport.emit([partial('| Gem')]);            // stale shorter snapshot → keep longer
+    transport.close();
+    await new Promise(r => setTimeout(r, 10));
+    expect(lastAiContent(subjects)).toBe('| Gem | Color |');
+    destroy$.next();
+  });
+
+  it('values-sync mid-run keeps snapshot semantics (lagging state does not rewind)', async () => {
+    const { transport, subjects, destroy$, bridge } = setup();
+    bridge.submit({});
+    transport.emit([tupleEvent('ai-1', '| a | b |')]);
+    transport.emit([tupleEvent('ai-1', ' | c |')]);
+    transport.emit([{
+      type: 'values',
+      data: { messages: [
+        { id: 'h-1', type: 'human', content: 'hi' },
+        { id: 'ai-1', type: 'ai', content: '| a | b |' },
+      ] },
+    } as any]);
+    transport.close();
+    await new Promise(r => setTimeout(r, 10));
+    expect(lastAiContent(subjects)).toBe('| a | b | | c |');
+    destroy$.next();
+  });
+});