diff --git a/ui/desktop/src/hooks/useSessionEvents.test.tsx b/ui/desktop/src/hooks/useSessionEvents.test.tsx new file mode 100644 index 0000000000..71eace5dd3 --- /dev/null +++ b/ui/desktop/src/hooks/useSessionEvents.test.tsx @@ -0,0 +1,94 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { act, renderHook } from '@testing-library/react'; +import { useSessionEvents, type SessionEvent } from './useSessionEvents'; + +vi.mock('../api', () => ({ + sessionEvents: vi.fn(), +})); + +import { sessionEvents } from '../api'; + +const sessionEventsMock = sessionEvents as unknown as ReturnType; + +function emptyStream() { + return { + stream: (async function* () { + // no events + })(), + }; +} + +function boundedMock( + limit: number, + factory: (callIndex: number) => Promise<{ stream: AsyncGenerator }> +) { + let calls = 0; + return () => { + const idx = calls++; + if (idx >= limit) { + return new Promise(() => {}); + } + return factory(idx); + }; +} + +async function flush(times = 200) { + for (let i = 0; i < times; i++) { + await Promise.resolve(); + } +} + +describe('useSessionEvents reconnect (issue #8717)', () => { + let originalSetTimeout: typeof globalThis.setTimeout; + + beforeEach(() => { + sessionEventsMock.mockReset(); + originalSetTimeout = globalThis.setTimeout; + globalThis.setTimeout = ((cb: () => void) => { + globalThis.queueMicrotask(cb); + return 0 as unknown as ReturnType; + }) as typeof globalThis.setTimeout; + }); + + afterEach(() => { + globalThis.setTimeout = originalSetTimeout; + }); + + it('synthesises a terminal Error for active listeners after a sustained failure streak', async () => { + const realDateNow = Date.now; + let fakeNow = 1_000_000; + Date.now = () => fakeNow; + + try { + sessionEventsMock.mockImplementation( + boundedMock(60, () => { + fakeNow += 10_000; + return Promise.resolve(emptyStream()); + }) + ); + + const { result, unmount } = renderHook(() => useSessionEvents('sess-1')); + + const handler = vi.fn(); + act(() => { + result.current.addListener('req-1', handler); + }); + + await flush(); + + const errorCalls = handler.mock.calls.filter( + (args) => (args[0] as SessionEvent).type === 'Error' + ); + expect(errorCalls.length).toBeGreaterThanOrEqual(1); + + const firstError = errorCalls[0][0] as SessionEvent & { error: string }; + expect(firstError.error).toBe('Lost connection to server'); + expect(firstError.request_id).toBe('req-1'); + expect(firstError.chat_request_id).toBe('req-1'); + + unmount(); + } finally { + Date.now = realDateNow; + } + }); +}); diff --git a/ui/desktop/src/hooks/useSessionEvents.ts b/ui/desktop/src/hooks/useSessionEvents.ts index 95c74947cb..36b10b7edd 100644 --- a/ui/desktop/src/hooks/useSessionEvents.ts +++ b/ui/desktop/src/hooks/useSessionEvents.ts @@ -7,7 +7,6 @@ import { sessionEvents, type MessageEvent } from '../api'; */ export type SessionEvent = MessageEvent & { request_id?: string; - /** Chat-level request UUID used for routing events to the correct handler. */ chat_request_id?: string; }; @@ -29,9 +28,29 @@ export function useSessionEvents(sessionId: string) { (async () => { let retryDelay = 500; const MAX_RETRY_DELAY = 10_000; - const MAX_CONSECUTIVE_ERRORS = 10; - let consecutiveErrors = 0; + const TERMINAL_ERROR_AFTER_MS = 5 * 60 * 1000; let lastEventId: string | undefined; + let failureStreakStartedAt: number | null = null; + + const broadcastTerminalErrorIfStuck = () => { + if (failureStreakStartedAt === null) return; + if (Date.now() - failureStreakStartedAt < TERMINAL_ERROR_AFTER_MS) return; + if (listenersRef.current.size === 0) { + failureStreakStartedAt = Date.now(); + return; + } + + const errorEvent: SessionEvent = { + type: 'Error', + error: 'Lost connection to server', + } as SessionEvent; + for (const [id, handlers] of listenersRef.current) { + for (const handler of [...handlers]) { + handler({ ...errorEvent, request_id: id, chat_request_id: id }); + } + } + failureStreakStartedAt = Date.now(); + }; while (!abortController.signal.aborted) { try { @@ -39,8 +58,6 @@ export function useSessionEvents(sessionId: string) { path: { id: sessionId }, signal: abortController.signal, headers: lastEventId ? { 'Last-Event-ID': lastEventId } : undefined, - // Disable the inner retry loop so errors surface to our outer - // loop which tracks consecutive failures and notifies listeners. sseMaxRetryAttempts: 1, onSseEvent: (event) => { if (event.id) { @@ -54,32 +71,22 @@ export function useSessionEvents(sessionId: string) { for await (const event of stream) { if (abortController.signal.aborted) break; - // Only mark as connected after the first real event arrives, - // since the HTTP request doesn't happen until iteration starts. if (!receivedEvent) { receivedEvent = true; setConnected(true); retryDelay = 500; - consecutiveErrors = 0; + failureStreakStartedAt = null; } - // The server adds chat_request_id (the chat UUID) and request_id - // to the JSON at the SSE framing layer. Route using chat_request_id - // so that Notification events (which carry their own MCP tool-call - // request_id) still reach the correct handler. const sessionEvent = event as SessionEvent; const routingId = sessionEvent.chat_request_id ?? sessionEvent.request_id; - // ActiveRequests events notify the client about in-flight requests - // it can reattach to (e.g. after a remount). if (sessionEvent.type === 'ActiveRequests') { const ids = (sessionEvent as unknown as { request_ids: string[] }).request_ids; activeRequestsHandlerRef.current?.(ids); continue; } - // Server-level errors without a request ID (e.g. "client too far - // behind") affect all active listeners — broadcast to everyone. if (!routingId && sessionEvent.type === 'Error') { for (const [id, handlers] of listenersRef.current) { for (const handler of handlers) { @@ -96,61 +103,22 @@ export function useSessionEvents(sessionId: string) { } } - // Stream ended. Reconnect unless we were intentionally aborted. if (abortController.signal.aborted) break; setConnected(false); - // If the stream ended without delivering any events, the connection - // likely failed silently (e.g. 404 with sseMaxRetryAttempts: 1). - // Treat it as an error so backoff and error counting apply. if (!receivedEvent) { - consecutiveErrors++; - console.warn( - `SSE stream ended with no events (${consecutiveErrors}/${MAX_CONSECUTIVE_ERRORS})` - ); - if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) { - console.error('SSE reconnect limit reached, notifying active listeners'); - const errorEvent: SessionEvent = { - type: 'Error', - error: 'Lost connection to server', - } as SessionEvent; - for (const [routingId, handlers] of listenersRef.current) { - for (const handler of handlers) { - handler({ ...errorEvent, request_id: routingId, chat_request_id: routingId }); - } - } - consecutiveErrors = 0; - } + if (failureStreakStartedAt === null) failureStreakStartedAt = Date.now(); + broadcastTerminalErrorIfStuck(); await new Promise((r) => setTimeout(r, retryDelay)); retryDelay = Math.min(retryDelay * 2, MAX_RETRY_DELAY); } } catch (error) { if (abortController.signal.aborted) break; - consecutiveErrors++; - console.warn( - `SSE connection error (${consecutiveErrors}/${MAX_CONSECUTIVE_ERRORS}), reconnecting:`, - error, - ); + console.warn('SSE connection error, reconnecting:', error); setConnected(false); - if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) { - console.error('SSE reconnect limit reached, notifying active listeners'); - // Send an error event to all active listeners so they can - // transition out of streaming state. Reset the counter so - // the loop keeps reconnecting for future requests. - const errorEvent: SessionEvent = { - type: 'Error', - error: 'Lost connection to server', - } as SessionEvent; - for (const [routingId, handlers] of listenersRef.current) { - for (const handler of handlers) { - handler({ ...errorEvent, request_id: routingId, chat_request_id: routingId }); - } - } - consecutiveErrors = 0; - } - - // Back off before retrying + if (failureStreakStartedAt === null) failureStreakStartedAt = Date.now(); + broadcastTerminalErrorIfStuck(); await new Promise((r) => setTimeout(r, retryDelay)); retryDelay = Math.min(retryDelay * 2, MAX_RETRY_DELAY); } @@ -168,25 +136,22 @@ export function useSessionEvents(sessionId: string) { }; }, [sessionId]); - const addListener = useCallback( - (requestId: string, handler: EventHandler): (() => void) => { - if (!listenersRef.current.has(requestId)) { - listenersRef.current.set(requestId, new Set()); - } - listenersRef.current.get(requestId)!.add(handler); + const addListener = useCallback((requestId: string, handler: EventHandler): (() => void) => { + if (!listenersRef.current.has(requestId)) { + listenersRef.current.set(requestId, new Set()); + } + listenersRef.current.get(requestId)!.add(handler); - return () => { - const set = listenersRef.current.get(requestId); - if (set) { - set.delete(handler); - if (set.size === 0) { - listenersRef.current.delete(requestId); - } + return () => { + const set = listenersRef.current.get(requestId); + if (set) { + set.delete(handler); + if (set.size === 0) { + listenersRef.current.delete(requestId); } - }; - }, - [] - ); + } + }; + }, []); const setActiveRequestsHandler = useCallback((handler: ActiveRequestsHandler | null) => { activeRequestsHandlerRef.current = handler;