diff --git a/apps/website/content/docs/chat/api/api-docs.json b/apps/website/content/docs/chat/api/api-docs.json index ecc3b6e92..94e7a6123 100644 --- a/apps/website/content/docs/chat/api/api-docs.json +++ b/apps/website/content/docs/chat/api/api-docs.json @@ -5382,7 +5382,21 @@ "optional": false } ], - "methods": [] + "methods": [ + { + "name": "resolve", + "signature": "resolve(child: MarkdownNode): Type | null", + "description": "", + "params": [ + { + "name": "child", + "type": "MarkdownNode", + "description": "", + "optional": false + } + ] + } + ] }, { "name": "MarkdownTextComponent", diff --git a/examples/chat/angular/e2e/markdown-surfaces.spec.ts b/examples/chat/angular/e2e/markdown-surfaces.spec.ts index 55f10a275..ae8f343a6 100644 --- a/examples/chat/angular/e2e/markdown-surfaces.spec.ts +++ b/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

', 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(""); }); + +async function tableColumnsAlign(bubble: Locator): Promise { + 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 + )); + }); + }); +} diff --git a/libs/chat/src/lib/markdown/views/markdown-table-row.component.spec.ts b/libs/chat/src/lib/markdown/views/markdown-table-row.component.spec.ts index 3d1e497b8..c1a89b4cb 100644 --- a/libs/chat/src/lib/markdown/views/markdown-table-row.component.spec.ts +++ b/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); + }); }); diff --git a/libs/chat/src/lib/markdown/views/markdown-table-row.component.ts b/libs/chat/src/lib/markdown/views/markdown-table-row.component.ts index 96432e107..80747cb38 100644 --- a/libs/chat/src/lib/markdown/views/markdown-table-row.component.ts +++ b/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: ` - + @for (child of node().children; track $index) { + @let comp = resolve(child); + @if (comp) { + + } + } `, providers: [ @@ -27,4 +34,11 @@ import { IS_HEADER_ROW } from '../markdown-table-row.token'; }) export class MarkdownTableRowComponent { readonly node = input.required(); + private readonly registry = inject(MARKDOWN_VIEW_REGISTRY); + + protected resolve(child: MarkdownNode): Type | null { + const entry = this.registry[child.type]; + if (!entry) return null; + return typeof entry === 'function' ? entry : entry.component; + } } diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts index d440c5acd..9804109bb 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts @@ -994,6 +994,10 @@ 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' }], @@ -1001,7 +1005,7 @@ describe('createStreamManagerBridge', () => { } 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(); + 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): 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(); + 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(); + }); +}); diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts index 93b634ce2..9b408f498 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts @@ -116,6 +116,10 @@ export function createStreamManagerBridge(); + // Message ids whose content is known-final (installed by a canonical + // replacement). Late streamed deltas for these ids are stale stragglers and + // are ignored — decided by identity, never by comparing text to text. + const canonicalMessageIds = new Set(); const queuedRuns: AgentQueueEntry[] = []; let drainingQueue = false; const subagentManager = new SubagentTracker({ @@ -162,6 +166,7 @@ export function createStreamManagerBridge, + mode: MergeMode = 'snapshot', + canonicalMessageIds?: Set, ): BaseMessage[] { const merged = [...existing]; for (const msg of incoming) { @@ -1071,6 +1082,13 @@ function mergeMessages( const existing = merged[idx]; const existingId = (existing as unknown as Record)['id']; const incomingRaw = msg as unknown as Record; + const targetId = (existingId ?? incomingRaw['id']) as string | undefined; + // Identity backstop: once a message's content is known-final, late + // streamed deltas for it are stale stragglers — ignore them outright. + if (mode === 'delta' && targetId && canonicalMessageIds?.has(targetId) + && !isFinalCanonicalReasoningContent(incomingRaw['content'])) { + continue; + } // Keep the *existing* id so downstream track-by-id sees stable identity. // For complex-content streaming (OpenAI gpt-5/o-series, Anthropic) the // SDK emits per-chunk *delta* arrays — not accumulated arrays — so a @@ -1081,7 +1099,11 @@ function mergeMessages( const accumulatedContent = accumulateContent( existing.content as unknown, incomingRaw['content'], + mode, ); + if (targetId && isFinalCanonicalReasoningContent(incomingRaw['content'])) { + canonicalMessageIds?.add(targetId); + } // Only accumulate reasoning when the incoming message explicitly carries // a `reasoning` field or complex-content array blocks with // type='reasoning'/'thinking'. Never use a plain string content value @@ -1132,12 +1154,22 @@ function mergeMessages( /** * Merge an incoming chunk's content into prior accumulated content for the - * same message id. + * same message id. Behavior is governed by `mode`, which reflects the + * DECLARED kind of the source event rather than a guess from comparing text: * - * - string + string → concat (delta append) - * - array + array → concat extracted text from existing + incoming blocks - * - array + string → use the string (server final-id swap) - * - empty existing → use incoming as-is + * - mode 'delta' (messages-tuple / `event.messageMetadata` truthy): the + * payload is a genuine per-chunk delta. Append unconditionally — a + * prefix-comparison "dedupe" here would silently drop legitimate tokens + * that coincide with the message-so-far (e.g. every bare "|" while + * streaming a markdown table). Staleness after the message goes canonical + * is instead handled by identity in `mergeMessages` (canonicalMessageIds). + * - mode 'snapshot' (messages/partial, values-sync): the payload carries the + * message-so-far, not a delta, so mutual prefix comparison picks the + * longer state and ignores stale shorter snapshots. + * + * In both modes, a "final canonical" reasoning+text array (see + * `isFinalCanonicalReasoningContent`) always replaces whatever was + * accumulated — it's the authoritative final message, not another chunk. * * We deliberately collapse complex content arrays to a string at this layer. * The langgraph-sdk client does not accumulate complex-content arrays the @@ -1166,7 +1198,7 @@ function isFinalCanonicalReasoningContent(content: unknown): boolean { return hasReasoning && hasText; } -function accumulateContent(existing: unknown, incoming: unknown): string { +function accumulateContent(existing: unknown, incoming: unknown, mode: MergeMode = 'snapshot'): string { const existingText = extractText(existing); const incomingText = extractText(incoming); @@ -1176,20 +1208,21 @@ function accumulateContent(existing: unknown, incoming: unknown): string { // dedupe from matching the streamed-chunk message after a partial chunk. if (existingText.length === 0) return incomingText; if (incomingText.length === 0) return existingText; - // Incoming is a strict-superset of accumulated (final-id swap with full content). + // Final-canonical detection applies in both modes: the authoritative + // "reasoning + text" array replaces whatever was accumulated. + if (isFinalCanonicalReasoningContent(incoming)) return incomingText; + if (mode === 'delta') { + // Tuple chunks are declared deltas. Append unconditionally — any + // text-comparison "dedupe" here can silently drop legitimate tokens + // that coincide with the message prefix (e.g. every bare "|" in a + // markdown table). Staleness is handled by identity in mergeMessages. + return existingText + incomingText; + } + // Snapshot mode (messages/partial, values-sync): payloads carry the + // message-so-far, so mutual prefix comparison picks the longer state and + // ignores stale shorter snapshots. if (incomingText.startsWith(existingText)) return incomingText; - // Existing already a strict-superset — chunk arrived after the canonical - // message merged in via values-sync. Keep what we have. if (existingText.startsWith(incomingText)) return existingText; - // Final-canonical detection: when incoming is the "reasoning + text" - // array shape that ships the authoritative final message after a - // streaming run, replace the partial streamed accumulator with the - // canonical text instead of appending. Without this branch a small - // formatting difference between the streamed accumulator and the - // canonical text breaks the prefix checks above and visible content - // is duplicated (`existingText + incomingText`). - if (isFinalCanonicalReasoningContent(incoming)) return incomingText; - // Otherwise treat incoming as a delta and append. return existingText + incomingText; }