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

FN-7853: preserve chat thread during active streaming turns

Browse Source 
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) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-07-12 11:26:53 -07:00
parent 20c6db9534
commit e559b2b538
6 changed files with 380 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fn-7853-chat-mid-turn-message-stability.md Normal file
View File

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

1
docs/architecture.md
View File

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

1
docs/dashboard-guide.md
View File

@@ -513,6 +513,7 @@ Chat view provides project-scoped conversations with agents.
<!-- FNXC:ChatEmptyMessage 2026-07-10-00:00: Empty final assistant responses can be legitimate provider output (for example a Grok CLI run ending without text). Document the shared Chat/Planner Chat behavior so operators see "No message" instead of interpreting a blank bubble as a rendering failure. -->
- 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.

130
packages/dashboard/app/components/__tests__/ChatView.streaming-thread.test.tsx
View File

@@ -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<ChatMessage> & Pick<ChatMessage, "id" |
content: overrides.content,
thinkingOutput: overrides.thinkingOutput ?? null,
metadata: overrides.metadata ?? null,
attachments: overrides.attachments,
createdAt: overrides.createdAt ?? "2026-04-08T00:00:00.000Z",
};
}
type StreamAppendHandlers = {
onText: (delta: string) => void;
onToolStart: (data: { toolName: string; args?: Record<string, unknown> }) => void;
onToolEnd: (data: { toolName: string; isError: boolean; result?: unknown }) => void;
};
function createDeferredPromise<T>() {
let resolve!: (value: T | PromiseLike<T>) => void;
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((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<string, (event: MessageEvent) => 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(<ChatView projectId="proj-123" addToast={vi.fn()} />);
});
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();
});
});

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

@@ -104,8 +104,16 @@ type StreamAppendHandlers = {
onThinking: (delta: string) => void;
onToolStart: (data: { toolName: string; args?: Record<string, unknown> }) => 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 = [

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

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