diff --git a/apps/website/content/docs/langgraph/api/api-docs.json b/apps/website/content/docs/langgraph/api/api-docs.json index ab7d7d4ef..8a727c72a 100644 --- a/apps/website/content/docs/langgraph/api/api-docs.json +++ b/apps/website/content/docs/langgraph/api/api-docs.json @@ -946,6 +946,12 @@ "description": "Custom message deserializer for non-standard message formats.", "optional": true }, + { + "name": "transcriptNodeNames", + "type": "string[]", + "description": "LangGraph node names whose `messages-tuple` LLM chunks should be projected\ninto the main chat transcript. Omit to accept all top-level message chunks.", + "optional": true + }, { "name": "transport", "type": "AgentTransport", @@ -1082,6 +1088,12 @@ "description": "Custom message deserializer for non-standard message formats.", "optional": true }, + { + "name": "transcriptNodeNames", + "type": "string[]", + "description": "LangGraph node names whose `messages-tuple` LLM chunks should be projected\ninto the main chat transcript. Omit to accept all top-level message chunks.\n\nUse this when a graph has side-effect LLM nodes, such as title generation,\nwhose streamed model output should not render as assistant chat content.", + "optional": true + }, { "name": "transport", "type": "AgentTransport", diff --git a/apps/website/content/docs/langgraph/api/provide-agent.mdx b/apps/website/content/docs/langgraph/api/provide-agent.mdx index 5e0e28867..ed0140ed8 100644 --- a/apps/website/content/docs/langgraph/api/provide-agent.mdx +++ b/apps/website/content/docs/langgraph/api/provide-agent.mdx @@ -41,6 +41,7 @@ bootstrapApplication(AppComponent, { | `telemetry` | `AgentRuntimeTelemetrySink \| false` | Optional app-owned telemetry sink. No telemetry is emitted unless this is provided. | | `filterSubagentMessages` | `boolean` | When true, subagent messages are filtered from the main messages signal. | | `subagentToolNames` | `string[]` | Tool names that indicate a subagent invocation. | +| `transcriptNodeNames` | `string[]` | LangGraph node names whose `messages-tuple` chunks should stream into the main chat transcript. Omit to accept all top-level chunks. | ## Singleton model @@ -56,6 +57,18 @@ provideAgent({ const chat = injectAgent(); ``` +## Transcript node filtering + +LangGraph streams `messages-tuple` chunks for every LLM node in a run. If your graph has side-effect LLM nodes, such as a title generator or evaluator, set `transcriptNodeNames` so only your conversational node updates `messages()`. + +```ts +provideAgent({ + apiUrl: 'https://api.example.com', + assistantId: 'support-agent', + transcriptNodeNames: ['generate'], +}); +``` + ## Test transports `transport` is an object that implements `AgentTransport`, not an Angular class token. Create an instance before passing it to `provideAgent()`. diff --git a/examples/chat/angular/e2e/aimock-runner.ts b/examples/chat/angular/e2e/aimock-runner.ts index 5392cb777..b95813e5e 100644 --- a/examples/chat/angular/e2e/aimock-runner.ts +++ b/examples/chat/angular/e2e/aimock-runner.ts @@ -47,15 +47,9 @@ function loadFixtureEntries(fixturePath: string): FixtureFileEntry[] { export async function startAimock(opts: AimockStartOptions): Promise { const entries = loadFixtureEntries(opts.fixturePath); - // Use a large chunkSize so each response arrives in 1-2 SSE deltas. This - // intentionally turns off the partial-markdown streaming path for harness - // tests: structural assertions (code fence, list) measure the FINAL rendered - // DOM, not the progressive render. With aggressive default chunking, the - // partial-markdown parser sometimes can't recover a triple-backtick fence - // that gets split mid-token, and the final state ends up as inline - // instead of
. Streaming-progressive behavior is covered by the
-  // Phase 1 unit-variance tables; the e2e harness is for final-state
-  // invariants and cross-stack integration.
+  // Use a large default chunkSize so ordinary fixture responses arrive in 1-2
+  // SSE deltas. Most e2e assertions measure the final rendered DOM, while
+  // targeted streaming regressions opt into smaller per-fixture chunks.
   const mock = new LLMock({ port: 0, chunkSize: 4096 });
   if (entries.length > 0) {
     mock.addFixturesFromJSON(entries as never);
diff --git a/examples/chat/angular/e2e/fixtures/streaming-markdown.json b/examples/chat/angular/e2e/fixtures/streaming-markdown.json
new file mode 100644
index 000000000..aa63f2104
--- /dev/null
+++ b/examples/chat/angular/e2e/fixtures/streaming-markdown.json
@@ -0,0 +1,28 @@
+{
+  "fixtures": [
+    {
+      "match": { "userMessage": "stream a markdown comparison table regression" },
+      "response": {
+        "content": "Here is the comparison:\n\n| Name | Mental model | When to use |\n| --- | --- | --- |\n| Angular Signals | Synchronous value graph | Local component state |\n| RxJS | Event stream | Async flows and cancellation |\n| zone.js | Async task patching | Zone-based change detection |\n\nDone."
+      },
+      "chunkSize": 4,
+      "latency": 75
+    },
+    {
+      "match": { "userMessage": "stream a blockquote then a markdown table regression" },
+      "response": {
+        "content": "> First line of the quote.\n> Second line of the quote.\n\n| Issue | Expected behavior | Verification |\n| --- | --- | --- |\n| Button click does not update UI | UI reflects new state immediately | Click button and observe state |\n| Slow initial render | Main content appears within target | Measure first meaningful paint |"
+      },
+      "chunkSize": 6,
+      "latency": 25
+    },
+    {
+      "match": { "userMessage": "stream a TypeScript code fence regression" },
+      "response": {
+        "content": "Here is the snippet:\n\n```typescript\nconst answer = 42;\n```\n\nThe constant is available for later use."
+      },
+      "chunkSize": 3,
+      "latency": 35
+    }
+  ]
+}
diff --git a/examples/chat/angular/e2e/markdown-surfaces.spec.ts b/examples/chat/angular/e2e/markdown-surfaces.spec.ts
index ae8f343a6..b2f2f2ea1 100644
--- a/examples/chat/angular/e2e/markdown-surfaces.spec.ts
+++ b/examples/chat/angular/e2e/markdown-surfaces.spec.ts
@@ -1,6 +1,13 @@
 // SPDX-License-Identifier: MIT
-import { test, expect, type Locator } from '@playwright/test';
-import { sendPromptAndWait } from './test-helpers';
+import { test, expect, type Locator, type Page } from '@playwright/test';
+import {
+  attachBrowserHygiene,
+  messageInput,
+  openDemo,
+  sendButton,
+  sendPromptAndWait,
+  waitForFinalAssistant,
+} from './test-helpers';
 
 test('heading: assistant bubble renders an 

', async ({ page }) => { const bubble = await sendPromptAndWait(page, 'respond with a heading'); @@ -55,6 +62,70 @@ test('markdown checklist matrix: rich markdown renders with escaped html', async await expect(bubble).toContainText(""); }); +test('streaming markdown table: keeps in-progress rows inside one table', async ({ page }) => { + const hygiene = attachBrowserHygiene(page); + await sendPrompt(page, 'stream a markdown comparison table regression'); + + const streamingAssistant = latestAssistant(page); + await expect(streamingAssistant).toBeAttached({ timeout: 15_000 }); + + const samples = await collectStreamingSamples(page, streamingAssistant, 2_000); + expect(contentChangedAcross(samples)).toBe(true); + const tableSamples = samples.filter((sample) => sample.tableCount > 0); + expect(tableSamples.length).toBeGreaterThan(2); + expect(tableSamples.every((sample) => sample.tableCount <= 1)).toBe(true); + expect(tableSamples.every((sample) => sample.rowsOutsideTable === 0)).toBe(true); + expect(tableSamples.every((sample) => sample.detachedTableCellText.length === 0)).toBe(true); + + const bubble = await waitForFinalAssistant(page); + await expect(bubble.locator('table')).toHaveCount(1); + await expect(bubble.locator('thead th')).toHaveText(['Name', 'Mental model', 'When to use']); + await expect(bubble.locator('tbody tr')).toHaveCount(3); + await expect(bubble.locator('tbody tr').nth(2)).toContainText('zone.js'); + await expect.poll(async () => tableColumnsAlign(bubble)).toBe(true); + expect(hygiene.consoleErrors).toEqual([]); + expect(hygiene.failedRequests).toEqual([]); +}); + +test('streaming markdown table: blockquote followed by table does not throw', async ({ page }) => { + const hygiene = attachBrowserHygiene(page); + await sendPrompt(page, 'stream a blockquote then a markdown table regression'); + + const bubble = await waitForFinalAssistant(page); + await expect(bubble.locator('blockquote')).toBeVisible(); + await expect(bubble.locator('blockquote')).toContainText('First line of the quote.'); + await expect(bubble.locator('table')).toHaveCount(1); + await expect(bubble.locator('thead th')).toHaveText([ + 'Issue', + 'Expected behavior', + 'Verification', + ]); + await expect(bubble.locator('tbody tr')).toHaveCount(2); + expect(hygiene.consoleErrors).toEqual([]); + expect(hygiene.failedRequests).toEqual([]); +}); + +test('streaming code fence: suppresses closing fence marker while streaming', async ({ page }) => { + const hygiene = attachBrowserHygiene(page); + await sendPrompt(page, 'stream a TypeScript code fence regression'); + + const streamingAssistant = latestAssistant(page); + await expect(streamingAssistant).toBeAttached({ timeout: 15_000 }); + + const samples = await collectStreamingSamples(page, streamingAssistant, 4_000); + expect(contentChangedAcross(samples)).toBe(true); + const codeSamples = samples.filter((sample) => sample.codeBlockCount > 0); + expect(codeSamples.length).toBeGreaterThan(2); + expect(codeSamples.every((sample) => !sample.hasRawFenceMarker)).toBe(true); + + const bubble = await waitForFinalAssistant(page); + await expect(bubble.locator('pre code')).toHaveCount(1); + await expect(bubble.locator('pre code')).toContainText('const answer = 42'); + expect(await bubbleContainsRawFenceMarker(bubble)).toBe(false); + expect(hygiene.consoleErrors).toEqual([]); + expect(hygiene.failedRequests).toEqual([]); +}); + async function tableColumnsAlign(bubble: Locator): Promise { const table = bubble.locator('table').first(); return table.evaluate((el) => { @@ -72,3 +143,78 @@ async function tableColumnsAlign(bubble: Locator): Promise { }); }); } + +async function sendPrompt(page: Page, prompt: string): Promise { + await openDemo(page, '/embed'); + await messageInput(page).fill(prompt); + await sendButton(page).click(); +} + +function latestAssistant(page: Page): Locator { + return page.locator('chat-message[data-role="assistant"]').last(); +} + +interface StreamingMarkdownSample { + readonly contentTextLength: number; + readonly tableCount: number; + readonly rowsOutsideTable: number; + readonly detachedTableCellText: string[]; + readonly codeBlockCount: number; + readonly hasRawFenceMarker: boolean; +} + +async function collectStreamingSamples( + page: Page, + bubble: Locator, + minimumDurationMs: number, +): Promise { + const samples: StreamingMarkdownSample[] = []; + const startedAt = Date.now(); + + do { + samples.push(await sampleStreamingMarkdown(bubble)); + await page.waitForTimeout(75); + } while (Date.now() - startedAt < minimumDurationMs); + + return samples; +} + +async function sampleStreamingMarkdown(bubble: Locator): Promise { + return bubble.evaluate((el) => { + const looksLikeTableRowFragment = (text: string): boolean => ( + /\|/.test(text) || /Angular Signals|RxJS|zone\.js/.test(text) + ); + const tables = Array.from(el.querySelectorAll('table')); + const rowsOutsideTable = Array.from(el.querySelectorAll('tr')).filter( + (row) => !row.closest('table'), + ).length; + const detachedTableCellText: string[] = []; + const walker = document.createTreeWalker(el, NodeFilter.SHOW_TEXT); + let node = walker.nextNode(); + while (node) { + const parent = node.parentElement; + const text = node.textContent?.trim() ?? ''; + if (text && !parent?.closest('table') && looksLikeTableRowFragment(text)) { + detachedTableCellText.push(text); + } + node = walker.nextNode(); + } + + return { + contentTextLength: el.textContent?.length ?? 0, + tableCount: tables.length, + rowsOutsideTable, + detachedTableCellText, + codeBlockCount: el.querySelectorAll('pre code').length, + hasRawFenceMarker: el.textContent?.includes('```') ?? false, + }; + }); +} + +async function bubbleContainsRawFenceMarker(bubble: Locator): Promise { + return bubble.evaluate((el) => el.textContent?.includes('```') ?? false); +} + +function contentChangedAcross(samples: StreamingMarkdownSample[]): boolean { + return new Set(samples.map((sample) => sample.contentTextLength)).size > 2; +} diff --git a/examples/chat/angular/e2e/test-helpers.ts b/examples/chat/angular/e2e/test-helpers.ts index 6702cce7d..0857b8424 100644 --- a/examples/chat/angular/e2e/test-helpers.ts +++ b/examples/chat/angular/e2e/test-helpers.ts @@ -12,6 +12,7 @@ export function attachBrowserHygiene(page: Page): { if (msg.type() !== 'error') return; const text = msg.text(); if (/PostHog|ERR_NAME_NOT_RESOLVED|license/i.test(text)) return; + if (/409 \(Conflict\)/i.test(text)) return; consoleErrors.push(text); }); page.on('pageerror', (err) => { diff --git a/examples/chat/angular/src/app/shell/demo-shell.component.ts b/examples/chat/angular/src/app/shell/demo-shell.component.ts index 48d8bfb61..30d2e8db7 100644 --- a/examples/chat/angular/src/app/shell/demo-shell.component.ts +++ b/examples/chat/angular/src/app/shell/demo-shell.component.ts @@ -164,6 +164,9 @@ export function isConflict(err: unknown): boolean { // subagent dispatches and to materialize agent.subagents() from the // resulting tools:-namespaced stream events. subagentToolNames: ['research'], + // The canonical graph has side-effect LLM nodes such as generate_title; + // only the generate node's token stream belongs in the chat transcript. + transcriptNodeNames: ['generate'], telemetry: (event) => telemetrySink?.(event), }), { provide: DEMO_AGENT, useFactory: () => inject(DemoShell).agent }, diff --git a/libs/chat/src/lib/streaming/streaming-markdown.table-stream.spec.ts b/libs/chat/src/lib/streaming/streaming-markdown.table-stream.spec.ts index aa458bbbb..675336457 100644 --- a/libs/chat/src/lib/streaming/streaming-markdown.table-stream.spec.ts +++ b/libs/chat/src/lib/streaming/streaming-markdown.table-stream.spec.ts @@ -111,6 +111,30 @@ describe('ChatStreamingMdComponent — streaming table rendering', () => { } }); + it('streams a realistic comparison table as one table across small chunks', () => { + host.streaming.set(true); + const content = + 'Here is the comparison:\n\n' + + '| Name | Mental model | When to use |\n' + + '| --- | --- | --- |\n' + + '| Angular Signals | Synchronous value graph | Local component state |\n' + + '| RxJS | Event stream | Async flows and cancellation |\n' + + '| zone.js | Async task patching | Zone-based change detection |\n\n' + + 'Done.'; + + for (let i = 6; i <= content.length; i += 6) { + grow(content.slice(0, i)); + const tableCount = el.querySelectorAll('table').length; + if (tableCount > 0) { + expect(tableCount, `one table at ${JSON.stringify(content.slice(0, i).slice(-40))}`).toBe(1); + } + expect( + [...el.querySelectorAll('p')].some((p) => (p.textContent || '').includes('| zone.js')), + `no detached row paragraph at ${JSON.stringify(content.slice(0, i).slice(-40))}`, + ).toBe(false); + } + }); + it('keeps a finalized partial body row in the table when the stream pauses', () => { vi.useFakeTimers(); try { diff --git a/libs/langgraph/src/lib/agent.provider.ts b/libs/langgraph/src/lib/agent.provider.ts index f85c4c847..d3400f8ab 100644 --- a/libs/langgraph/src/lib/agent.provider.ts +++ b/libs/langgraph/src/lib/agent.provider.ts @@ -47,6 +47,11 @@ export interface AgentConfig< filterSubagentMessages?: boolean; /** Tool names that indicate a subagent invocation. */ subagentToolNames?: string[]; + /** + * LangGraph node names whose `messages-tuple` LLM chunks should be projected + * into the main chat transcript. Omit to accept all top-level message chunks. + */ + transcriptNodeNames?: string[]; } /** @@ -84,6 +89,7 @@ function agentFactory(): LangGraphAgent { ...(config.telemetry !== undefined ? { telemetry: config.telemetry } : {}), ...(config.filterSubagentMessages !== undefined ? { filterSubagentMessages: config.filterSubagentMessages } : {}), ...(config.subagentToolNames !== undefined ? { subagentToolNames: config.subagentToolNames } : {}), + ...(config.transcriptNodeNames !== undefined ? { transcriptNodeNames: config.transcriptNodeNames } : {}), }); } diff --git a/libs/langgraph/src/lib/agent.types.ts b/libs/langgraph/src/lib/agent.types.ts index fadbc4551..598343902 100644 --- a/libs/langgraph/src/lib/agent.types.ts +++ b/libs/langgraph/src/lib/agent.types.ts @@ -283,6 +283,14 @@ export interface AgentOptions { filterSubagentMessages?: boolean; /** Tool names that indicate a subagent invocation. */ subagentToolNames?: string[]; + /** + * LangGraph node names whose `messages-tuple` LLM chunks should be projected + * into the main chat transcript. Omit to accept all top-level message chunks. + * + * Use this when a graph has side-effect LLM nodes, such as title generation, + * whose streamed model output should not render as assistant chat content. + */ + transcriptNodeNames?: string[]; } // ── SubagentStreamRef ──────────────────────────────────────────────────────── 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 9804109bb..7e6fe7978 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.spec.ts @@ -1459,12 +1459,12 @@ describe('stream-manager.bridge — captured streaming replay (Finding C)', () = describe('identity-based delta merge (messages-tuple)', () => { const META = { langgraph_node: 'chatbot' }; - function setup() { + function setup(extraOptions: Record = {}) { const transport = new MockAgentTransport(); const subjects = makeSubjects(); const destroy$ = new Subject(); const bridge = createStreamManagerBridge({ - options: { apiUrl: '', assistantId: 'test', transport }, + options: { apiUrl: '', assistantId: 'test', transport, ...extraOptions }, subjects, threadId$: of(null), destroy$: destroy$.asObservable(), @@ -1586,6 +1586,28 @@ describe('identity-based delta merge (messages-tuple)', () => { destroy$.next(); }); + it('ignores message tuple chunks from non-transcript nodes when configured', async () => { + const { transport, subjects, destroy$, bridge } = setup({ + transcriptNodeNames: ['chatbot'], + }); + bridge.submit({}); + transport.emit([ + tupleEvent('ai-1', 'Here is the answer.'), + { + type: 'messages', + data: [ + { id: 'title-1', type: 'ai', content: 'This title must not render.' }, + { langgraph_node: 'generate_title' }, + ], + messageMetadata: { langgraph_node: 'generate_title' }, + } as any, + ]); + transport.close(); + await new Promise(r => setTimeout(r, 10)); + expect(lastAiContent(subjects)).toBe('Here is the answer.'); + destroy$.next(); + }); + it('values-sync mid-run keeps snapshot semantics (lagging state does not rewind)', async () => { const { transport, subjects, destroy$, bridge } = setup(); bridge.submit({}); diff --git a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts index 9b408f498..ef181c585 100644 --- a/libs/langgraph/src/lib/internals/stream-manager.bridge.ts +++ b/libs/langgraph/src/lib/internals/stream-manager.bridge.ts @@ -506,6 +506,11 @@ export function createStreamManagerBridge