feat(FN-4381): flush queued chat messages on recovery completion

Fixes chat hooks to flush queued messages when recovery completes, preventing stale messages from lingering in useChat and useQuickChat; includes tests for both hooks.

Fusion-Task-Id: FN-4381

.changeset/FN-4381-flush-queued-chat-on-recovery.md

@@ -0,0 +1,5 @@
+---
+"@runfusion/fusion": patch
+---
+
+Fix chat queued follow-up delivery so pending messages auto-send when streaming completes through recovery paths (SSE message-added recovery, polling finalization, and visibility-resume), not only fresh-send onDone/onError handlers.

packages/dashboard/app/hooks/__tests__/useChat.test.ts

@@ -1317,6 +1317,108 @@ describe("useChat", () => {
     });
   });
 
+  describe("queued message recovery paths", () => {
+    it("flushes queued message when recovery completes via chat:message:added SSE", async () => {
+      const session = {
+        ...makeSession({ id: "session-001", agentId: "agent-001" }),
+        isGenerating: true,
+      };
+      mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
+      mockFetchChatMessages.mockResolvedValue({ messages: [] });
+      mockAttachChatStream.mockReturnValue(null as never);
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      act(() => {
+        result.current.selectSession("session-001");
+      });
+
+      await waitFor(() => {
+        expect(result.current.isStreaming).toBe(true);
+      });
+
+      act(() => {
+        result.current.sendMessage("Queued follow-up");
+      });
+
+      await waitFor(() => {
+        expect(result.current.pendingMessage).toBe("Queued follow-up");
+      });
+
+      const subscribeOptions = mockSubscribeSse.mock.calls.at(-1)?.[1];
+      const messageAdded = subscribeOptions?.events?.["chat:message:added"];
+      expect(messageAdded).toBeTypeOf("function");
+
+      act(() => {
+        messageAdded?.({
+          data: JSON.stringify(makeMessage({
+            id: "msg-002",
+            sessionId: "session-001",
+            role: "assistant",
+            content: "Recovered",
+          })),
+        } as MessageEvent);
+      });
+
+      await waitFor(() => {
+        expect(mockStreamChatResponse).toHaveBeenCalledTimes(1);
+        expect(mockStreamChatResponse.mock.calls[0]?.[1]).toBe("Queued follow-up");
+        expect(result.current.pendingMessage).toBe("");
+      });
+    });
+
+    it("flushes queued message when visibility resume sees generation complete", async () => {
+      const session = {
+        ...makeSession({ id: "session-001", agentId: "agent-001" }),
+        isGenerating: true,
+      };
+      mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
+      mockFetchChatMessages.mockResolvedValue({ messages: [] });
+      mockAttachChatStream.mockReturnValue(null as never);
+      mockFetchChatSession.mockResolvedValue({
+        session: { ...session, isGenerating: false },
+      });
+
+      const { result } = renderHook(() => useChat("proj-123"));
+
+      await waitFor(() => {
+        expect(result.current.sessions).toHaveLength(1);
+      });
+
+      act(() => {
+        result.current.selectSession("session-001");
+      });
+
+      await waitFor(() => {
+        expect(result.current.isStreaming).toBe(true);
+      });
+
+      act(() => {
+        result.current.sendMessage("Queued follow-up");
+      });
+
+      await waitFor(() => {
+        expect(result.current.pendingMessage).toBe("Queued follow-up");
+      });
+
+      act(() => {
+        setDocumentVisibilityState("hidden");
+        setDocumentVisibilityState("visible");
+      });
+
+      await waitFor(() => {
+        expect(mockFetchChatSession).toHaveBeenCalledWith("session-001", "proj-123");
+        expect(mockStreamChatResponse).toHaveBeenCalledTimes(1);
+        expect(mockStreamChatResponse.mock.calls[0]?.[1]).toBe("Queued follow-up");
+        expect(result.current.pendingMessage).toBe("");
+      });
+    });
+  });
+
   it("queued message is not auto-sent after user-initiated stop", async () => {
     const session = makeSession({ id: "session-001", agentId: "agent-001" });
     mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });

packages/dashboard/app/hooks/__tests__/useQuickChat.test.ts

@@ -739,6 +739,52 @@ describe("useQuickChat", () => {
     });
   });
 
+  describe("queued message recovery paths", () => {
+    it("flushes queued message when attached recovery stream completes", async () => {
+      const existingSession = {
+        ...makeSession({ id: "session-existing", agentId: "agent-001" }),
+        isGenerating: true,
+      };
+      const attachHandlers: Array<Parameters<typeof mockAttachChatStream>[1]> = [];
+
+      mockFetchResumeChatSession.mockResolvedValueOnce({ session: existingSession });
+      mockFetchChatMessages.mockResolvedValue({ messages: [] });
+      mockAttachChatStream.mockImplementation((_sessionId, handlers) => {
+        attachHandlers.push(handlers);
+        return { close: vi.fn(), isConnected: () => true };
+      });
+
+      const { result } = renderHook(() => useQuickChat("proj-123"));
+
+      await act(async () => {
+        await result.current.switchSession("agent-001");
+      });
+
+      await waitFor(() => {
+        expect(result.current.isStreaming).toBe(true);
+      });
+
+      const queuedSend = result.current.sendMessage("Queued follow-up");
+
+      await waitFor(() => {
+        expect(result.current.pendingMessage).toBe("Queued follow-up");
+      });
+
+      act(() => {
+        attachHandlers[0]?.onDone?.({ messageId: "msg-recovery" });
+      });
+
+      await waitFor(() => {
+        expect(mockStreamChatResponse).toHaveBeenCalledTimes(1);
+        expect(mockStreamChatResponse.mock.calls[0]?.[1]).toBe("Queued follow-up");
+        expect(result.current.pendingMessage).toBe("");
+      });
+
+      await expect(queuedSend).resolves.toBeUndefined();
+    });
+
+  });
+
   it("onError does not remove user message from local state", async () => {
     const existingSession = makeSession({ id: "session-existing", agentId: "agent-001" });
     let onErrorHandler: ((data: string) => void) | undefined;

packages/dashboard/app/hooks/useChat.ts

@@ -406,6 +406,22 @@ export function useChat(
     setIsStreaming(false);
   }, []);
 
+  const clearPendingMessage = useCallback(() => {
+    pendingMessageRef.current = "";
+    setPendingMessage("");
+  }, []);
+
+  const flushPendingMessage = useCallback(() => {
+    const queuedMessage = pendingMessageRef.current.trim();
+    if (!queuedMessage) {
+      return;
+    }
+
+    pendingMessageRef.current = "";
+    setPendingMessage("");
+    sendMessageRef.current(queuedMessage);
+  }, []);
+
   const attachIfGenerating = useCallback((
     sessionId: string,
     inFlightGeneration?: ChatInFlightGenerationState | null,
@@ -446,6 +462,7 @@ export function useChat(
         isStreamingRef.current = false;
         streamRef.current = null;
         void loadMessages(sessionId);
+        flushPendingMessage();
       },
       onError: (data) => {
         setStreamingText("");
@@ -459,6 +476,7 @@ export function useChat(
           addToast?.(failureInfo.summary, "error");
         }
         void loadMessages(sessionId);
+        flushPendingMessage();
       },
     });
 
@@ -469,7 +487,7 @@ export function useChat(
     });
     streamRef.current = stream;
     return true;
-  }, [addToast, loadMessages, projectId]);
+  }, [addToast, loadMessages, projectId, flushPendingMessage]);
 
   // Select a session
   const selectSession = useCallback(
@@ -617,11 +635,6 @@ export function useChat(
     setStreamingToolCalls([]);
   }, [activeSession, projectId]);
 
-  const clearPendingMessage = useCallback(() => {
-    pendingMessageRef.current = "";
-    setPendingMessage("");
-  }, []);
-
   /**
    * Send a user message to the active chat session.
    * @param content Message text content to send.
@@ -754,12 +767,7 @@ export function useChat(
 
           refreshSessions();
 
-          const queuedMessage = pendingMessageRef.current.trim();
+          flushPendingMessage();
-          if (queuedMessage) {
-            pendingMessageRef.current = "";
-            setPendingMessage("");
-            sendMessageRef.current(queuedMessage);
-          }
         },
         onError: (data, tempUserMessageId) => {
           const failureInfo = normalizeFailureInfo(data);
@@ -807,19 +815,14 @@ export function useChat(
           }
 
           if (!cancelledByUserRef.current) {
-            const queuedMessage = pendingMessageRef.current.trim();
+            flushPendingMessage();
-            if (queuedMessage) {
-              pendingMessageRef.current = "";
-              setPendingMessage("");
-              sendMessageRef.current(queuedMessage);
-            }
           }
         },
       });
 
       streamRef.current = streamChatResponse(activeSession.id, content, handlers, attachments, projectId);
     },
-    [activeSession, projectId, refreshSessions, addToast, attachIfGenerating, reconnectSessionSilently],
+    [activeSession, projectId, refreshSessions, addToast, attachIfGenerating, reconnectSessionSilently, flushPendingMessage],
   );
 
   sendMessageRef.current = sendMessage;
@@ -860,6 +863,8 @@ export function useChat(
           setStreamingThinking("");
           setStreamingToolCalls([]);
           setIsStreaming(false);
+          isStreamingRef.current = false;
+          flushPendingMessage();
         }
       } catch {
         // Silently fail - will retry next interval
@@ -867,7 +872,7 @@ export function useChat(
     }, 3000);
 
     return () => clearInterval(interval);
-  }, [attachIfGenerating, loadMessages, projectId, activeSession]);
+  }, [attachIfGenerating, loadMessages, projectId, activeSession, flushPendingMessage]);
 
   useEffect(() => {
     const unsubscribe = visibilitySuspension.onBecameVisible(() => {
@@ -899,6 +904,7 @@ export function useChat(
             setStreamingToolCalls([]);
             setIsStreaming(false);
             isStreamingRef.current = false;
+            flushPendingMessage();
             void loadMessages(currentSession.id);
           }
         })
@@ -908,7 +914,7 @@ export function useChat(
     });
 
     return unsubscribe;
-  }, [attachIfGenerating, loadMessages, projectId, visibilitySuspension]);
+  }, [attachIfGenerating, loadMessages, projectId, visibilitySuspension, flushPendingMessage]);
 
   // SSE real-time updates
   useEffect(() => {
@@ -983,6 +989,8 @@ export function useChat(
         setStreamingThinking("");
         setStreamingToolCalls([]);
         setIsStreaming(false);
+        isStreamingRef.current = false;
+        flushPendingMessage();
         return;
       }
 
@@ -1031,7 +1039,7 @@ export function useChat(
     });
 
     return unsubscribe;
-  }, [attachIfGenerating, projectId]);
+  }, [attachIfGenerating, projectId, flushPendingMessage]);
 
   // Cleanup on unmount
   useEffect(() => {

packages/dashboard/app/hooks/useQuickChat.ts

@@ -269,6 +269,22 @@ export function useQuickChat(
     [projectId],
   );
 
+  const clearPendingMessage = useCallback(() => {
+    pendingMessageRef.current = "";
+    setPendingMessage("");
+  }, []);
+
+  const flushPendingMessage = useCallback(() => {
+    const queuedMessage = pendingMessageRef.current.trim();
+    if (!queuedMessage) {
+      return;
+    }
+
+    pendingMessageRef.current = "";
+    setPendingMessage("");
+    void sendMessageRef.current(queuedMessage);
+  }, []);
+
   const attachIfGenerating = useCallback((
     sessionId: string,
     inFlightGeneration?: ChatInFlightGenerationState | null,
@@ -311,6 +327,7 @@ export function useQuickChat(
         void fetchChatMessages(sessionId, { limit: 50 }, projectId).then((data) => {
           setMessages(data.messages.map(mapChatMessageToInfo));
         }).catch(() => {});
+        flushPendingMessage();
       },
       onError: (data) => {
         setStreamingText("");
@@ -326,6 +343,7 @@ export function useQuickChat(
         void fetchChatMessages(sessionId, { limit: 50 }, projectId).then((resp) => {
           setMessages(resp.messages.map(mapChatMessageToInfo));
         }).catch(() => {});
+        flushPendingMessage();
       },
     });
 
@@ -335,7 +353,7 @@ export function useQuickChat(
         : {}),
     });
     return true;
-  }, [addToast, projectId]);
+  }, [addToast, projectId, flushPendingMessage]);
 
   // Fetch existing sessions and find/create one for the given target
   const initializeSession = useCallback(
@@ -446,6 +464,8 @@ export function useQuickChat(
           setStreamingThinking("");
           setStreamingToolCalls([]);
           setIsStreaming(false);
+          isStreamingRef.current = false;
+          flushPendingMessage();
         }
       } catch {
         // Silently fail - will retry on next interval
@@ -453,7 +473,7 @@ export function useQuickChat(
     }, 3000);
 
     return () => clearInterval(interval);
-  }, [activeSession, attachIfGenerating, projectId]);
+  }, [activeSession, attachIfGenerating, projectId, flushPendingMessage]);
 
   // Reload messages from server (for same-session revisit)
   const reloadMessages = useCallback(async () => {
@@ -611,11 +631,6 @@ export function useQuickChat(
     setStreamingToolCalls([]);
   }, [activeSession, projectId]);
 
-  const clearPendingMessage = useCallback(() => {
-    pendingMessageRef.current = "";
-    setPendingMessage("");
-  }, []);
-
   const sendMessageRef = useRef<(content: string, attachments?: File[]) => Promise<void>>(() => Promise.resolve());
   const visibilitySuspension = useTabVisibilitySuspension();
 
@@ -747,12 +762,7 @@ export function useQuickChat(
             sendCompletionRef.current?.resolve();
             sendCompletionRef.current = null;
 
-            const queuedMessage = pendingMessageRef.current.trim();
+            flushPendingMessage();
-            if (queuedMessage) {
-              pendingMessageRef.current = "";
-              setPendingMessage("");
-              void sendMessageRef.current(queuedMessage);
-            }
           },
           onError: (data) => {
             setStreamingText("");
@@ -784,12 +794,7 @@ export function useQuickChat(
             sendCompletionRef.current = null;
 
             if (!cancelledByUserRef.current) {
-              const queuedMessage = pendingMessageRef.current.trim();
+              flushPendingMessage();
-              if (queuedMessage) {
-                pendingMessageRef.current = "";
-                setPendingMessage("");
-                void sendMessageRef.current(queuedMessage);
-              }
             }
 
             if (!shouldSuppressSuspensionError) {
@@ -806,7 +811,7 @@ export function useQuickChat(
       void completionPromise.catch(() => {});
       return completionPromise;
     },
-    [activeSession, projectId, addToast, reloadMessages, reconnectSessionSilently],
+    [activeSession, projectId, addToast, reloadMessages, reconnectSessionSilently, flushPendingMessage],
   );
 
   sendMessageRef.current = sendMessage;
@@ -821,7 +826,7 @@ export function useQuickChat(
         return;
       }
 
-      void fetchChatSession(currentSession.id, projectId)
+      void Promise.resolve(fetchChatSession(currentSession.id, projectId))
         .then((data) => {
           if (streamRef.current || activeSessionRef.current?.id !== currentSession.id) {
             return;
@@ -843,6 +848,7 @@ export function useQuickChat(
             setStreamingToolCalls([]);
             setIsStreaming(false);
             isStreamingRef.current = false;
+            flushPendingMessage();
             void reloadMessages();
           }
         })
@@ -852,7 +858,7 @@ export function useQuickChat(
     });
 
     return unsubscribe;
-  }, [attachIfGenerating, projectId, reloadMessages, visibilitySuspension]);
+  }, [attachIfGenerating, projectId, reloadMessages, visibilitySuspension, flushPendingMessage]);
 
   // Cleanup on unmount
   useEffect(() => {