Fix streaming error handling, attempt to fix performance issues, and add tests

Author: elias-ba

CHANGELOG.md

@@ -23,6 +23,12 @@ and this project adheres to
   [#3701](https://github.com/OpenFn/lightning/issues/3701)
 - Delete nodes from Job panel in Collaborative Editor
   [#3702](https://github.com/OpenFn/lightning/issues/3702)
+- **AI Assistant Streaming**: AI responses now stream in real-time with status updates
+  - Users see AI responses appear word-by-word as they're generated
+  - Status indicators show thinking progress (e.g., "Researching...", "Generating code...")
+  - Automatic error recovery with retry/cancel options
+  - Configurable timeout based on Apollo settings
+  [#3585](https://github.com/OpenFn/lightning/issues/3585)
 
 ### Changed
 
@@ -38,6 +44,13 @@ and this project adheres to
   unauthorized edits when user roles change during active collaboration sessions
   [#3749](https://github.com/OpenFn/lightning/issues/3749)
 
+### Technical
+
+- Added `Lightning.ApolloClient.SSEStream` for Server-Sent Events handling
+- Enhanced `MessageProcessor` to support streaming responses
+- Updated AI Assistant component with real-time markdown rendering
+- Improved error handling for network failures and timeouts
+
 ## [2.14.11] - 2025-10-15
 
 ## [2.14.11-pre1] - 2025-10-15

assets/js/hooks/index.ts

@@ -4,6 +4,7 @@ import tippy, {
 } from 'tippy.js';
 import { format, formatRelative } from 'date-fns';
 import { enUS } from 'date-fns/locale';
+import { marked } from 'marked';
 import type { PhoenixHook } from './PhoenixHook';
 
 import LogLineHighlight from './LogLineHighlight';
@@ -684,9 +685,39 @@ export const BlurDataclipEditor = {
 
 export const ScrollToMessage = {
   mounted() {
+    this.shouldAutoScroll = true;
+
+    // Throttle scroll tracking to reduce CPU usage
+    this.handleScrollThrottled = this.throttle(() => {
+      const isAtBottom = this.isAtBottom();
+      this.shouldAutoScroll = isAtBottom;
+    }, 100); // Only check every 100ms
+
+    this.el.addEventListener('scroll', this.handleScrollThrottled);
     this.handleScroll();
   },
 
+  destroyed() {
+    if (this.handleScrollThrottled) {
+      this.el.removeEventListener('scroll', this.handleScrollThrottled);
+    }
+    if (this.throttleTimeout !== undefined) {
+      clearTimeout(this.throttleTimeout);
+    }
+  },
+
+  throttle(func: () => void, wait: number): () => void {
+    return () => {
+      if (this.throttleTimeout !== undefined) {
+        clearTimeout(this.throttleTimeout);
+      }
+      this.throttleTimeout = setTimeout(() => {
+        func();
+        this.throttleTimeout = undefined;
+      }, wait) as unknown as number;
+    };
+  },
+
   updated() {
     this.handleScroll();
   },
@@ -696,7 +727,8 @@ export const ScrollToMessage = {
 
     if (targetMessageId) {
       this.scrollToSpecificMessage(targetMessageId);
-    } else {
+    } else if (this.shouldAutoScroll) {
+      // Only auto-scroll if user hasn't manually scrolled up
       this.scrollToBottom();
     }
   },
@@ -717,18 +749,26 @@ export const ScrollToMessage = {
     }
   },
 
+  isAtBottom() {
+    const threshold = 50; // pixels from bottom
+    const position = this.el.scrollTop + this.el.clientHeight;
+    const height = this.el.scrollHeight;
+    return height - position <= threshold;
+  },
+
   scrollToBottom() {
-    setTimeout(() => {
-      this.el.scrollTo({
-        top: this.el.scrollHeight,
-        behavior: 'smooth',
-      });
-    }, 600);
+    // Use instant scroll during updates to prevent jank
+    this.el.scrollTop = this.el.scrollHeight;
   },
 } as PhoenixHook<{
+  shouldAutoScroll: boolean;
+  handleScrollThrottled?: () => void;
+  throttleTimeout?: number;
+  throttle: (func: () => void, wait: number) => () => void;
   handleScroll: () => void;
   scrollToSpecificMessage: (messageId: string) => void;
   scrollToBottom: () => void;
+  isAtBottom: () => boolean;
 }>;
 
 export const Copy = {
@@ -1020,3 +1060,98 @@ export const LocalTimeConverter = {
   convertDateTime: () => void;
   convertToDisplayTime: (isoTimestamp: string, display: string) => void;
 }>;
+
+export const StreamingText = {
+  mounted() {
+    this.lastContent = '';
+    this.renderer = this.createCustomRenderer();
+    this.parseCount = 0;
+    this.pendingUpdate = undefined;
+    this.updateContent();
+  },
+
+  updated() {
+    // Debounce updates by 50ms to batch rapid chunk arrivals
+    if (this.pendingUpdate !== undefined) {
+      clearTimeout(this.pendingUpdate);
+    }
+
+    this.pendingUpdate = setTimeout(() => {
+      this.updateContent();
+      this.pendingUpdate = undefined;
+    }, 50) as unknown as number;
+  },
+
+  destroyed() {
+    if (this.pendingUpdate !== undefined) {
+      clearTimeout(this.pendingUpdate);
+    }
+  },
+
+  createCustomRenderer() {
+    const renderer = new marked.Renderer();
+
+    // Apply custom CSS classes to match backend Earmark styles
+    renderer.code = (code, language) => {
+      const lang = language ? ` class="${language}"` : '';
+      return `<pre class="rounded-md font-mono bg-slate-100 border-2 border-slate-200 text-slate-800 my-4 p-2 overflow-auto"><code${lang}>${code}</code></pre>`;
+    };
+
+    renderer.link = (href, title, text) => {
+      return `<a href="${href}" class="text-primary-400 hover:text-primary-600" target="_blank">${text}</a>`;
+    };
+
+    renderer.heading = (text, level) => {
+      const classes = level === 1 ? 'text-2xl font-bold mb-6' : 'text-xl font-semibold mb-4 mt-8';
+      return `<h${level} class="${classes}">${text}</h${level}>`;
+    };
+
+    renderer.list = (body, ordered) => {
+      const tag = ordered ? 'ol' : 'ul';
+      const classes = ordered ? 'list-decimal pl-8 space-y-1' : 'list-disc pl-8 space-y-1';
+      return `<${tag} class="${classes}">${body}</${tag}>`;
+    };
+
+    renderer.listitem = (text) => {
+      return `<li class="text-gray-800">${text}</li>`;
+    };
+
+    renderer.paragraph = (text) => {
+      return `<p class="mt-1 mb-2 text-gray-800">${text}</p>`;
+    };
+
+    return renderer;
+  },
+
+  updateContent() {
+    const start = performance.now();
+    const newContent = this.el.dataset.streamingContent || '';
+
+    if (newContent !== this.lastContent) {
+      this.parseCount++;
+
+      // Re-parse entire content as markdown
+      // This handles split ticks because we always parse the full accumulated string
+      const htmlContent = marked.parse(newContent, {
+        renderer: this.renderer,
+        breaks: true,
+        gfm: true,
+      });
+
+      this.el.innerHTML = htmlContent;
+      this.lastContent = newContent;
+
+      const duration = performance.now() - start;
+      console.debug(
+        `[StreamingText] Parse #${this.parseCount}: ${duration.toFixed(2)}ms for ${newContent.length} chars`
+      );
+    }
+  },
+} as PhoenixHook<{
+  lastContent: string;
+  renderer: marked.Renderer;
+  parseCount: number;
+  pendingUpdate?: number;
+  createCustomRenderer: () => marked.Renderer;
+  updateContent: () => void;
+}>;

lib/lightning/ai_assistant/ai_assistant.ex

@@ -569,12 +569,16 @@ defmodule Lightning.AiAssistant do
 
   ## Returns
 
-  List of `%ChatMessage{}` structs with `:role` of `:user` and `:status` of `:pending`.
+  List of `%ChatMessage{}` structs with `:role` of `:user` and `:status` of `:pending` or `:processing`.
   """
   @spec find_pending_user_messages(ChatSession.t()) :: [ChatMessage.t()]
   def find_pending_user_messages(session) do
     messages = session.messages || []
-    Enum.filter(messages, &(&1.role == :user && &1.status == :pending))
+
+    Enum.filter(
+      messages,
+      &(&1.role == :user && &1.status in [:pending, :processing])
+    )
   end
 
   @doc """