Merge pull request #1387 from Runfusion/gsxdsm/queued-messages

fix: confirm server generation state before flushing queued chat messages

.changeset/fn-5852-queued-message-stale-flush.md

@@ -0,0 +1,9 @@
+---
+"@runfusion/fusion": patch
+---
+
+Stop queued chat messages from disappearing after back-navigation while the assistant is still responding (GitHub #1279).
+
+Re-entering a chat restored the queued follow-up and immediately flushed it based on the client's local `isGenerating` flag — which is stale mid-generation (it is a route-level enrichment the `chat:session:updated` SSE payload lacks). The premature send aborted the live generation server-side and could lose the queued message entirely, since its persisted copy was deleted before the send.
+
+The restore path in both Chat and Quick Chat now confirms with the server before flushing: if a generation is still in flight it re-attaches to the stream and lets completion deliver the queued message; the message is sent immediately only when the server reports no active generation. On a failed check the queued bubble is kept for a later flush trigger.

CONCEPTS.md

@@ -88,6 +88,18 @@ The post-merge step that rebases locally-landed merge commits onto the upstream
 
 ### Contamination
 Foreign commits — work attributed to other Tasks — appearing on a Task's branch beyond its recorded Fork point. Contamination checks must compute their reference base fresh from the Integration branch rather than reuse the Task's stored base, since a stale stored base makes every legitimately merged commit look foreign.
+
+## Chat
+
+### Generation
+A single in-flight assistant turn for a chat session, owned server-side and identified by a generation id. At most one Generation runs per session: starting a new one aborts whatever Generation is still active for that session, so an "extra" send from a client is never harmless. A Generation periodically persists an in-flight snapshot so a reconnecting client can recover the streaming UI.
+
+### Queued message
+A follow-up the user sends while a Generation is active. It is held client-side (and persisted per session so navigation cannot lose it) and flushed — actually sent — only when the session's Generation settles. Flushing decisions must be made against the server's authoritative generation state, not a locally cached copy.
+
+### Enrichment field
+A session field computed at the API route from live server state (whether a Generation is running, last-message preview) rather than stored on the session row. Enrichment fields exist only in responses from enriching endpoints: store-event payloads and SSE broadcasts lack them, so a client that overwrites its session state from those sources silently degrades enrichment fields to absent — any side-effecting decision gated on one must re-fetch from an enriching endpoint.
+
 ## Compound Engineering sessions
 
 ### CE Stage

docs/solutions/logic-errors/queued-chat-message-flush-trusts-stale-isgenerating.md

@@ -0,0 +1,68 @@
+---
+title: Queued chat message flush trusted stale client-side isGenerating
+date: 2026-06-03
+category: logic-errors
+module: dashboard
+problem_type: logic_error
+component: frontend_stimulus
+symptoms:
+  - "Queued follow-up message vanishes after back-navigating out of a regular chat and re-entering — not sent, not in the composer (FN-5852, GitHub #1279)"
+  - "Re-entering a chat could abort the assistant's in-flight reply mid-stream"
+  - "Bug survived two prior fixes (FN-5852, FN-5921) that made the queued draft persist to localStorage"
+root_cause: async_timing
+resolution_type: code_fix
+severity: high
+related_components:
+  - chat-store
+  - sse
+tags:
+  - chat
+  - queued-messages
+  - sse
+  - stale-state
+  - isgenerating
+  - enrichment-fields
+  - useChat
+  - useQuickChat
+---
+
+# Queued chat message flush trusted stale client-side isGenerating
+
+## Problem
+
+Messages queued while the assistant was responding disappeared if the user hit back and re-entered the chat before they were sent. Two prior fixes made the queued draft *persist*, but the restore path still flushed it immediately based on local state, firing a send that aborted the live generation server-side and deleted the persisted copy before the send could fail.
+
+## Symptoms
+
+- Queued bubble gone after back → re-enter; message never sent, persisted localStorage copy deleted
+- Original assistant reply could be killed mid-stream on re-entry (server `beginGeneration` aborts the prior generation on any new send)
+- Hook-level tests for the exact navigation flow passed while production failed
+
+## What Didn't Work
+
+- **FN-5852** — persisting the queued draft per session in localStorage. Necessary but insufficient: the restore path still flushed from stale state.
+- **FN-5921** — removing the eager `removePersistedPendingChatMessage` calls from `resetTransientComposerState`/`selectSession`. Also necessary, also insufficient — same restore-path flaw.
+- Both rounds of tests hand-crafted `isGenerating: true` in the client's sessions list, a state production never has mid-generation, so the suite green-lit a broken flow twice.
+
+## Solution
+
+The restore effect in `useChat.ts` / `useQuickChat.ts` no longer flushes from local state. It restores the queued bubble, then calls `fetchChatSession` (authoritative, route-enriched) and decides:
+
+- server says generating → `attachIfGenerating(...)` and let the stream's `onDone`/`onError` flush
+- server says idle → flush now
+- fetch failed → keep the bubble; a later trigger (stream completion, visibility resume, manual send) delivers it
+
+## Why This Works
+
+`isGenerating` is **not a stored field** — it's a route-level enrichment computed from `ChatManager`'s in-memory generation map (`register-chat-routes.ts`). The `chat:session:updated` SSE event emits the raw `ChatSession` store row (`chat-store.ts` → `sse.ts`), which has no `isGenerating`, and the client handler replaces `sessions[]`/`activeSession` **wholesale** with that payload. So mid-generation, the client's local copy reliably reports `isGenerating: undefined`. Any client logic that gates a side-effecting action on the locally cached flag acts on fiction; only a fresh `GET /chat/sessions/:id` reflects reality.
+
+## Prevention
+
+- **Treat enrichment fields as expired the moment they arrive via any path that doesn't enrich.** Before gating a destructive/irreversible action (sending, deleting, aborting) on a cached server-state flag, re-fetch from the authoritative endpoint.
+- **The SSE wholesale-replace is a standing trap**: `handleChatSessionUpdated` overwrites enriched session objects with un-enriched ones, silently degrading `isGenerating` and `lastMessagePreview` for every consumer. Merging instead of replacing (or enriching the SSE payload server-side) would eliminate this class — flagged as a follow-up, not yet done.
+- **When a regression test passes but production fails, audit the fixture state against what production can actually contain.** Both prior test rounds modeled an unreachable state; the fixed regression tests model the stale-falsy-flag + server-generating combination.
+- Remember `beginGeneration` aborts any in-flight generation for the session — an "extra" client send is never harmless.
+
+## Related Issues
+
+- Runfusion/Fusion#1279 / FN-5852 / FN-5921; fixed in PR Runfusion/Fusion#1387

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

@@ -2088,6 +2088,102 @@ describe("useChat", () => {
     });
   });
 
+  it("does not flush a restored queued message while the server still reports an in-flight generation", async () => {
+    // Reproduces FN-5852 back-navigation loss: the sessions-list entry has a
+    // stale falsy isGenerating (it is a route-level enrichment that the
+    // chat:session:updated SSE payload lacks), while the server is actually
+    // still generating. The restored queued message must NOT be flushed from
+    // local state alone — doing so aborts the live generation server-side.
+    const sessionA = makeSession({ id: "session-001", agentId: "agent-001" });
+    mockFetchChatSessions.mockResolvedValue({ sessions: [sessionA] });
+    mockFetchChatMessages.mockResolvedValue({ messages: [] });
+    mockFetchChatSession.mockResolvedValue({
+      session: {
+        ...sessionA,
+        isGenerating: true,
+        inFlightGeneration: {
+          streamingText: "partial",
+          streamingThinking: "",
+          toolCalls: [],
+        },
+      },
+    });
+
+    const attachHandlers: Array<Parameters<typeof mockAttachChatStream>[1]> = [];
+    mockAttachChatStream.mockImplementation((_sessionId, nextHandlers) => {
+      attachHandlers.push(nextHandlers);
+      return { close: vi.fn(), isConnected: () => true };
+    });
+
+    localStorage.setItem(getChatPendingMessageKey("session-001")!, "Queued follow-up");
+
+    const { result } = renderHook(() => useChat("proj-123"));
+
+    await waitFor(() => {
+      expect(result.current.sessions).toHaveLength(1);
+    });
+
+    act(() => {
+      result.current.selectSession("session-001");
+    });
+
+    // The queued message is restored and the authoritative session fetch
+    // reveals the in-flight generation, so the hook attaches instead of
+    // flushing.
+    await waitFor(() => {
+      expect(result.current.pendingMessage).toBe("Queued follow-up");
+      expect(result.current.isStreaming).toBe(true);
+    });
+
+    expect(mockStreamChatResponse).not.toHaveBeenCalled();
+    expect(localStorage.getItem(getChatPendingMessageKey("session-001"))).toBe("Queued follow-up");
+
+    // Once the attached generation completes, the queued message flushes.
+    act(() => {
+      attachHandlers[0]?.onDone?.({ messageId: "msg-001" });
+    });
+
+    await waitFor(() => {
+      expect(mockStreamChatResponse).toHaveBeenCalledTimes(1);
+      expect(mockStreamChatResponse.mock.calls[0]?.[0]).toBe("session-001");
+      expect(mockStreamChatResponse.mock.calls[0]?.[1]).toBe("Queued follow-up");
+      expect(result.current.pendingMessage).toBe("");
+      expect(localStorage.getItem(getChatPendingMessageKey("session-001"))).toBeNull();
+    });
+  });
+
+  it("keeps a restored queued message un-flushed while the server validation fetch is pending", async () => {
+    // Production latency case: the authoritative fetch takes one network
+    // RTT. Nothing may flush (or delete) the restored queue in the interim.
+    const sessionA = makeSession({ id: "session-001", agentId: "agent-001" });
+    mockFetchChatSessions.mockResolvedValue({ sessions: [sessionA] });
+    mockFetchChatMessages.mockResolvedValue({ messages: [] });
+    // Server check never resolves within the test — simulates in-flight RTT.
+    mockFetchChatSession.mockReturnValue(new Promise(() => {}) as never);
+
+    localStorage.setItem(getChatPendingMessageKey("session-001")!, "Queued follow-up");
+
+    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.pendingMessage).toBe("Queued follow-up");
+      expect(mockStreamChatResponse).not.toHaveBeenCalled();
+      expect(localStorage.getItem(getChatPendingMessageKey("session-001"))).toBe("Queued follow-up");
+    });
+
+    expect(result.current.pendingMessage).toBe("Queued follow-up");
+    expect(mockStreamChatResponse).not.toHaveBeenCalled();
+    expect(localStorage.getItem(getChatPendingMessageKey("session-001"))).toBe("Queued follow-up");
+  });
+
   it("preserves queued messages across session switches and rehydrates them when returning", async () => {
     const sessionA = {
       ...makeSession({ id: "session-001", agentId: "agent-001" }),

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

@@ -972,6 +972,94 @@ describe("useQuickChat", () => {
     });
   });
 
+  it("does not flush a restored queued message while the server still reports an in-flight generation", async () => {
+    // Mirrors the useChat FN-5852 regression: the locally-held session has a
+    // stale falsy isGenerating, but the server is still generating. The
+    // restored queued message must wait for the authoritative fetch instead
+    // of flushing immediately (which would abort the live generation).
+    const staleSessionA = makeSession({ id: "session-a", agentId: "agent-001" });
+    mockFetchChatMessages.mockResolvedValue({ messages: [] });
+    mockFetchChatSession.mockResolvedValue({
+      session: {
+        ...staleSessionA,
+        isGenerating: true,
+        inFlightGeneration: {
+          streamingText: "partial",
+          streamingThinking: "",
+          toolCalls: [],
+        },
+      },
+    });
+
+    const attachHandlers: Array<Parameters<typeof mockAttachChatStream>[1]> = [];
+    mockAttachChatStream.mockImplementation((_sessionId, nextHandlers) => {
+      attachHandlers.push(nextHandlers);
+      return { close: vi.fn(), isConnected: () => true };
+    });
+
+    localStorage.setItem(getChatPendingMessageKey("session-a")!, "Queued follow-up");
+
+    const { result } = renderHook(() => useQuickChat("proj-123"));
+
+    await act(async () => {
+      await result.current.selectSession(staleSessionA);
+    });
+
+    await waitFor(() => {
+      expect(result.current.pendingMessage).toBe("Queued follow-up");
+      expect(result.current.isStreaming).toBe(true);
+    });
+
+    expect(mockStreamChatResponse).not.toHaveBeenCalled();
+    expect(localStorage.getItem(getChatPendingMessageKey("session-a"))).toBe("Queued follow-up");
+
+    // The hook attached to the in-flight generation rather than flushing.
+    expect(mockAttachChatStream).toHaveBeenCalledTimes(1);
+    expect(attachHandlers.length).toBeGreaterThan(0);
+
+    // Once the attached generation completes, the queued message flushes.
+    act(() => {
+      attachHandlers[0]?.onDone?.({ messageId: "msg-001" });
+    });
+
+    await waitFor(() => {
+      expect(mockStreamChatResponse).toHaveBeenCalledTimes(1);
+      expect(mockStreamChatResponse.mock.calls[0]?.[0]).toBe("session-a");
+      expect(mockStreamChatResponse.mock.calls[0]?.[1]).toBe("Queued follow-up");
+      expect(result.current.pendingMessage).toBe("");
+      expect(localStorage.getItem(getChatPendingMessageKey("session-a"))).toBeNull();
+    });
+  });
+
+  it("does not let the session-activation auto-flush send a restored queue while server validation is pending", async () => {
+    // The restore effect's fetchChatSession check takes one network RTT in
+    // production. The session-activation auto-flush effect runs in the same
+    // commit that restores pendingMessageRef, so without the pre-session
+    // gate it would send the restored queue before the check resolves and
+    // re-open the stale-isGenerating loss path (FN-5852).
+    const staleSessionA = makeSession({ id: "session-a", agentId: "agent-001" });
+    mockFetchChatMessages.mockResolvedValue({ messages: [] });
+    // Server check never resolves within the test — simulates in-flight RTT.
+    mockFetchChatSession.mockReturnValue(new Promise(() => {}) as never);
+
+    localStorage.setItem(getChatPendingMessageKey("session-a")!, "Queued follow-up");
+
+    const { result } = renderHook(() => useQuickChat("proj-123"));
+
+    await act(async () => {
+      await result.current.selectSession(staleSessionA);
+    });
+
+    await act(async () => {
+      await new Promise((resolve) => setTimeout(resolve, 25));
+    });
+
+    // The restored queue is intact and nothing was sent.
+    expect(result.current.pendingMessage).toBe("Queued follow-up");
+    expect(mockStreamChatResponse).not.toHaveBeenCalled();
+    expect(localStorage.getItem(getChatPendingMessageKey("session-a"))).toBe("Queued follow-up");
+  });
+
   it("pre-session queueing does not write a null localStorage key", async () => {
     const session = makeSession({ id: "session-pre", agentId: "agent-001" });
     mockFetchResumeChatSession.mockResolvedValueOnce({ session });

packages/dashboard/app/hooks/useChat.ts

@@ -683,17 +683,44 @@ export function useChat(
     pendingMessageRef.current = restoredPendingMessage;
     setPendingMessage(restoredPendingMessage);
 
-    queueMicrotask(() => {
-      if (
-        activeSessionRef.current?.id === sessionId &&
-        pendingMessageRef.current.trim().length > 0 &&
-        !isStreamingRef.current &&
-        !streamRef.current
-      ) {
-        flushPendingMessage();
-      }
-    });
-  }, [activeSession?.id, flushPendingMessage]);
+    // Flush only once the server confirms no generation is in flight. The
+    // local sessions list can hold a stale falsy `isGenerating` (it is a
+    // route-level enrichment that the chat:session:updated SSE payload
+    // lacks), so flushing from local state alone fires a send that aborts a
+    // live generation server-side and can lose the queued message (FN-5852).
+    let cancelled = false;
+    void fetchChatSession(sessionId, projectId)
+      .then(({ session: refreshedSession }) => {
+        if (
+          cancelled ||
+          activeSessionRef.current?.id !== sessionId ||
+          pendingMessageRef.current.trim().length === 0
+        ) {
+          return;
+        }
+
+        if (refreshedSession.isGenerating) {
+          // Still generating: attach (if not already) and let the stream's
+          // onDone/onError flush the queued message.
+          if (!streamRef.current) {
+            attachIfGenerating(sessionId, refreshedSession.inFlightGeneration);
+          }
+          return;
+        }
+
+        if (!isStreamingRef.current && !streamRef.current) {
+          flushPendingMessage();
+        }
+      })
+      .catch(() => {
+        // Keep the restored bubble; another flush trigger (stream
+        // completion, visibility resume, manual send) will deliver it.
+      });
+
+    return () => {
+      cancelled = true;
+    };
+  }, [activeSession?.id, attachIfGenerating, flushPendingMessage, projectId]);
 
   // Create a new session
   const createSession = useCallback(

packages/dashboard/app/hooks/useQuickChat.ts

@@ -656,17 +656,44 @@ export function useQuickChat(
     pendingMessageRef.current = restoredPendingMessage;
     setPendingMessage(restoredPendingMessage);
 
-    queueMicrotask(() => {
-      if (
-        activeSessionRef.current?.id === sessionId &&
-        pendingMessageRef.current.trim().length > 0 &&
-        !isStreamingRef.current &&
-        !streamRef.current
-      ) {
-        void flushPendingMessage();
-      }
-    });
-  }, [activeSession?.id, flushPendingMessage]);
+    // Flush only once the server confirms no generation is in flight. The
+    // local session snapshot can hold a stale falsy `isGenerating` (it is a
+    // route-level enrichment that the chat:session:updated SSE payload
+    // lacks), so flushing from local state alone fires a send that aborts a
+    // live generation server-side and can lose the queued message (FN-5852).
+    let cancelled = false;
+    void fetchChatSession(sessionId, projectId)
+      .then(({ session: refreshedSession }) => {
+        if (
+          cancelled ||
+          activeSessionRef.current?.id !== sessionId ||
+          pendingMessageRef.current.trim().length === 0
+        ) {
+          return;
+        }
+
+        if (refreshedSession.isGenerating) {
+          // Still generating: attach (if not already) and let the stream's
+          // onDone/onError flush the queued message.
+          if (!streamRef.current) {
+            attachIfGenerating(sessionId, refreshedSession.inFlightGeneration);
+          }
+          return;
+        }
+
+        if (!isStreamingRef.current && !streamRef.current) {
+          void flushPendingMessage();
+        }
+      })
+      .catch(() => {
+        // Keep the restored bubble; another flush trigger (stream
+        // completion, visibility resume, manual send) will deliver it.
+      });
+
+    return () => {
+      cancelled = true;
+    };
+  }, [activeSession?.id, attachIfGenerating, flushPendingMessage, projectId]);
 
   const startModelChat = useCallback(
     async (modelProvider: string, modelId: string) => {
@@ -1027,6 +1054,15 @@ export function useQuickChat(
       return;
     }
 
+    // Only the pre-session queue (a send issued before session init
+    // completed) may auto-flush on session activation. Restored queued
+    // messages must wait for the restore effect's authoritative
+    // fetchChatSession check — flushing them here would race ahead of it
+    // and re-open the stale-isGenerating loss path (FN-5852).
+    if (!queuedPreSessionCompletionRef.current) {
+      return;
+    }
+
     if (pendingMessageRef.current.trim().length === 0) {
       return;
     }