fix(FN-3760): add SSE first-event watchdog for legacy chat streams

- Add a first-event watchdog in legacy chat SSE handling to prevent hangs before initial stream data
- Cover watchdog behavior with legacy chat stream API tests
- Add hook-level regression tests for useChat and useQuickChat first-send stream scenarios
- Include a changeset for @runfusion/fusion patch release

Fusion-Task-Id: FN-3760
This commit is contained in:
Fusion
2026-05-08 20:45:45 -07:00
committed by gsxdsm
parent 9cfc2875d3
commit 9c86771295
5 changed files with 133 additions and 5 deletions

View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Fix first chat message send hanging on "Connecting…" — the initial SSE stream now completes reliably on cold-start.

View File

@@ -120,4 +120,23 @@ describe("streamChatResponse SSE parser", () => {
expect(donePayloads).toEqual([{ messageId: "" }]); expect(donePayloads).toEqual([{ messageId: "" }]);
}); });
}); });
it("fires onError when no stream events arrive before timeout", async () => {
vi.useFakeTimers();
vi.spyOn(globalThis, "fetch").mockResolvedValue(new Response(new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(new TextEncoder().encode(": connected\n\n"));
},
}), { status: 200 }));
const onError = vi.fn();
streamChatResponse("s-1", "hi", { onError }, undefined, undefined, { firstEventTimeoutMs: 1_000 });
await Promise.resolve();
await Promise.resolve();
await vi.advanceTimersByTimeAsync(1_100);
expect(onError).toHaveBeenCalledWith("Timed out waiting for first response event");
vi.useRealTimers();
});
}); });

View File

@@ -8315,21 +8315,39 @@ export function streamChatResponse(
}, },
attachments?: File[], attachments?: File[],
projectId?: string, projectId?: string,
options?: { maxReconnectAttempts?: number }, options?: { maxReconnectAttempts?: number; firstEventTimeoutMs?: number },
): { close: () => void; isConnected: () => boolean } { ): { close: () => void; isConnected: () => boolean } {
const url = buildApiUrl(withProjectId(`/chat/sessions/${encodeURIComponent(sessionId)}/messages`, projectId)); const url = buildApiUrl(withProjectId(`/chat/sessions/${encodeURIComponent(sessionId)}/messages`, projectId));
void options;
const abortController = new AbortController(); const abortController = new AbortController();
let closedByUser = false; let closedByUser = false;
let terminated = false; let terminated = false;
let receivedStreamEvent = false;
const firstEventTimeoutMs = Math.max(1_000, options?.firstEventTimeoutMs ?? 60_000);
let firstEventTimer: ReturnType<typeof setTimeout> | null = null;
const clearFirstEventTimer = (): void => {
if (firstEventTimer) {
clearTimeout(firstEventTimer);
firstEventTimer = null;
}
};
const markFirstEventReceived = (): void => {
if (receivedStreamEvent) {
return;
}
receivedStreamEvent = true;
clearFirstEventTimer();
};
const dispatchEvent = (eventName: string, rawData: string): void => { const dispatchEvent = (eventName: string, rawData: string): void => {
if (!eventName) { if (!eventName) {
return; return;
} }
markFirstEventReceived();
switch (eventName) { switch (eventName) {
case "thinking": case "thinking":
try { try {
@@ -8427,6 +8445,14 @@ export function streamChatResponse(
} }
handlers.onConnectionStateChange?.("connected"); handlers.onConnectionStateChange?.("connected");
firstEventTimer = setTimeout(() => {
if (terminated || closedByUser || receivedStreamEvent) {
return;
}
terminated = true;
handlers.onError?.("Timed out waiting for first response event");
abortController.abort();
}, firstEventTimeoutMs);
const reader = res.body.getReader(); const reader = res.body.getReader();
const decoder = new TextDecoder(); const decoder = new TextDecoder();
@@ -8498,14 +8524,20 @@ export function streamChatResponse(
if (!terminated && !closedByUser && !hasUndispatchedTrailingFragment) { if (!terminated && !closedByUser && !hasUndispatchedTrailingFragment) {
handlers.onError?.("Connection closed unexpectedly"); handlers.onError?.("Connection closed unexpectedly");
} }
clearFirstEventTimer();
} catch (err: unknown) { } catch (err: unknown) {
if (err instanceof DOMException && err.name === "AbortError") { if (err instanceof DOMException && err.name === "AbortError") {
if (!closedByUser) { if (!closedByUser && !terminated) {
handlers.onError?.("Connection aborted"); handlers.onError?.("Connection aborted");
} }
clearFirstEventTimer();
return; return;
} }
if (closedByUser) return; if (closedByUser) {
clearFirstEventTimer();
return;
}
clearFirstEventTimer();
handlers.onError?.(err instanceof Error ? err.message : "Connection error"); handlers.onError?.(err instanceof Error ? err.message : "Connection error");
} }
})(); })();
@@ -8513,6 +8545,7 @@ export function streamChatResponse(
return { return {
close: () => { close: () => {
closedByUser = true; closedByUser = true;
clearFirstEventTimer();
abortController.abort(); abortController.abort();
}, },
isConnected: () => !closedByUser, isConnected: () => !closedByUser,

View File

@@ -456,6 +456,46 @@ describe("useChat", () => {
}); });
}); });
it("sets isStreaming true during first send and clears on delayed done", async () => {
const session = makeSession({
id: "session-001",
agentId: "agent-001",
title: "Test Session",
});
mockFetchChatSessions.mockResolvedValue({ sessions: [session] });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
const { result } = renderHook(() => useChat(undefined, "project-123"));
await waitFor(() => {
expect(result.current.sessions).toHaveLength(1);
});
act(() => {
result.current.selectSession("session-001");
});
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
setTimeout(() => {
handlers.onDone?.({ messageId: "msg-001" });
}, 200);
return { close: vi.fn(), isConnected: () => true };
});
act(() => {
void result.current.sendMessage("Hello");
});
await waitFor(() => {
expect(result.current.isStreaming).toBe(true);
});
await waitFor(() => {
expect(result.current.isStreaming).toBe(false);
});
});
it("uses done payload assistant snapshot when no text chunks were streamed", async () => { it("uses done payload assistant snapshot when no text chunks were streamed", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" }); const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] }); mockFetchChatSessions.mockResolvedValueOnce({ sessions: [session] });

View File

@@ -78,6 +78,37 @@ describe("useQuickChat", () => {
await expect(sendResult).resolves.toBeUndefined(); await expect(sendResult).resolves.toBeUndefined();
}); });
it("sets isStreaming true during first send and clears on delayed done", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchResumeChatSession.mockResolvedValue({ session });
mockFetchChatMessages.mockResolvedValue({ messages: [] });
const { result } = renderHook(() => useQuickChat("proj-123"));
await act(async () => {
await result.current.switchSession("agent-001");
});
mockStreamChatResponse.mockImplementation((_sessionId, _content, handlers) => {
setTimeout(() => {
handlers.onDone?.({ messageId: "msg-001" });
}, 200);
return { close: vi.fn(), isConnected: () => true };
});
void act(() => {
void result.current.sendMessage("Hello");
});
await waitFor(() => {
expect(result.current.isStreaming).toBe(true);
});
await waitFor(() => {
expect(result.current.isStreaming).toBe(false);
});
});
it("uses done payload assistant snapshot when no text chunks were streamed", async () => { it("uses done payload assistant snapshot when no text chunks were streamed", async () => {
const session = makeSession({ id: "session-001", agentId: "agent-001" }); const session = makeSession({ id: "session-001", agentId: "agent-001" });
mockFetchResumeChatSession.mockResolvedValue({ session }); mockFetchResumeChatSession.mockResolvedValue({ session });