From e559b2b538e44cecac83b7e7bcc7e2c4ed75a4fa Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 12 Jul 2026 11:26:53 -0700 Subject: [PATCH] FN-7853: preserve chat thread during active streaming turns Fix useChat so already-rendered user/assistant messages no longer flicker away while an agent turn is actively streaming. - useChat.ts: during an active streaming turn for the current session, treat stale/empty/cross-session loadMessages responses as append-only against the visible thread instead of replacing it, merging any genuinely new same-session messages in and skipping the session-cache write when the active thread is being preserved. - ChatView.streaming-thread.test.tsx: add coverage asserting the rendered thread stays visible across mid-turn session-update/tool-call/stale-reload churn. - useChat.test.ts: add hook-level regression tests for the append-only/merge/cache-skip behavior during active streaming. - docs/architecture.md, docs/dashboard-guide.md: document the append-only mid-turn thread-stability behavior. - Add changeset (patch) for @runfusion/fusion describing the user-facing fix. Files changed: .../fn-7853-chat-mid-turn-message-stability.md | 7 + docs/architecture.md | 1 + docs/dashboard-guide.md | 1 + .../__tests__/ChatView.streaming-thread.test.tsx | 130 +++++++++++++ .../dashboard/app/hooks/__tests__/useChat.test.ts | 208 +++++++++++++++++++++ packages/dashboard/app/hooks/useChat.ts | 35 +++- 6 files changed, 380 insertions(+), 2 deletions(-) Fusion-Task-Id: FN-7853 Fusion-Task-Lineage: d9909469-082c-4eeb-81fb-b36d1a9e4705 Co-authored-by: Fusion (runfusion.ai) --- ...fn-7853-chat-mid-turn-message-stability.md | 7 + docs/architecture.md | 1 + docs/dashboard-guide.md | 1 + .../ChatView.streaming-thread.test.tsx | 130 +++++++++++ .../app/hooks/__tests__/useChat.test.ts | 208 ++++++++++++++++++ packages/dashboard/app/hooks/useChat.ts | 35 ++- 6 files changed, 380 insertions(+), 2 deletions(-) create mode 100644 .changeset/fn-7853-chat-mid-turn-message-stability.md diff --git a/.changeset/fn-7853-chat-mid-turn-message-stability.md b/.changeset/fn-7853-chat-mid-turn-message-stability.md new file mode 100644 index 0000000000..3aae985235 --- /dev/null +++ b/.changeset/fn-7853-chat-mid-turn-message-stability.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep chat history stable while an agent is mid-turn so prior user and agent messages no longer flicker away. +category: fix +dev: useChat mid-turn message stability — intermediate chat:session:updated / tool-call / streaming events no longer blank or reflow the rendered `messages` thread (FN-7853, sibling of FN-6496/FN-6599 reattach fixes). diff --git a/docs/architecture.md b/docs/architecture.md index 9a1e40cf39..7a578fdff3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -315,6 +315,7 @@ Intentional exclusions from shared snapshots: - When the active session is still generating after reload/reconnect (`isGenerating: true`), `useChat`/`useQuickChat` hydrate the UI from `inFlightGeneration` immediately, seed the shared stream handlers with that same text/thinking/tool-call snapshot, then reconnect `/api/chat/sessions/:id/stream` with `Last-Event-ID = replayFromEventId` so newly replayed deltas append to the restored bubble instead of replacing it or re-appending already-known deltas. - Hooks also auto-reattach if a stale cached session is selected and a later refresh (or session re-fetch) flips `isGenerating` to true with an `inFlightGeneration` snapshot; dedupe is guarded by a last-attached `(sessionId, replayFromEventId)` ref so snapshot checkpoint bumps do not open duplicate SSE streams. - Attach-triggered message loads may commit the persisted transcript when they match the last attached generation even if React has not yet settled the active-session state/ref. Cache misses during that attach path must preserve the already visible thread so prior user/assistant messages remain visible beside the live streaming assistant response. +- During an active main-chat turn, intermediate session snapshots, tool-call events, and stale message-load responses must never blank or reflow the already-rendered prior thread; the visible transcript remains append-only until the authoritative completion/recovery reload. - Chat message submission uses SSE streaming responses from dashboard chat routes. - Direct-chat terminal failures now persist as a distinct assistant message with `metadata.failureInfo` (`summary`, optional `errorClass`, optional `code`, optional `detail`, optional reference metadata) so the chat thread remains the durable primary failure surface after reload/reconnect. - `ChatManager.sendMessage()` preserves any interrupted partial assistant output as its own message, then appends a separate persisted failure bubble instead of overwriting the partial reply. diff --git a/docs/dashboard-guide.md b/docs/dashboard-guide.md index 602c40490f..86d9112832 100644 --- a/docs/dashboard-guide.md +++ b/docs/dashboard-guide.md @@ -513,6 +513,7 @@ Chat view provides project-scoped conversations with agents. - Final assistant messages with no text, tool calls, thinking output, attachments, or failure details render a muted **No message** placeholder instead of a blank bubble. In-progress responses still use the existing **Working…** / **Thinking…** streaming state until the run finishes. - In-progress assistant responses now survive refresh/navigation while generation is still active: Chat restores the last durable in-flight text/thinking/tool state immediately, keeps the prior persisted conversation visible, then resumes streaming from the stored replay point; any new text, thinking, or tool-call updates append to that restored bubble instead of replacing it or starting from an empty "Working…" placeholder. +- While a Chat response is actively streaming, prior user and assistant messages stay visible across session-update snapshots, tool-call churn, and stale message reloads; the thread does not flicker to an empty history mid-turn. - If a regular Chat stream drops with a hidden-tab/browser-suspension error (for example `Load failed`) while the server is still generating, Chat suppresses the false error banner, re-attaches to the in-progress stream using the durable replay state, and reconciles the final assistant reply when generation completes. - If you queue follow-up user messages while the assistant is still streaming, Chat persists them per session, stacks each queued preview above the input box with one shared divider, and restores/sends them one at a time in FIFO order once each active response finishes if you leave and return. - Chat message lists now track near-bottom scroll state: while you are reading older messages, live streaming/new replies do not force-scroll; a **Latest** jump control appears until you return to the tail. diff --git a/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx b/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx index 82be1e20fd..3d712b23a6 100644 --- a/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx +++ b/packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx @@ -55,6 +55,7 @@ vi.mock("../../hooks/useNavigationHistory", async (importOriginal) => { import * as apiModule from "../../api"; import * as projectStorageModule from "../../utils/projectStorage"; +import * as sseBusModule from "../../sse-bus"; import * as useChatRoomsModule from "../../hooks/useChatRooms"; const mockFetchChatSessions = vi.mocked(apiModule.fetchChatSessions); @@ -62,6 +63,7 @@ const mockFetchChatSession = vi.mocked(apiModule.fetchChatSession); const mockFetchChatMessages = vi.mocked(apiModule.fetchChatMessages); const mockAttachChatStream = vi.mocked(apiModule.attachChatStream); const mockGetScopedItem = vi.mocked(projectStorageModule.getScopedItem); +const mockSubscribeSse = vi.mocked(sseBusModule.subscribeSse); const mockUseChatRooms = vi.mocked(useChatRoomsModule.useChatRooms); const defaultRoomsState: UseChatRoomsResult = { @@ -103,16 +105,41 @@ function makeMessage(overrides: Partial & Pick void; + onToolStart: (data: { toolName: string; args?: Record }) => void; + onToolEnd: (data: { toolName: string; isError: boolean; result?: unknown }) => void; +}; + +function createDeferredPromise() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +function cacheMessages(projectId: string, sessionId: string, messages: ChatMessage[]) { + localStorage.setItem( + `kb-dashboard-chat-messages-cache:${projectId}:${sessionId}`, + JSON.stringify({ savedAt: Date.now(), data: messages }), + ); +} + describe("FN-6599 ChatView streaming prior thread", () => { beforeEach(() => { vi.clearAllMocks(); localStorage.clear(); mockUseChatRooms.mockReturnValue(defaultRoomsState); mockGetScopedItem.mockReturnValue(undefined); + mockSubscribeSse.mockReturnValue(() => {}); mockFetchChatSession.mockResolvedValue({ session: makeSession({ id: "session-001", agentId: "agent-001" }) }); mockAttachChatStream.mockReturnValue({ close: vi.fn(), isConnected: () => true }); }); @@ -165,4 +192,107 @@ describe("FN-6599 ChatView streaming prior thread", () => { expect(screen.getByText("Second question")).toBeInTheDocument(); expect(screen.getByText("Second answer")).toBeInTheDocument(); }); + + it.each([ + ["desktop", 1280], + ["mobile", 390], + ])("FN-7853 keeps cached multi-turn prior thread visible across mid-turn churn on %s", async (_label, width) => { + Object.defineProperty(window, "innerWidth", { configurable: true, value: width }); + window.dispatchEvent(new Event("resize")); + const generatingSession = makeSession({ + id: "session-mid-turn-stable", + agentId: "agent-001", + title: "Mid turn stable", + isGenerating: true, + inFlightGeneration: { + status: "generating" as const, + streamingText: "working", + streamingThinking: "thinking", + toolCalls: [], + replayFromEventId: 201, + updatedAt: "2026-04-08T00:00:00.000Z", + }, + }); + const priorThread = [ + makeMessage({ id: "msg-001", sessionId: generatingSession.id, role: "user", content: "First question" }), + makeMessage({ id: "msg-002", sessionId: generatingSession.id, role: "assistant", content: "First answer" }), + makeMessage({ id: "msg-003", sessionId: generatingSession.id, role: "user", content: "Second question" }), + makeMessage({ id: "msg-004", sessionId: generatingSession.id, role: "assistant", content: "Second answer" }), + ]; + const staleFetch = createDeferredPromise<{ messages: ChatMessage[] }>(); + let attachedHandlers: StreamAppendHandlers | undefined; + let subscribeHandler: Record void> = {}; + + cacheMessages("proj-123", generatingSession.id, priorThread); + mockGetScopedItem.mockImplementation((key) => key === "kb-chat-active-session" ? generatingSession.id : undefined); + mockFetchChatSessions.mockResolvedValue({ sessions: [generatingSession] }); + mockFetchChatMessages.mockReturnValue(staleFetch.promise); + mockAttachChatStream.mockImplementation((_sessionId, handlers) => { + attachedHandlers = handlers; + return { close: vi.fn(), isConnected: () => true }; + }); + mockSubscribeSse.mockImplementation((_url, options) => { + subscribeHandler = options?.events as typeof subscribeHandler; + return () => {}; + }); + + await act(async () => { + render(); + }); + + await waitFor(() => { + expect(screen.getByText("working")).toBeInTheDocument(); + expect(screen.getByText("First question")).toBeInTheDocument(); + expect(screen.getByText("First answer")).toBeInTheDocument(); + expect(screen.getByText("Second question")).toBeInTheDocument(); + expect(screen.getByText("Second answer")).toBeInTheDocument(); + }); + + const expectPriorThreadVisible = () => { + expect(screen.getByText("First question")).toBeInTheDocument(); + expect(screen.getByText("First answer")).toBeInTheDocument(); + expect(screen.getByText("Second question")).toBeInTheDocument(); + expect(screen.getByText("Second answer")).toBeInTheDocument(); + }; + + act(() => { + subscribeHandler["chat:session:updated"]?.({ + data: JSON.stringify({ + ...generatingSession, + inFlightGeneration: { ...generatingSession.inFlightGeneration, streamingText: "working harder", replayFromEventId: 202 }, + }), + } as MessageEvent); + }); + expectPriorThreadVisible(); + + act(() => { + attachedHandlers?.onToolStart({ toolName: "read", args: { path: "README.md" } }); + attachedHandlers?.onText(" now"); + attachedHandlers?.onToolEnd({ toolName: "read", isError: false, result: "ok" }); + }); + await act(async () => { + await Promise.resolve(); + }); + expectPriorThreadVisible(); + + act(() => { + subscribeHandler["chat:message:added"]?.({ + data: JSON.stringify(makeMessage({ + id: "msg-005", + sessionId: generatingSession.id, + role: "user", + content: "Follow-up question", + })), + } as MessageEvent); + }); + expectPriorThreadVisible(); + + await act(async () => { + staleFetch.resolve({ messages: [] }); + await staleFetch.promise; + }); + + expectPriorThreadVisible(); + expect(screen.getByText(/working/)).toBeInTheDocument(); + }); }); diff --git a/packages/dashboard/app/hooks/__tests__/useChat.test.ts b/packages/dashboard/app/hooks/__tests__/useChat.test.ts index 7757b12ada..9638210f9d 100644 --- a/packages/dashboard/app/hooks/__tests__/useChat.test.ts +++ b/packages/dashboard/app/hooks/__tests__/useChat.test.ts @@ -104,8 +104,16 @@ type StreamAppendHandlers = { onThinking: (delta: string) => void; onToolStart: (data: { toolName: string; args?: Record }) => void; onToolEnd: (data: { toolName: string; isError: boolean; result?: unknown }) => void; + onDone?: (data: { messageId?: string; message?: ChatMessage; accumulated: { text: string; thinking: string; toolCalls: unknown[]; fallbackInfo?: unknown } }) => void; }; +function cacheMessages(projectId: string, sessionId: string, messages: ChatMessage[]) { + localStorage.setItem( + `${swrCacheModule.SWR_CACHE_KEYS.CHAT_MESSAGES_PREFIX}${projectId}:${sessionId}`, + JSON.stringify({ savedAt: Date.now(), data: messages }), + ); +} + const setDocumentVisibilityState = (state: DocumentVisibilityState) => { Object.defineProperty(document, "visibilityState", { configurable: true, @@ -3513,6 +3521,206 @@ describe("useChat", () => { }); }); + it("FN-7853 keeps cached prior thread stable when stale mid-turn load resolves empty", async () => { + const generatingSession = { + ...makeSession({ id: "session-mid-turn-stable", agentId: "agent-001", title: "Mid turn stable" }), + isGenerating: true, + inFlightGeneration: { + status: "generating" as const, + streamingText: "working", + streamingThinking: "thinking", + toolCalls: [], + replayFromEventId: 201, + updatedAt: "2026-04-08T00:00:00.000Z", + }, + }; + const priorThread = [ + makeMessage({ id: "msg-001", sessionId: generatingSession.id, role: "user", content: "First question" }), + makeMessage({ id: "msg-002", sessionId: generatingSession.id, role: "assistant", content: "First answer" }), + makeMessage({ id: "msg-003", sessionId: generatingSession.id, role: "user", content: "Second question" }), + makeMessage({ id: "msg-004", sessionId: generatingSession.id, role: "assistant", content: "Second answer" }), + ]; + const staleFetch = createDeferredPromise<{ messages: ChatMessage[] }>(); + let attachedHandlers: StreamAppendHandlers | undefined; + + cacheMessages("proj-123", generatingSession.id, priorThread); + mockGetScopedItem.mockImplementation((key) => key === "kb-chat-active-session" ? generatingSession.id : undefined); + mockFetchChatSessions.mockResolvedValueOnce({ sessions: [generatingSession] }); + mockFetchChatMessages.mockReturnValue(staleFetch.promise); + mockAttachChatStream.mockImplementation((_sessionId, handlers) => { + attachedHandlers = handlers; + return { close: vi.fn(), isConnected: () => true }; + }); + + const { result } = renderHook(() => useChat("proj-123")); + + await waitFor(() => { + expect(result.current.isStreaming).toBe(true); + expect(result.current.streamingText).toBe("working"); + expect(result.current.messages.map((message) => message.content)).toEqual([ + "First question", + "First answer", + "Second question", + "Second answer", + ]); + }); + + const expectPriorThreadStable = () => { + const contents = result.current.messages.map((message) => message.content); + expect(contents).toEqual(expect.arrayContaining([ + "First question", + "First answer", + "Second question", + "Second answer", + ])); + }; + + act(() => { + subscribeHandler["chat:session:updated"]?.({ + data: JSON.stringify({ + ...generatingSession, + inFlightGeneration: { ...generatingSession.inFlightGeneration, streamingText: "working harder", replayFromEventId: 202 }, + }), + } as MessageEvent); + }); + expectPriorThreadStable(); + + vi.useFakeTimers(); + act(() => { + attachedHandlers?.onToolStart({ toolName: "read", args: { path: "README.md" } }); + attachedHandlers?.onText(" now"); + attachedHandlers?.onToolEnd({ toolName: "read", isError: false, result: "ok" }); + }); + act(() => { + vi.advanceTimersToNextTimer(); + }); + expectPriorThreadStable(); + + act(() => { + subscribeHandler["chat:message:added"]?.({ + data: JSON.stringify(makeMessage({ + id: "msg-005", + sessionId: generatingSession.id, + role: "user", + content: "Follow-up question", + })), + } as MessageEvent); + }); + expectPriorThreadStable(); + + await act(async () => { + staleFetch.resolve({ messages: [] }); + await staleFetch.promise; + }); + + expectPriorThreadStable(); + expect(result.current.isStreaming).toBe(true); + + vi.useRealTimers(); + }); + + it("FN-7853 keeps prior thread stable during the client's own streaming send turn", async () => { + const session = makeSession({ id: "session-live-send", agentId: "agent-001", title: "Live send" }); + const priorThreadNewestFirst = [ + makeMessage({ id: "msg-004", sessionId: session.id, role: "assistant", content: "Second answer" }), + makeMessage({ id: "msg-003", sessionId: session.id, role: "user", content: "Second question" }), + makeMessage({ id: "msg-002", sessionId: session.id, role: "assistant", content: "First answer" }), + makeMessage({ id: "msg-001", sessionId: session.id, role: "user", content: "First question" }), + ]; + let streamHandlers: StreamAppendHandlers | undefined; + + mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] }); + mockFetchChatSession.mockResolvedValue({ session }); + mockFetchChatMessages.mockResolvedValue({ messages: priorThreadNewestFirst }); + mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => { + streamHandlers = handlers; + 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.id); + }); + + await waitFor(() => { + expect(result.current.messages.map((message) => message.content)).toEqual([ + "First question", + "First answer", + "Second question", + "Second answer", + ]); + }); + + act(() => { + result.current.sendMessage("Third question"); + }); + + expect(result.current.messages.map((message) => message.content)).toEqual([ + "First question", + "First answer", + "Second question", + "Second answer", + "Third question", + ]); + expect(result.current.isStreaming).toBe(true); + + vi.useFakeTimers(); + act(() => { + streamHandlers?.onText("Partial answer"); + subscribeHandler["chat:session:updated"]?.({ + data: JSON.stringify({ + ...session, + isGenerating: true, + inFlightGeneration: { + status: "generating" as const, + streamingText: "Partial answer", + streamingThinking: "", + toolCalls: [], + replayFromEventId: 301, + updatedAt: "2026-04-08T00:00:00.000Z", + }, + }), + } as MessageEvent); + subscribeHandler["chat:message:added"]?.({ + data: JSON.stringify(makeMessage({ id: "msg-ignored-assistant", sessionId: session.id, role: "assistant", content: "Duplicate assistant" })), + } as MessageEvent); + }); + act(() => { + vi.advanceTimersToNextTimer(); + }); + + expect(result.current.messages.map((message) => message.content)).toEqual([ + "First question", + "First answer", + "Second question", + "Second answer", + "Third question", + ]); + expect(result.current.streamingText).toBe("Partial answer"); + + act(() => { + streamHandlers?.onDone?.({ + messageId: "msg-final-live-send", + accumulated: { text: "Final live answer", thinking: "", toolCalls: [] }, + }); + }); + + expect(result.current.messages.map((message) => message.content)).toEqual([ + "First question", + "First answer", + "Second question", + "Second answer", + "Third question", + "Partial answer", + ]); + vi.useRealTimers(); + }); + it("FN-6496 loads prior thread during chat:session:updated in-flight attach", async () => { const session = makeSession({ id: "session-001", agentId: "agent-001", title: "Existing" }); const priorThreadNewestFirst = [ diff --git a/packages/dashboard/app/hooks/useChat.ts b/packages/dashboard/app/hooks/useChat.ts index 86d524f499..b588aade5e 100644 --- a/packages/dashboard/app/hooks/useChat.ts +++ b/packages/dashboard/app/hooks/useChat.ts @@ -548,9 +548,40 @@ export function useChat( } } else { if (shouldCommitMessages) { - setMessages(mappedMessages); + const isActiveStreamingSession = isStreamingRef.current && activeSessionRef.current?.id === sessionId; + const responseBelongsToSession = mappedMessages.every((message) => message.sessionId === sessionId); + const currentMessagesBelongToSession = messagesRef.current.length > 0 && messagesRef.current.every((message) => message.sessionId === sessionId); + const shouldPreserveActiveStreamingThread = isActiveStreamingSession + && currentMessagesBelongToSession + && (!responseBelongsToSession || mappedMessages.length === 0); + setMessages((prev) => { + if (isActiveStreamingSession && prev.length > 0) { + const previousBelongsToSession = prev.every((message) => message.sessionId === sessionId); + if (previousBelongsToSession && (!responseBelongsToSession || mappedMessages.length === 0)) { + /* + FNXC:ChatStreaming 2026-07-12-11:08: + During an active assistant turn, the visible prior thread is append-only until onDone/recovery performs the authoritative reload. Mid-turn chat:session:updated, tool-call, and streaming churn can leave an older loadMessages request in flight; an empty or cross-session response must not blank/reflow messages because chat:message:added assistant echoes are suppressed while streaming. + */ + return prev; + } + + if (previousBelongsToSession && mappedMessages.length > 0) { + const merged = [...prev]; + const seen = new Set(prev.map((message) => message.id)); + for (const message of mappedMessages) { + if (!seen.has(message.id)) { + merged.push(message); + seen.add(message.id); + } + } + return merged; + } + } + + return mappedMessages; + }); setHasMoreMessages(data.messages.length >= 50); - if (cacheKey) writeCache(cacheKey, mappedMessages, { maxBytes: 500_000 }); + if (cacheKey && responseBelongsToSession && !shouldPreserveActiveStreamingThread) writeCache(cacheKey, mappedMessages, { maxBytes: 500_000 }); } } } catch {