Checkpointed reconnect logic

Author: Sg312

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts

@@ -169,9 +169,6 @@ const MAX_RECONNECT_ATTEMPTS = 10
 const RECONNECT_BASE_DELAY_MS = 1000
 const RECONNECT_MAX_DELAY_MS = 30_000
 const RECONNECT_EMPTY_BATCH_DELAY_MS = 500
-const RECONNECT_BATCH_REQUEST_TIMEOUT_MS = 15_000
-const RECONNECT_LIVE_TAIL_TIMEOUT_MS = 60_000
-const RECONNECT_RECOVERY_TIMEOUT_MS = 2 * 60_000
 
 const logger = createLogger('useChat')
 
@@ -197,43 +194,6 @@ function stringArrayParam(value: unknown): string[] {
   return value.filter((item): item is string => typeof item === 'string' && item.trim().length > 0)
 }
 
-function createTimedAbortSignal(
-  parentSignal: AbortSignal | undefined,
-  timeoutMs: number,
-  timeoutMessage: string
-): {
-  signal: AbortSignal
-  didTimeout: () => boolean
-  cleanup: () => void
-} {
-  const controller = new AbortController()
-  let timedOut = false
-
-  const onAbort = () => {
-    controller.abort(parentSignal?.reason)
-  }
-
-  if (parentSignal?.aborted) {
-    controller.abort(parentSignal.reason)
-  } else if (parentSignal) {
-    parentSignal.addEventListener('abort', onAbort, { once: true })
-  }
-
-  const timeoutId = setTimeout(() => {
-    timedOut = true
-    controller.abort(new Error(timeoutMessage))
-  }, timeoutMs)
-
-  return {
-    signal: controller.signal,
-    didTimeout: () => timedOut,
-    cleanup: () => {
-      clearTimeout(timeoutId)
-      parentSignal?.removeEventListener('abort', onAbort)
-    },
-  }
-}
-
 function resolveWorkflowNameForDisplay(workflowId: unknown): string | undefined {
   const id = stringParam(workflowId)
   if (!id) return undefined
@@ -2340,36 +2300,21 @@ export function useChat(
       afterCursor: string,
       signal?: AbortSignal
     ): Promise<StreamBatchResponse> => {
-      const timeoutMessage = `Timed out fetching stream batch for ${streamId}`
-      const timedAbort = createTimedAbortSignal(
-        signal,
-        RECONNECT_BATCH_REQUEST_TIMEOUT_MS,
-        timeoutMessage
+      const response = await fetch(
+        `/api/mothership/chat/stream?streamId=${encodeURIComponent(streamId)}&after=${encodeURIComponent(afterCursor)}&batch=true`,
+        { signal }
       )
-      try {
-        const response = await fetch(
-          `/api/mothership/chat/stream?streamId=${encodeURIComponent(streamId)}&after=${encodeURIComponent(afterCursor)}&batch=true`,
-          { signal: timedAbort.signal }
+      if (!response.ok) {
+        throw await createResumeTransportError(
+          response,
+          `Stream resume batch failed: ${response.status}`
         )
-        if (!response.ok) {
-          throw await createResumeTransportError(
-            response,
-            `Stream resume batch failed: ${response.status}`
-          )
-        }
-        const batch = parseStreamBatchResponse(await response.json())
-        if (Array.isArray(batch.previewSessions) && batch.previewSessions.length > 0) {
-          seedPreviewSessions(batch.previewSessions)
-        }
-        return batch
-      } catch (error) {
-        if (timedAbort.didTimeout() && !signal?.aborted) {
-          throw new Error(timeoutMessage)
-        }
-        throw error
-      } finally {
-        timedAbort.cleanup()
       }
+      const batch = parseStreamBatchResponse(await response.json())
+      if (Array.isArray(batch.previewSessions) && batch.previewSessions.length > 0) {
+        seedPreviewSessions(batch.previewSessions)
+      }
+      return batch
     },
     [seedPreviewSessions]
   )
@@ -2387,7 +2332,6 @@ export function useChat(
       let seedEvents = opts.initialBatch?.events ?? []
       let streamStatus = opts.initialBatch?.status ?? 'unknown'
       let streamError = opts.initialBatch?.error
-      const reconnectStartedAt = Date.now()
 
       const isStaleReconnect = () =>
         streamGenRef.current !== expectedGen || abortControllerRef.current?.signal.aborted === true
@@ -2401,15 +2345,6 @@ export function useChat(
 
       try {
         while (streamGenRef.current === expectedGen) {
-          if (Date.now() - reconnectStartedAt >= RECONNECT_RECOVERY_TIMEOUT_MS) {
-            logger.warn('Reconnect tail timed out waiting for terminal state', {
-              streamId,
-              afterCursor: latestCursor,
-              elapsedMs: Date.now() - reconnectStartedAt,
-            })
-            throw new Error(RECONNECT_TAIL_ERROR)
-          }
-
           if (seedEvents.length > 0) {
             const replayResult = await processSSEStreamRef.current(
               buildReplayStream(seedEvents).getReader(),
@@ -2440,47 +2375,26 @@ export function useChat(
 
           logger.info('Opening live stream tail', { streamId, afterCursor: latestCursor })
 
-          const tailTimeoutMessage = `Timed out waiting for live reconnect tail for ${streamId}`
-          const timedTailAbort = createTimedAbortSignal(
-            activeAbort.signal,
-            RECONNECT_LIVE_TAIL_TIMEOUT_MS,
-            tailTimeoutMessage
+          const sseRes = await fetch(
+            `/api/mothership/chat/stream?streamId=${encodeURIComponent(streamId)}&after=${encodeURIComponent(latestCursor)}`,
+            { signal: activeAbort.signal }
           )
-          let liveResult: { sawStreamError: boolean; sawComplete: boolean }
-          try {
-            const sseRes = await fetch(
-              `/api/mothership/chat/stream?streamId=${encodeURIComponent(streamId)}&after=${encodeURIComponent(latestCursor)}`,
-              { signal: timedTailAbort.signal }
-            )
-            if (!sseRes.ok || !sseRes.body) {
-              throw await createResumeTransportError(sseRes, RECONNECT_TAIL_ERROR)
-            }
+          if (!sseRes.ok || !sseRes.body) {
+            throw await createResumeTransportError(sseRes, RECONNECT_TAIL_ERROR)
+          }
 
-            if (isStaleReconnect()) {
-              return { error: false, aborted: true }
-            }
+          if (isStaleReconnect()) {
+            return { error: false, aborted: true }
+          }
 
-            setTransportStreaming()
+          setTransportStreaming()
 
-            liveResult = await processSSEStreamRef.current(
-              sseRes.body.getReader(),
-              assistantId,
-              expectedGen,
-              { preserveExistingState: true }
-            )
-          } catch (error) {
-            if (timedTailAbort.didTimeout() && !activeAbort.signal.aborted) {
-              logger.warn('Live reconnect tail request timed out', {
-                streamId,
-                afterCursor: latestCursor,
-                timeoutMs: RECONNECT_LIVE_TAIL_TIMEOUT_MS,
-              })
-              throw new Error(RECONNECT_TAIL_ERROR)
-            }
-            throw error
-          } finally {
-            timedTailAbort.cleanup()
-          }
+          const liveResult = await processSSEStreamRef.current(
+            sseRes.body.getReader(),
+            assistantId,
+            expectedGen,
+            { preserveExistingState: true }
+          )
 
           if (liveResult.sawStreamError) {
             return { error: true, aborted: false }