fix(chat): stop streaming tables rendering as raw pipes mid-stream

ChatStreamingMdComponent called parser.finish() on every render where its
[streaming] input was false. finish() is destructive: an incomplete table
(a header row before its delimiter row) reverts to a CommonMark paragraph —
raw "| a | b |" text. The [streaming] flag (agent.isLoading() && i===last) is
not a reliable "still arriving" signal: it reads false at cold start and flaps
false mid-stream, so the parser kept being finalized while tokens were still
arriving and the whole table rendered as raw pipes until the message completed.
(Prose hid this because finalized prose looks identical to streaming prose;
tables are the first place finished != streaming is visible. partial-markdown
0.5.2 renders streaming tables correctly when the parser is not finished early.)

Fix: never finalize while content is still growing — the live parser.root
projection already renders everything (incl. streaming tables) since 0.5.x.
Finalize via a debounce: only after streaming is false AND no new content has
arrived for 600ms; any token or streaming=true flap re-arms it, so finish()
fires exactly once after the stream truly stops. No markdown view depends on
node status, so the delay has no visual cost. Also guard against pushing into
an already-finished parser.

Tests: new streaming-markdown.table-stream.spec (reproduces the bug + a
fake-timer flap test); updated the one variants assertion that depended on
synchronous finalize. Live-verified in examples/chat: a streamed table renders
a <table> every frame, never raw pipes.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: blove

libs/chat/src/lib/streaming/streaming-markdown.component.ts

@@ -8,6 +8,7 @@ import {
   effect,
   inject,
   input,
+  signal,
 } from '@angular/core';
 import {
   createPartialMarkdownParser,
@@ -22,6 +23,16 @@ import { MarkdownChildrenComponent } from '../markdown/markdown-children.compone
 import { cacheplaneMarkdownViews } from '../markdown/cacheplane-markdown-views';
 import { CitationsResolverService } from '../markdown/citations-resolver.service';
 
+// How long streaming must be false AND content stable before we finalize the
+// parser. finish() is only needed to mark final node status (not used visually)
+// and to revert a genuinely-truncated trailing construct to CommonMark — the
+// live `parser.root` projection already renders everything during streaming, so
+// this delay has NO visual cost. It must comfortably exceed real inter-chunk
+// gaps (e.g. the pause between a table's header row and its delimiter row) and
+// any `streaming` flag flap, so finalize never fires mid-stream and reverts an
+// in-progress table to raw "| a | b |" text.
+const FINALIZE_DEBOUNCE_MS = 600;
+
 /**
  * Renders streaming markdown by walking a @cacheplane/partial-markdown AST
  * through @threadplane/render's view registry.
@@ -77,6 +88,42 @@ export class ChatStreamingMdComponent {
         this.resolver.markdownDefs.set(r.citations ?? new Map());
       }
     });
+
+    // Debounced finalization. `finish()` is terminal and DESTRUCTIVE: it reverts
+    // an incomplete trailing construct to its CommonMark fallback — e.g. a table
+    // header before its delimiter row becomes raw "| a | b |" paragraph text. We
+    // must therefore never finalize while tokens are still arriving. The
+    // `streaming` input is not a reliable "still arriving" signal — it can flap
+    // false mid-stream, and at cold start it can read false for an entire live
+    // stream — so we finalize only once streaming is false AND no new content
+    // has arrived for a short, imperceptible window. Any new content or a
+    // streaming=true flap re-arms the timer, so finalize fires exactly once,
+    // after the stream truly stops. Until then the live `parser.root` projection
+    // (0.5.x) renders the in-progress content, including streaming tables.
+    let timer: ReturnType<typeof setTimeout> | null = null;
+    effect((onCleanup) => {
+      const isStreaming = this.streaming();
+      this.content(); // re-arm whenever new content arrives
+      if (timer) {
+        clearTimeout(timer);
+        timer = null;
+      }
+      if (isStreaming || this.finished) return;
+      timer = setTimeout(() => {
+        timer = null;
+        if (this.streaming() || this.finished) return;
+        if (!this.prior.endsWith('\n')) this.parser.push('\n');
+        this.parser.finish();
+        this.finished = true;
+        this.finalizeTick.update((v) => v + 1);
+      }, FINALIZE_DEBOUNCE_MS);
+      onCleanup(() => {
+        if (timer) {
+          clearTimeout(timer);
+          timer = null;
+        }
+      });
+    });
   }
 
   // Parser instance is rebuilt only when content diverges from the prior
@@ -85,36 +132,27 @@ export class ChatStreamingMdComponent {
   private parser: PartialMarkdownParser = createPartialMarkdownParser();
   private prior = '';
   private finished = false;
+  // Bumped by the debounced finalizer so the `root` computed re-materializes
+  // the now-finished parser tree.
+  private readonly finalizeTick = signal(0);
 
   readonly root = computed<MarkdownDocumentNode | null>(() => {
     const c = this.content();
-    const isStreaming = this.streaming();
+    this.finalizeTick(); // re-materialize after a debounced finalize
     if (c !== this.prior) {
-      if (c.startsWith(this.prior)) {
+      // Re-parse from scratch when the content diverged from the prior prefix,
+      // OR when the parser was already finalized — finish() is terminal, so
+      // pushing further deltas into a finished parser corrupts its state. A
+      // transient `streaming=false` mid-stream that finalized early thus
+      // recovers here: new content rebuilds an open, projecting parser.
+      if (c.startsWith(this.prior) && !this.finished) {
         this.parser.push(c.slice(this.prior.length));
       } else {
-        // Content shrank or diverged — reset.
         this.parser = createPartialMarkdownParser();
         this.finished = false;
         if (c.length > 0) this.parser.push(c);
       }
-      if (!isStreaming && !this.finished) {
-        // @cacheplane/partial-markdown@0.3 does not flush trailing text on
-        // finish() unless the buffer ends with a newline. Plain LLM
-        // responses often omit the trailing newline, which causes the
-        // parser to emit a document with zero children — i.e. the message
-        // renders empty. Push a sentinel newline first to force the open
-        // paragraph closed before we finalize.
-        if (!c.endsWith('\n')) this.parser.push('\n');
-        this.parser.finish();
-        this.finished = true;
-      }
       this.prior = c;
-    } else if (!isStreaming && !this.finished) {
-      // Streaming flipped to false without new content; ensure parser is finalized.
-      if (!this.prior.endsWith('\n')) this.parser.push('\n');
-      this.parser.finish();
-      this.finished = true;
     }
     // Materialize for Angular reactivity: produces a NEW root reference when
     // any descendant subtree changed; same reference when nothing changed

libs/chat/src/lib/streaming/streaming-markdown.table-stream.spec.ts

@@ -0,0 +1,100 @@
+// libs/chat/src/lib/streaming/streaming-markdown.table-stream.spec.ts
+// SPDX-License-Identifier: MIT
+//
+// Regression: a streaming table must render as a <table> as it arrives, not as
+// raw "| a | b |" paragraph text. The bug was that ChatStreamingMdComponent
+// called parser.finish() on every render where [streaming] was false — and
+// finish() reverts an incomplete table (header with no delimiter row yet) to a
+// CommonMark paragraph (raw pipes). Because the [streaming] flag is unreliable
+// (observed false for an entire live stream at cold start), the whole table
+// rendered as raw pipes until the message completed. The fix: do not finalize
+// the parser while content is still growing; finalize only once it settles.
+import { describe, it, expect, beforeEach, vi } from 'vitest';
+import { TestBed } from '@angular/core/testing';
+import { Component, signal } from '@angular/core';
+import { ChatStreamingMdComponent } from './streaming-markdown.component';
+
+@Component({
+  standalone: true,
+  imports: [ChatStreamingMdComponent],
+  template: `<chat-streaming-md [content]="content()" [streaming]="streaming()" />`,
+})
+class HostComponent {
+  content = signal<string>('');
+  streaming = signal<boolean>(true);
+}
+
+describe('ChatStreamingMdComponent — streaming table rendering', () => {
+  let fixture: ReturnType<typeof TestBed.createComponent<HostComponent>>;
+  let host: HostComponent;
+  let el: HTMLElement;
+  beforeEach(() => {
+    TestBed.configureTestingModule({ imports: [HostComponent] });
+    fixture = TestBed.createComponent(HostComponent);
+    host = fixture.componentInstance;
+    el = fixture.nativeElement as HTMLElement;
+  });
+  const grow = (c: string) => { host.content.set(c); fixture.detectChanges(); };
+
+  it('renders a <table> as a table streams in even when [streaming] lags false', () => {
+    // The cold-start race: content is actively growing but streaming is false.
+    host.streaming.set(false);
+    grow('Here is a table:\n\n| Name ');
+    grow('Here is a table:\n\n| Name | Age |'); // header on the open line, no delimiter
+    // Before the fix: finish() reverted this to raw-pipe paragraphs.
+    expect(el.querySelector('table'), 'header should render as a table, not raw pipes').toBeTruthy();
+    const paras = [...el.querySelectorAll('p')].map((p) => p.textContent || '');
+    expect(paras.some((t) => t.includes('| Name | Age |')), 'no raw-pipe paragraph').toBe(false);
+  });
+
+  it('renders a <table> while streaming (flag true), through the delimiter wait', () => {
+    grow('| Name | Age |');
+    expect(el.querySelector('table')).toBeTruthy();
+    grow('| Name | Age |\n'); // header committed, awaiting delimiter
+    expect(el.querySelector('table')).toBeTruthy();
+  });
+
+  it('finalizes the table once the stream settles (streaming -> false)', () => {
+    host.streaming.set(true);
+    grow('| Name | Age |\n| --- | --- |\n| Ada | 36 |\n');
+    expect(el.querySelector('table')).toBeTruthy();
+    host.streaming.set(false); // settle
+    fixture.detectChanges();
+    const table = el.querySelector('table');
+    expect(table).toBeTruthy();
+    expect(el.querySelectorAll('thead th').length).toBe(2);
+    expect(el.querySelectorAll('tbody tr').length).toBe(1);
+  });
+
+  it('renders a complete one-shot (non-streaming) table message', () => {
+    host.streaming.set(false);
+    grow('| Name | Age |\n| --- | --- |\n| Ada | 36 |\n');
+    expect(el.querySelector('table')).toBeTruthy();
+    expect(el.querySelectorAll('thead th').length).toBe(2);
+  });
+
+  it('does not flash raw pipes when [streaming] flaps false mid-stream', () => {
+    vi.useFakeTimers();
+    try {
+      host.streaming.set(true);
+      grow('| Name | Age |'); // streaming header → table
+      expect(el.querySelector('table')).toBeTruthy();
+      // Flap: streaming reads false for a moment with no new content.
+      host.streaming.set(false);
+      fixture.detectChanges();
+      vi.advanceTimersByTime(60); // less than the debounce — must NOT finalize
+      expect(el.querySelector('table'), 'table must survive the flap').toBeTruthy();
+      expect(
+        [...el.querySelectorAll('p')].some((p) => (p.textContent || '').includes('|')),
+        'no raw-pipe paragraph during the flap',
+      ).toBe(false);
+      // Flap recovers: streaming true again + more content arrives.
+      host.streaming.set(true);
+      grow('| Name | Age |\n| --- | --- |\n');
+      vi.advanceTimersByTime(300);
+      expect(el.querySelector('table')).toBeTruthy();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});

libs/chat/src/lib/streaming/streaming-markdown.variants.spec.ts

@@ -1,5 +1,5 @@
 // SPDX-License-Identifier: MIT
-import { describe, it, expect } from 'vitest';
+import { describe, it, expect, vi } from 'vitest';
 import { TestBed } from '@angular/core/testing';
 import { Component, signal } from '@angular/core';
 import { ChatStreamingMdComponent } from './streaming-markdown.component';
@@ -76,14 +76,24 @@ const midStreamRows: MidStreamRow[] = [
 
 describe('ChatStreamingMdComponent — mid-stream input variance', () => {
   it.each(midStreamRows)('$name', (row) => {
-    TestBed.configureTestingModule({ imports: [HostComponent] });
-    const fixture = TestBed.createComponent(HostComponent);
-    fixture.componentInstance.content.set(row.midStream);
-    fixture.componentInstance.streaming.set(true);
-    fixture.detectChanges();
-    fixture.componentInstance.content.set(row.onFinish ?? row.midStream);
-    fixture.componentInstance.streaming.set(false);
-    fixture.detectChanges();
-    expect(normalize(fixture.nativeElement.textContent ?? '')).toBe(row.expectedText);
+    // Finalization is debounced (the component must not finalize on a transient
+    // streaming=false), so an unclosed construct only reverts to its literal
+    // CommonMark form once the debounce elapses. Drive fake timers to settle.
+    vi.useFakeTimers();
+    try {
+      TestBed.configureTestingModule({ imports: [HostComponent] });
+      const fixture = TestBed.createComponent(HostComponent);
+      fixture.componentInstance.content.set(row.midStream);
+      fixture.componentInstance.streaming.set(true);
+      fixture.detectChanges();
+      fixture.componentInstance.content.set(row.onFinish ?? row.midStream);
+      fixture.componentInstance.streaming.set(false);
+      fixture.detectChanges();
+      vi.advanceTimersByTime(800); // elapse the finalize debounce
+      fixture.detectChanges();
+      expect(normalize(fixture.nativeElement.textContent ?? '')).toBe(row.expectedText);
+    } finally {
+      vi.useRealTimers();
+    }
   });
 });