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:
5
.changeset/FN-3760-first-send-hang-fix.md
Normal file
5
.changeset/FN-3760-first-send-hang-fix.md
Normal 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.
|
||||||
@@ -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();
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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] });
|
||||||
|
|||||||
@@ -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 });
|
||||||
|
|||||||
Reference in New Issue
Block a user