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

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/[email protected] 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();
+    }
   });
 });