root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(FN-5852): confirm server generation state before flushing queued chat messages

Browse Source 
Re-entering a chat flushed the restored queued message based on the
client's stale isGenerating flag (a route-level enrichment the
chat:session:updated SSE payload lacks), firing a send that aborted the
live generation server-side and could lose the message entirely. The
restore path in useChat and useQuickChat now asks the server first:
attach and defer the flush while generating, send immediately only when
no generation is in flight, and keep the bubble on a failed check.

Fixes Runfusion/Fusion#1279

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-06-03 19:54:02 -07:00
parent 07dd24e37a
commit 684baa0eb7
5 changed files with 204 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
.changeset/fn-5852-queued-message-stale-flush.md Normal file
View File

@@ -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.

64
packages/dashboard/app/hooks/__tests__/useChat.test.ts
View File

@@ -2088,6 +2088,70 @@ 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("preserves queued messages across session switches and rehydrates them when returning", async () => {
const sessionA = {
...makeSession({ id: "session-001", agentId: "agent-001" }),

55
packages/dashboard/app/hooks/__tests__/useQuickChat.test.ts
View File

@@ -972,6 +972,61 @@ 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");
// 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("pre-session queueing does not write a null localStorage key", async () => {
const session = makeSession({ id: "session-pre", agentId: "agent-001" });
mockFetchResumeChatSession.mockResolvedValueOnce({ session });

49
packages/dashboard/app/hooks/useChat.ts
View File

@@ -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(

49
packages/dashboard/app/hooks/useQuickChat.ts
View File

@@ -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 Promise.resolve(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) => {