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

feat(FN-3432): document chat streaming recovery guarantees in architecture

Browse Source 
Added documentation for chat streaming recovery guarantees to the architecture docs.

Fusion-Task-Id: FN-3432
This commit is contained in:
Fusion
2026-05-05 00:06:56 -07:00
committed by gsxdsm
parent 91151305c3
commit cdb64e008d
5 changed files with 180 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
packages/dashboard/app/components/ChatView.tsx
View File

@@ -1465,6 +1465,7 @@ export function ChatView({ projectId, addToast }: ChatViewProps) {
};
const activeModelTag = formatModelTag(activeSession?.modelProvider, activeSession?.modelId);
const hasThreadInView = Boolean(activeSession || isStreaming || messages.length > 0);
const threadHeaderTitle = activeSession?.agentId === FN_AGENT_ID
? (activeModelTag ?? "Fusion")
@@ -1666,9 +1667,9 @@ export function ChatView({ projectId, addToast }: ChatViewProps) {
{/* Thread */}
<div className="chat-thread" style={threadKeyboardStyle}>
{/* Header - always rendered in desktop/tablet, only rendered in mobile when viewing a thread */}
{(activeSession || !isMobile) && (
{(hasThreadInView || !isMobile) && (
<div className="chat-thread-header">
{isMobile && activeSession && (
{isMobile && hasThreadInView && (
<button className="btn-icon" onClick={handleBack} data-testid="chat-back-btn">
<ChevronLeft size={16} />
</button>

36
packages/dashboard/app/components/__tests__/ChatView.test.tsx
View File

@@ -1512,6 +1512,42 @@ describe("ChatView", () => {
});
describe("streaming states", () => {
it("keeps mobile thread visible when active session metadata refreshes during streaming", () => {
const mediaQuerySpy = mockViewportMode("mobile");
const streamingState: UseChatReturn = {
...defaultChatState,
sessions: [{ ...activeSessionFixture }],
filteredSessions: [{ ...activeSessionFixture }],
activeSession: { ...activeSessionFixture },
messages: [],
isStreaming: true,
streamingText: "",
streamingThinking: "",
};
const refreshedStreamingState: UseChatReturn = {
...streamingState,
sessions: [{ ...activeSessionFixture, updatedAt: "2026-04-08T00:05:00.000Z" }],
filteredSessions: [{ ...activeSessionFixture, updatedAt: "2026-04-08T00:05:00.000Z" }],
activeSession: null,
};
mockUseChat
.mockReturnValueOnce(streamingState)
.mockReturnValue(refreshedStreamingState);
const { rerender } = render(<ChatView projectId="proj-123" addToast={vi.fn()} />);
expect(document.querySelector(".chat-message--streaming")?.textContent).toContain("Connecting");
rerender(<ChatView projectId="proj-123" addToast={vi.fn()} />);
expect(document.querySelector(".chat-message--streaming")?.textContent).toContain("Connecting");
expect(screen.queryByText("Start a new conversation")).not.toBeInTheDocument();
expect(screen.queryByText("No messages yet. Start the conversation!")).not.toBeInTheDocument();
expect(screen.getByTestId("chat-back-btn")).toBeInTheDocument();
void mediaQuerySpy;
});
it("keeps the streaming indicator visible while message history is still loading", () => {
setupMockChat({
activeSession: { id: "session-001", agentId: "agent-001", status: "active", title: "Test Chat", createdAt: "2026-04-08T00:00:00.000Z", updatedAt: "2026-04-08T00:00:00.000Z" },

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

@@ -1485,6 +1485,88 @@ describe("useChat", () => {
});
describe("FN-3336: streaming state recovery on reload", () => {
it("does not re-select and reset active session on subsequent session refreshes", async () => {
const session = { ...makeSession({ id: "session-001", agentId: "agent-001" }), isGenerating: true };
mockGetScopedItem.mockReturnValue("session-001");
mockFetchChatSessions
.mockResolvedValueOnce({ sessions: [session] })
.mockResolvedValueOnce({ sessions: [{ ...session, updatedAt: "2026-04-08T00:05:00.000Z" }] });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
const { result } = renderHook(() => useChat("proj-123"));
await waitFor(() => {
expect(result.current.activeSession?.id).toBe("session-001");
});
expect(mockFetchChatMessages).toHaveBeenCalledTimes(1);
await act(async () => {
await result.current.refreshSessions();
});
await waitFor(() => {
expect(result.current.isStreaming).toBe(true);
});
// A sessions refresh should not auto-reselect/reset the active thread.
expect(mockFetchChatMessages).toHaveBeenCalledTimes(1);
});
it("preserves streaming text/thinking/tool state across sessions refresh", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchChatSessions
.mockResolvedValueOnce({ sessions: [session] })
.mockResolvedValueOnce({ sessions: [{ ...session, updatedAt: "2026-04-08T00:06:00.000Z" }] });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
let textHandler: ((data: string) => void) | undefined;
let thinkingHandler: ((data: string) => void) | undefined;
let toolStartHandler: ((data: { toolName: string; args?: Record<string, unknown> }) => void) | undefined;
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
textHandler = handlers.onText;
thinkingHandler = handlers.onThinking;
toolStartHandler = handlers.onToolStart;
return { close: vi.fn(), isConnected: () => true };
});
const { result } = renderHook(() => useChat("proj-123"));
await waitFor(() => {
expect(result.current.sessions).toHaveLength(1);
});
act(() => {
result.current.selectSession("session-001");
});
await act(async () => {
result.current.sendMessage("Hello");
});
await act(async () => {
textHandler?.("Hi");
thinkingHandler?.("plan");
toolStartHandler?.({ toolName: "read", args: { path: "a.ts" } });
});
await waitFor(() => {
expect(result.current.isStreaming).toBe(true);
expect(result.current.streamingText).toBe("Hi");
expect(result.current.streamingThinking).toBe("plan");
expect(result.current.streamingToolCalls).toHaveLength(1);
});
await act(async () => {
await result.current.refreshSessions();
});
expect(result.current.isStreaming).toBe(true);
expect(result.current.streamingText).toBe("Hi");
expect(result.current.streamingThinking).toBe("plan");
expect(result.current.streamingToolCalls).toHaveLength(1);
});
it("sets isStreaming=true when selecting a session with isGenerating=true", async () => {
const session = { ...makeSession({ id: "session-001", agentId: "agent-001" }), isGenerating: true };
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });

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

@@ -290,22 +290,35 @@ export function useChat(
refreshSessions();
}, [refreshSessions]);
// Restore active session from localStorage after initial load
// Uses a ref to avoid circular dependency with selectSession
// Restore active session from localStorage after initial load.
// Uses refs to avoid circular dependency with selectSession and to avoid
// re-selecting/resetting the thread on every sessions refresh.
const selectSessionRef = useRef<(id: string, sessionOverride?: ChatSessionInfo) => void>(() => {
/* noop - will be replaced after selectSession is defined */
});
const hasRestoredActiveSessionRef = useRef(false);
useEffect(() => {
if (sessionsLoading) return; // Wait for sessions to load
hasRestoredActiveSessionRef.current = false;
}, [projectId]);
useEffect(() => {
if (sessionsLoading || hasRestoredActiveSessionRef.current || activeSessionRef.current) return;
const savedSessionId = getScopedItem(ACTIVE_SESSION_STORAGE_KEY, projectId);
if (savedSessionId) {
// Check if the saved session exists in the loaded sessions
const session = sessions.find((s) => s.id === savedSessionId);
if (session) {
selectSessionRef.current(savedSessionId);
}
if (!savedSessionId) {
hasRestoredActiveSessionRef.current = true;
return;
}
const session = sessions.find((s) => s.id === savedSessionId);
if (session) {
hasRestoredActiveSessionRef.current = true;
selectSessionRef.current(savedSessionId, session);
return;
}
hasRestoredActiveSessionRef.current = true;
}, [sessionsLoading, sessions, projectId]);
// Load messages when active session changes
@@ -345,6 +358,11 @@ export function useChat(
// Select a session
const selectSession = useCallback(
(id: string, sessionOverride?: ChatSessionInfo) => {
const currentActiveSessionId = activeSessionRef.current?.id ?? null;
if (id && currentActiveSessionId === id && !sessionOverride) {
return;
}
// Close any existing stream
if (streamRef.current) {
streamRef.current.close();
@@ -708,6 +726,36 @@ export function useChat(
)
: sessions;
// Recovery mode polling: if reloaded mid-generation, keep waiting state alive
// until generation finishes and messages can be reloaded.
useEffect(() => {
if (!isStreaming || streamRef.current || !activeSessionRef.current) return;
const interval = setInterval(async () => {
if (!isStreamingRef.current || streamRef.current || !activeSessionRef.current) {
clearInterval(interval);
return;
}
try {
const data: ChatSessionListResponse = await fetchChatSessions(projectId);
const session = data.sessions.find((candidate) => candidate.id === activeSessionRef.current?.id);
if (!session?.isGenerating) {
clearInterval(interval);
await loadMessages(activeSessionRef.current.id);
setStreamingText("");
setStreamingThinking("");
setStreamingToolCalls([]);
setIsStreaming(false);
}
} catch {
// Silently fail - will retry next interval
}
}, 3000);
return () => clearInterval(interval);
}, [isStreaming, loadMessages, projectId]);
// SSE real-time updates
useEffect(() => {
const contextVersionAtStart = projectContextVersionRef.current;