Merge pull request #791 from cacheplane/blove/streaming-markdown-e2e-regressions

Filter transcript streaming nodes

Author: blove

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",

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()`.

examples/chat/angular/e2e/aimock-runner.ts

@@ -47,15 +47,9 @@ function loadFixtureEntries(fixturePath: string): FixtureFileEntry[] {
 export async function startAimock(opts: AimockStartOptions): Promise<AimockHandle> {
   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 <code>
-  // instead of <pre><code>. 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);

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
+    }
+  ]
+}

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 <h1>', 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("<script>alert('xss')</script>");
 });
 
+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<boolean> {
   const table = bubble.locator('table').first();
   return table.evaluate((el) => {
@@ -72,3 +143,78 @@ async function tableColumnsAlign(bubble: Locator): Promise<boolean> {
     });
   });
 }
+
+async function sendPrompt(page: Page, prompt: string): Promise<void> {
+  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<StreamingMarkdownSample[]> {
+  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<StreamingMarkdownSample> {
+  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<boolean> {
+  return bubble.evaluate((el) => el.textContent?.includes('```') ?? false);
+}
+
+function contentChangedAcross(samples: StreamingMarkdownSample[]): boolean {
+  return new Set(samples.map((sample) => sample.contentTextLength)).size > 2;
+}

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) => {

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:<id>-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 },

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 {

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<T>(): LangGraphAgent<T> {
     ...(config.telemetry !== undefined ? { telemetry: config.telemetry } : {}),
     ...(config.filterSubagentMessages !== undefined ? { filterSubagentMessages: config.filterSubagentMessages } : {}),
     ...(config.subagentToolNames !== undefined ? { subagentToolNames: config.subagentToolNames } : {}),
+    ...(config.transcriptNodeNames !== undefined ? { transcriptNodeNames: config.transcriptNodeNames } : {}),
   });
 }
 

libs/langgraph/src/lib/agent.types.ts

@@ -283,6 +283,14 @@ export interface AgentOptions<T, _ResolvedBag extends BagTemplate> {
   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 ────────────────────────────────────────────────────────