fix(ui): keep SSE reconnect loop alive on long disconnects (#8717) (#8846)

Signed-off-by: fresh3nough <anonwurcod@proton.me>
Signed-off-by: Douwe Osinga <douwe@squareup.com>
Co-authored-by: Douwe Osinga <douwe@squareup.com>

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<typeof vi.fn>;
+
+function emptyStream() {
+  return {
+    stream: (async function* () {
+      // no events
+    })(),
+  };
+}
+
+function boundedMock(
+  limit: number,
+  factory: (callIndex: number) => Promise<{ stream: AsyncGenerator<unknown> }>
+) {
+  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<typeof globalThis.setTimeout>;
+    }) 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;
+    }
+  });
+});

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;