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

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

Browse Source 
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
This commit is contained in:
Fusion
2026-05-13 13:46:21 -07:00
committed by gsxdsm
parent 039e9a2d02
commit c50e152a87
5 changed files with 211 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/FN-4381-flush-queued-chat-on-recovery.md Normal file
View File

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

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

@@ -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] });

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

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

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

@@ -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();
if (queuedMessage) {
pendingMessageRef.current = "";
setPendingMessage("");
sendMessageRef.current(queuedMessage);
}
flushPendingMessage();
},
onError: (data, tempUserMessageId) => {
const failureInfo = normalizeFailureInfo(data);
@@ -807,19 +815,14 @@ export function useChat(
}
if (!cancelledByUserRef.current) {
const queuedMessage = pendingMessageRef.current.trim();
if (queuedMessage) {
pendingMessageRef.current = "";
setPendingMessage("");
sendMessageRef.current(queuedMessage);
}
flushPendingMessage();
}
},
});
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(() => {

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

@@ -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();
if (queuedMessage) {
pendingMessageRef.current = "";
setPendingMessage("");
void sendMessageRef.current(queuedMessage);
}
flushPendingMessage();
},
onError: (data) => {
setStreamingText("");
@@ -784,12 +794,7 @@ export function useQuickChat(
sendCompletionRef.current = null;
if (!cancelledByUserRef.current) {
const queuedMessage = pendingMessageRef.current.trim();
if (queuedMessage) {
pendingMessageRef.current = "";
setPendingMessage("");
void sendMessageRef.current(queuedMessage);
}
flushPendingMessage();
}
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(() => {