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

FN-6632: preserve reattached chat stream chunks

Browse Source 
Seed streaming accumulators from durable in-flight snapshots so reattached chat bubbles keep prior chunks.

- Add initial text, thinking, and tool-call snapshots to chat stream handlers.
- Wire main chat and QuickChat reattach flows to seed handler accumulators before replayed deltas arrive.
- Cover text, thinking, and tool-call continuation across shared handler, main chat, and QuickChat tests.
- Document the reattach accumulator invariant and add a published package changeset.

Files changed:
 .changeset/fn-6632-chat-stream-reattach.md         |   5 +
 docs/architecture.md                               |   2 +-
 docs/dashboard-guide.md                            |   2 +-
 .../__tests__/createChatStreamHandlers.test.ts     |  63 +++++++
 .../dashboard/app/hooks/__tests__/useChat.test.ts  | 182 +++++++++++++++++++++
 .../app/hooks/__tests__/useQuickChat.test.ts       | 147 +++++++++++++++++
 .../app/hooks/createChatStreamHandlers.ts          |  19 ++-
 packages/dashboard/app/hooks/useChat.ts            |   7 +
 packages/dashboard/app/hooks/useQuickChat.ts       |   7 +
 9 files changed, 429 insertions(+), 5 deletions(-)

Fusion-Task-Id: FN-6632
Fusion-Task-Lineage: e2f2eb42-08aa-4bae-a79f-60cfc9fb03d4
This commit is contained in:
gsxdsm
2026-06-18 06:36:03 -07:00
parent dae0bde6f1
commit 3d28b3b212
9 changed files with 429 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/fn-6632-chat-stream-reattach.md Normal file
View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Preserve already-streamed chat text, thinking, and tool-call state when the dashboard reattaches to an in-flight assistant response.

2
docs/architecture.md
View File

@@ -312,7 +312,7 @@ Intentional exclusions from shared snapshots:
- Main `useChat` session restore/recovery must not reset the active thread during session-list refresh or `chat:session:updated` metadata churn while a response is in flight.
- `chat_sessions.inFlightGeneration` stores a durable JSON snapshot while generation is active: latest streamed text/thinking, tool-call state, and `replayFromEventId` for SSE resume.
- `ChatManager.sendMessage()` updates that snapshot during streaming (debounced) and clears it on done/error/cancel so stale partial state does not survive completion.
- When the active session is still generating after reload/reconnect (`isGenerating: true`), `useChat`/`useQuickChat` hydrate the UI from `inFlightGeneration` immediately, then reconnect `/api/chat/sessions/:id/stream` with `Last-Event-ID = replayFromEventId` to avoid re-appending already-known deltas.
- 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.
- Chat message submission uses SSE streaming responses from dashboard chat routes.

2
docs/dashboard-guide.md
View File

@@ -238,7 +238,7 @@ Chat view provides project-scoped conversations with agents.
- Entering `/new` or `/clear` (exact match after trimming) in the composer starts a fresh thread for the current chat target instead of sending the literal command to the model
- On mobile, the New Chat and Delete Conversation dialogs use a compact inset treatment (centered, viewport-bounded, internally scrollable) instead of the app's default full-height mobile modal chrome.
- Full Chat and Quick Chat both consume the same streamed `/api/chat/sessions/:id/messages` response contract, and both now prefer the authoritative assistant `message` snapshot on `done` while still accumulating `text` chunks when present (so providers without incremental text streaming still render output immediately)
- 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 instead of starting from an empty "Connecting…" placeholder.
- 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 "Connecting…" placeholder.
- 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 a follow-up user message while the assistant is still streaming, Chat now persists that queued text per session so leaving and returning to the view still restores and sends it once the active response finishes.
- 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.

63
packages/dashboard/app/hooks/__tests__/createChatStreamHandlers.test.ts
View File

@@ -1,5 +1,6 @@
import { describe, expect, it, vi } from "vitest";
import { createChatStreamHandlers } from "../createChatStreamHandlers";
import type { ToolCallInfo } from "../chatTypes";
describe("createChatStreamHandlers", () => {
it.each([
@@ -62,4 +63,66 @@ describe("createChatStreamHandlers", () => {
vi.useRealTimers();
});
it("FN-6632 seeds reattached accumulators before appending new chunks", () => {
vi.useFakeTimers();
let text = "Hello ";
let thinking = "thinking…";
let toolCalls: ToolCallInfo[] = [
{ toolName: "read", status: "completed", isError: false, result: "seeded" },
];
const onDone = vi.fn();
const cancelStreamingFlushesRef = { current: null } as { current: (() => void) | null };
const { handlers } = createChatStreamHandlers({
sessionId: "s-1",
tempUserMessageId: "",
initialText: "Hello ",
initialThinking: "thinking…",
initialToolCalls: toolCalls,
setStreamingText: (value) => {
text = typeof value === "function" ? value(text) : value;
},
setStreamingThinking: (value) => {
thinking = typeof value === "function" ? value(thinking) : value;
},
setStreamingToolCalls: (value) => {
toolCalls = typeof value === "function" ? value(toolCalls) : value;
},
cancelStreamingFlushesRef,
onDone,
onError: vi.fn(),
});
handlers.onText("world");
handlers.onText("!");
handlers.onThinking(" more");
handlers.onToolStart({ toolName: "write", args: { path: "a.ts" } });
handlers.onToolEnd({ toolName: "write", isError: false, result: "done" });
vi.advanceTimersToNextTimer();
vi.advanceTimersToNextTimer();
expect(text).toBe("Hello world!");
expect(thinking).toBe("thinking… more");
expect(toolCalls).toEqual([
{ toolName: "read", status: "completed", isError: false, result: "seeded" },
{ toolName: "write", args: { path: "a.ts" }, status: "completed", isError: false, result: "done" },
]);
handlers.onDone({ messageId: "m-1" });
expect(onDone).toHaveBeenCalledWith({
messageId: "m-1",
message: undefined,
accumulated: {
text: "Hello world!",
thinking: "thinking… more",
toolCalls,
fallbackInfo: undefined,
},
});
vi.useRealTimers();
});
});

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

@@ -95,6 +95,13 @@ function createDeferredPromise<T>() {
return { promise, resolve, reject };
}
type StreamAppendHandlers = {
onText: (delta: string) => void;
onThinking: (delta: string) => void;
onToolStart: (data: { toolName: string; args?: Record<string, unknown> }) => void;
onToolEnd: (data: { toolName: string; isError: boolean; result?: unknown }) => void;
};
const setDocumentVisibilityState = (state: DocumentVisibilityState) => {
Object.defineProperty(document, "visibilityState", {
configurable: true,
@@ -3091,6 +3098,71 @@ describe("useChat", () => {
});
});
it("FN-6632 preserves prior streamed chunks during chat:session:updated reattach", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001", title: "Existing" });
const generatingSession = {
...session,
isGenerating: true,
inFlightGeneration: {
status: "generating" as const,
streamingText: "Hello ",
streamingThinking: "plan ",
toolCalls: [{ toolName: "read", status: "running" as const, isError: false }],
replayFromEventId: 5,
updatedAt: "2026-04-08T00:00:00.000Z",
},
};
let attachedHandlers: StreamAppendHandlers | undefined;
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
mockAttachChatStream.mockImplementation((_sessionId, handlers) => {
attachedHandlers = 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);
});
act(() => {
subscribeHandler["chat:session:updated"]?.({
data: JSON.stringify(generatingSession),
} as MessageEvent);
});
await waitFor(() => {
expect(result.current.isStreaming).toBe(true);
expect(result.current.streamingText).toBe("Hello ");
expect(attachedHandlers).toBeDefined();
});
vi.useFakeTimers();
act(() => {
attachedHandlers?.onText("world");
attachedHandlers?.onText("!");
attachedHandlers?.onThinking("more");
attachedHandlers?.onToolEnd({ toolName: "read", isError: false, result: "done" });
});
act(() => {
vi.advanceTimersToNextTimer();
vi.advanceTimersToNextTimer();
});
expect(result.current.isStreaming).toBe(true);
expect(result.current.streamingText).toBe("Hello world!");
expect(result.current.streamingThinking).toBe("plan more");
expect(result.current.streamingToolCalls).toEqual([
{ toolName: "read", status: "completed", isError: false, result: "done" },
]);
vi.useRealTimers();
});
it("FN-6496 loads prior thread when auto-reattach effect observes refreshed generation", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001", title: "Stale" });
const priorThreadNewestFirst = [
@@ -3117,6 +3189,11 @@ describe("useChat", () => {
mockFetchChatMessages
.mockResolvedValueOnce({ messages: [] })
.mockResolvedValueOnce({ messages: priorThreadNewestFirst });
let attachedHandlers: StreamAppendHandlers | undefined;
mockAttachChatStream.mockImplementation((_sessionId, nextHandlers) => {
attachedHandlers = nextHandlers;
return { close: vi.fn(), isConnected: () => true };
});
const { result } = renderHook(() => useChat("proj-123"));
@@ -3145,6 +3222,16 @@ describe("useChat", () => {
"msg-004",
]);
});
vi.useFakeTimers();
act(() => {
attachedHandlers?.onText(" plus");
});
act(() => {
vi.advanceTimersToNextTimer();
});
expect(result.current.streamingText).toBe("refreshed partial plus");
vi.useRealTimers();
});
it("FN-6496 loads prior thread when reconnectSessionSilently reattaches after send suspension", async () => {
@@ -3836,6 +3923,101 @@ describe("useChat", () => {
);
});
it("FN-6632 preserves chunks across selectSession recovery and repeated reattach", async () => {
const generatingSession = {
...makeSession({ id: "session-001", agentId: "agent-001", title: "Generating" }),
isGenerating: true,
inFlightGeneration: {
status: "generating" as const,
streamingText: "Hello ",
streamingThinking: "plan ",
toolCalls: [],
replayFromEventId: 5,
updatedAt: "2026-04-08T00:00:00.000Z",
},
};
const otherSession = makeSession({ id: "session-002", agentId: "agent-002", title: "Other" });
const handlers: StreamAppendHandlers[] = [];
const closeFirstStream = vi.fn();
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [generatingSession, otherSession] });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
mockAttachChatStream.mockImplementation((_sessionId, nextHandlers) => {
handlers.push(nextHandlers);
return {
close: handlers.length === 1 ? closeFirstStream : vi.fn(),
isConnected: () => true,
};
});
const { result } = renderHook(() => useChat("proj-123"));
await waitFor(() => {
expect(result.current.sessions).toHaveLength(2);
});
act(() => {
result.current.selectSession("session-001");
});
await waitFor(() => {
expect(result.current.streamingText).toBe("Hello ");
expect(handlers).toHaveLength(1);
});
vi.useFakeTimers();
act(() => {
handlers[0]?.onText("world");
});
act(() => {
vi.advanceTimersToNextTimer();
});
expect(result.current.streamingText).toBe("Hello world");
vi.useRealTimers();
act(() => {
result.current.selectSession("session-002");
});
expect(closeFirstStream).toHaveBeenCalledTimes(1);
act(() => {
result.current.selectSession("session-001", {
...generatingSession,
inFlightGeneration: {
...generatingSession.inFlightGeneration,
streamingText: "Hello world",
streamingThinking: "plan next ",
replayFromEventId: 6,
},
});
});
await waitFor(() => {
expect(result.current.streamingText).toBe("Hello world");
expect(handlers).toHaveLength(2);
});
vi.useFakeTimers();
act(() => {
handlers[1]?.onText("!");
handlers[1]?.onThinking("step");
});
act(() => {
vi.advanceTimersToNextTimer();
vi.advanceTimersToNextTimer();
});
expect(result.current.isStreaming).toBe(true);
expect(result.current.streamingText).toBe("Hello world!");
expect(result.current.streamingThinking).toBe("plan next step");
expect(mockAttachChatStream).toHaveBeenLastCalledWith(
"session-001",
expect.any(Object),
"proj-123",
{ lastEventId: 6 },
);
vi.useRealTimers();
});
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] });

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

@@ -64,6 +64,13 @@ function makeMessage(overrides: Partial<ChatMessage> & Pick<ChatMessage, "id" |
};
}
type StreamAppendHandlers = {
onText: (delta: string) => void;
onThinking: (delta: string) => void;
onToolStart: (data: { toolName: string; args?: Record<string, unknown> }) => void;
onToolEnd: (data: { toolName: string; isError: boolean; result?: unknown }) => void;
};
const setDocumentVisibilityState = (state: DocumentVisibilityState) => {
Object.defineProperty(document, "visibilityState", {
configurable: true,
@@ -2289,6 +2296,146 @@ describe("useQuickChat", () => {
);
});
it("FN-6632 preserves prior streamed chunks during QuickChat reattach", async () => {
const session = {
...makeSession({ id: "session-001", agentId: "agent-001" }),
isGenerating: true,
inFlightGeneration: {
status: "generating" as const,
streamingText: "Hello ",
streamingThinking: "plan ",
toolCalls: [{ toolName: "read", status: "running" as const, isError: false }],
replayFromEventId: 17,
updatedAt: "2026-04-08T00:00:00.000Z",
},
};
let attachedHandlers: StreamAppendHandlers | undefined;
mockFetchResumeChatSession.mockResolvedValue({ session });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
mockAttachChatStream.mockImplementation((_sessionId, handlers) => {
attachedHandlers = 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);
expect(result.current.streamingText).toBe("Hello ");
expect(attachedHandlers).toBeDefined();
});
vi.useFakeTimers();
act(() => {
attachedHandlers?.onText("world");
attachedHandlers?.onText("!");
attachedHandlers?.onThinking("more");
attachedHandlers?.onToolEnd({ toolName: "read", isError: false, result: "done" });
});
act(() => {
vi.advanceTimersToNextTimer();
vi.advanceTimersToNextTimer();
});
expect(result.current.isStreaming).toBe(true);
expect(result.current.streamingText).toBe("Hello world!");
expect(result.current.streamingThinking).toBe("plan more");
expect(result.current.streamingToolCalls).toEqual([
{ toolName: "read", status: "completed", isError: false, result: "done" },
]);
vi.useRealTimers();
});
it("FN-6632 preserves QuickChat chunks across selectSession and repeated reattach", async () => {
const generatingSession = {
...makeSession({ id: "session-001", agentId: "agent-001" }),
isGenerating: true,
inFlightGeneration: {
status: "generating" as const,
streamingText: "Hello ",
streamingThinking: "plan ",
toolCalls: [],
replayFromEventId: 17,
updatedAt: "2026-04-08T00:00:00.000Z",
},
};
const otherSession = makeSession({ id: "session-002", agentId: "agent-002" });
const handlers: StreamAppendHandlers[] = [];
const closeFirstStream = vi.fn();
mockFetchChatMessages.mockResolvedValue({ messages: [] });
mockAttachChatStream.mockImplementation((_sessionId, nextHandlers) => {
handlers.push(nextHandlers);
return {
close: handlers.length === 1 ? closeFirstStream : vi.fn(),
isConnected: () => true,
};
});
const { result } = renderHook(() => useQuickChat("proj-123"));
await act(async () => {
await result.current.selectSession(generatingSession);
});
await waitFor(() => {
expect(result.current.streamingText).toBe("Hello ");
expect(handlers).toHaveLength(1);
});
vi.useFakeTimers();
act(() => {
handlers[0]?.onText("world");
});
act(() => {
vi.advanceTimersToNextTimer();
});
expect(result.current.streamingText).toBe("Hello world");
vi.useRealTimers();
await act(async () => {
await result.current.selectSession(otherSession);
});
expect(closeFirstStream).toHaveBeenCalledTimes(1);
await act(async () => {
await result.current.selectSession({
...generatingSession,
inFlightGeneration: {
...generatingSession.inFlightGeneration,
streamingText: "Hello world",
replayFromEventId: 18,
},
});
});
await waitFor(() => {
expect(result.current.streamingText).toBe("Hello world");
expect(handlers).toHaveLength(2);
});
vi.useFakeTimers();
act(() => {
handlers[1]?.onText("!");
});
act(() => {
vi.advanceTimersToNextTimer();
});
expect(result.current.isStreaming).toBe(true);
expect(result.current.streamingText).toBe("Hello world!");
expect(mockAttachChatStream).toHaveBeenLastCalledWith(
"session-001",
expect.any(Object),
"proj-123",
{ lastEventId: 18 },
);
vi.useRealTimers();
});
it("FN-5104 reattaches once when selectSession refresh reveals generation from stale cache", async () => {
const staleSession = {
...makeSession({ id: "session-001", agentId: "agent-001" }),

19
packages/dashboard/app/hooks/createChatStreamHandlers.ts
View File

@@ -19,6 +19,12 @@ export interface CreateChatStreamHandlersOptions {
sessionId: string;
/** Optimistic temp id of the user message added before the stream started. */
tempUserMessageId: string;
/** Existing text snapshot for reattaching to an in-flight generation. */
initialText?: string;
/** Existing thinking snapshot for reattaching to an in-flight generation. */
initialThinking?: string;
/** Existing tool-call snapshot for reattaching to an in-flight generation. */
initialToolCalls?: ToolCallInfo[];
/**
* The latest text/thinking/tool-call snapshots that are committed to React
* state. We pass setters (not values) so the factory can flush per-frame
@@ -90,6 +96,9 @@ export function createChatStreamHandlers(
const {
sessionId,
tempUserMessageId,
initialText,
initialThinking,
initialToolCalls,
setStreamingText,
setStreamingThinking,
setStreamingToolCalls,
@@ -100,9 +109,13 @@ export function createChatStreamHandlers(
onFallbackSession,
} = options;
let capturedText = "";
let capturedThinking = "";
let capturedToolCalls: ToolCallInfo[] = [];
/**
* FNXC:ChatStreaming 2026-06-18-05:59:
* Reattached streams must seed their private accumulators from the durable in-flight snapshot, not only paint that snapshot into React state. The SSE replay starts after replayFromEventId, so the first post-reattach delta must append to snapshot text/thinking/tool calls instead of replacing everything the user already saw on load.
*/
let capturedText = initialText ?? "";
let capturedThinking = initialThinking ?? "";
let capturedToolCalls: ToolCallInfo[] = initialToolCalls ? [...initialToolCalls] : [];
let capturedFallbackInfo: FallbackInfo | undefined;
// Coalesce per-token state updates to one render per animation frame.

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

@@ -557,6 +557,10 @@ export function useChat(
void loadMessages(sessionId, { commitForStreamingAttach: true });
}
if (inFlightGeneration) {
/*
FNXC:ChatStreaming 2026-06-18-06:00:
Main chat paints the durable in-flight snapshot immediately for reattach UX, and passes the same snapshot into createChatStreamHandlers so the first replayed delta appends to accumulated text/thinking/tool calls instead of replacing the visible prefix.
*/
setStreamingText(inFlightGeneration.streamingText);
setStreamingThinking(inFlightGeneration.streamingThinking);
setStreamingToolCalls(inFlightGeneration.toolCalls);
@@ -566,6 +570,9 @@ export function useChat(
const { handlers } = createChatStreamHandlers({
sessionId,
tempUserMessageId: "",
initialText: inFlightGeneration?.streamingText,
initialThinking: inFlightGeneration?.streamingThinking,
initialToolCalls: inFlightGeneration?.toolCalls,
setStreamingText,
setStreamingThinking,
setStreamingToolCalls,

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

@@ -368,6 +368,10 @@ export function useQuickChat(
void loadMessagesForSession(sessionId, { commitForStreamingAttach: true });
}
if (inFlightGeneration) {
/*
FNXC:ChatStreaming 2026-06-18-06:01:
QuickChat must mirror main chat reattach semantics: paint the durable snapshot for the first frame and seed the handler accumulators from it so post-replay deltas continue the in-flight bubble instead of clobbering prior chunks.
*/
setStreamingText(inFlightGeneration.streamingText);
setStreamingThinking(inFlightGeneration.streamingThinking);
setStreamingToolCalls(inFlightGeneration.toolCalls);
@@ -377,6 +381,9 @@ export function useQuickChat(
const { handlers } = createChatStreamHandlers({
sessionId,
tempUserMessageId: "",
initialText: inFlightGeneration?.streamingText,
initialThinking: inFlightGeneration?.streamingThinking,
initialToolCalls: inFlightGeneration?.toolCalls,
setStreamingText,
setStreamingThinking,
setStreamingToolCalls,