fix(sse): raise the ceiling and narrow reconnect resync to the lists

Deciding from cache whether a chat is still streaming is not reliable — the optimistic markers outlive the turn, and each refinement of that predicate exposed another state where it answers wrongly. Drop it: the resync now invalidates only the workspace lists, which is always safe, and chat detail reconciliation stays as it is today rather than being half-solved here.

Raise the ceiling to 4h, matching lib/realtime/event-stream-route.ts. A healthy client is drained and so is never unread; the unread check is what reclaims a vanished consumer, and it does so within minutes. A short ceiling would therefore only force reconnects on the connections that are working, and every reconnect is a window where a transient event can be missed. Retention stays bounded by the ceiling instead of by process uptime.

Author: waleedlatif1

apps/sim/hooks/use-mothership-chat-events.test.ts

@@ -432,55 +432,20 @@ describe('resyncMothershipChatCaches', () => {
     vi.clearAllMocks()
   })
 
-  function detailPredicate() {
-    resyncMothershipChatCaches(queryClient, 'ws-1')
-    const predicate = queryClient.invalidateQueries.mock.calls
-      .map(([arg]: [{ predicate?: (query: unknown) => boolean }]) => arg.predicate)
-      .find(Boolean)
-    if (!predicate) throw new Error('detail invalidation did not pass a predicate')
-    return (data: unknown) => predicate({ state: { data } })
-  }
-
-  it('invalidates the workspace lists and the chat details', () => {
+  it('invalidates the workspace lists', () => {
     resyncMothershipChatCaches(queryClient, 'ws-1')
 
-    expect(queryClient.invalidateQueries).toHaveBeenCalledTimes(2)
+    expect(queryClient.invalidateQueries).toHaveBeenCalledTimes(1)
     expect(queryClient.invalidateQueries).toHaveBeenCalledWith({
       queryKey: mothershipChatKeys.workspaceLists('ws-1'),
     })
-    expect(queryClient.invalidateQueries).toHaveBeenCalledWith(
-      expect.objectContaining({ queryKey: mothershipChatKeys.details() })
-    )
   })
 
-  it('skips the detail of a chat this client is still streaming', () => {
-    expect(
-      detailPredicate()({
-        messages: [{ id: 'new-stream' }, { id: 'live-assistant:new-stream' }],
-        activeStreamId: 'new-stream',
-        streamSnapshot: { events: [], previewSessions: [], status: 'streaming' },
-      })
-    ).toBe(false)
-  })
-
-  it('invalidates a chat whose stream finished but left its optimistic markers cached', () => {
-    const predicate = detailPredicate()
-    const finished = (status: string) => ({
-      messages: [{ id: 'new-stream' }, { id: 'live-assistant:new-stream' }],
-      activeStreamId: 'new-stream',
-      streamSnapshot: { events: [], previewSessions: [], status },
-    })
-
-    expect(predicate(finished('complete'))).toBe(true)
-    expect(predicate(finished('error'))).toBe(true)
-    expect(predicate(finished('cancelled'))).toBe(true)
-  })
-
-  it('invalidates details with no active stream, and streams not rendered locally', () => {
-    const predicate = detailPredicate()
+  it('leaves chat details untouched so a mounted stream cannot be refetched mid-turn', () => {
+    resyncMothershipChatCaches(queryClient, 'ws-1')
 
-    expect(predicate(undefined)).toBe(true)
-    expect(predicate({ messages: [{ id: 'stream-1' }] })).toBe(true)
-    expect(predicate({ messages: [{ id: 'stream-1' }], activeStreamId: 'stream-1' })).toBe(true)
+    expect(queryClient.invalidateQueries).not.toHaveBeenCalledWith(
+      expect.objectContaining({ queryKey: mothershipChatKeys.details() })
+    )
   })
 })

apps/sim/hooks/use-mothership-chat-events.ts

@@ -2,10 +2,7 @@ import { useEffect } from 'react'
 import { createLogger } from '@sim/logger'
 import type { QueryClient } from '@tanstack/react-query'
 import { useQueryClient } from '@tanstack/react-query'
-import {
-  getLiveAssistantMessageId,
-  isTerminalStreamStatus,
-} from '@/lib/copilot/chat/effective-transcript'
+import { getLiveAssistantMessageId } from '@/lib/copilot/chat/effective-transcript'
 import { isChatEnabled } from '@/lib/core/config/env-flags'
 import { suspendDesktopChatScopes } from '@/lib/desktop/chat-scope'
 import { type MothershipChatHistory, mothershipChatKeys } from '@/hooks/queries/mothership-chats'
@@ -41,23 +38,6 @@ function isLocalOptimisticActiveStream(current: MothershipChatHistory | undefine
   return current.messages.some((message) => message.id === liveAssistantId)
 }
 
-/**
- * True while this client is still rendering a stream for the chat.
- *
- * The optimistic markers alone are not enough: a finished turn can leave
- * `activeStreamId` and its live-assistant message in the cache (finalization
- * skips detail invalidation when a follow-up is queued), and treating those as
- * live would exclude the chat from every future resync — permanently, since
- * only a refetch would clear them. Requiring a non-terminal snapshot status
- * keeps the skip to turns that are genuinely still streaming.
- */
-function isStreamingLocally(current: MothershipChatHistory | undefined) {
-  return (
-    isLocalOptimisticActiveStream(current) &&
-    !isTerminalStreamStatus(current?.streamSnapshot?.status)
-  )
-}
-
 /**
  * Returns true when the cached active stream is known to be later in the
  * chronological transcript than the stream that emitted this status event.
@@ -152,28 +132,25 @@ export function handleMothershipChatStatusEvent(
 }
 
 /**
- * Re-syncs chat caches after a gap in the event stream.
+ * Re-syncs the workspace chat lists after a gap in the event stream.
  *
  * `task_status` events are transient — nothing replays what was published while
- * no connection was open — so any reconnect may have missed a create, rename,
- * delete, or completion. Invalidating the workspace lists and every chat detail
- * reconciles from the server; only queries that are currently mounted refetch.
+ * no connection was open — so a reconnect may have missed a create, rename, or
+ * delete. The lists carry that workspace-level state, and refetching them is
+ * always safe.
  *
- * A chat this client is still streaming is left alone, for the same reason
- * status events skip it: refetching mid-stream would replace the optimistic
- * transcript with a server copy that does not yet hold the in-flight message.
- * That chat reconciles when its own stream finishes.
+ * Chat details are deliberately left alone. The only one that would refetch is
+ * the mounted chat, which may be rendering an in-flight stream, and refetching
+ * there replaces the optimistic transcript with a server copy that does not yet
+ * hold the streaming message. Cached state cannot reliably say whether a turn is
+ * still running — the optimistic markers outlive it — so detail reconciliation
+ * stays as it is today and belongs with the streaming state that can answer it.
  */
 export function resyncMothershipChatCaches(
   queryClient: Pick<QueryClient, 'invalidateQueries'>,
   workspaceId: string
 ): void {
   queryClient.invalidateQueries({ queryKey: mothershipChatKeys.workspaceLists(workspaceId) })
-  queryClient.invalidateQueries({
-    queryKey: mothershipChatKeys.details(),
-    predicate: (query) =>
-      !isStreamingLocally(query.state.data as MothershipChatHistory | undefined),
-  })
 }
 
 /**

apps/sim/lib/copilot/chat/effective-transcript.ts

@@ -52,7 +52,7 @@ function asPayloadRecord(value: unknown): Record<string, unknown> | undefined {
   return isRecordLike(value) ? value : undefined
 }
 
-export function isTerminalStreamStatus(status: string | null | undefined): boolean {
+function isTerminalStreamStatus(status: string | null | undefined): boolean {
   return (
     status === MothershipStreamV1CompletionStatus.complete ||
     status === MothershipStreamV1CompletionStatus.error ||

apps/sim/lib/events/sse-endpoint.ts

@@ -29,16 +29,22 @@ const encoder = new TextEncoder()
 export const HEARTBEAT_INTERVAL_MS = 30_000
 
 /**
- * Hard ceiling on one connection's lifetime.
+ * Defensive ceiling on one connection's lifetime; `EventSource` reconnects past
+ * this, so delivery continues across the boundary.
  *
  * `request.signal` abort and stream `cancel()` are the primary teardown paths,
- * but both fire only when the runtime reports the client disconnect. This
- * ceiling releases the connection without depending on that report, so a
- * missed disconnect costs one connection rather than accumulating for the life
- * of the process. `EventSource` reconnects on its own, so delivery continues
- * across the boundary.
+ * but both fire only when the runtime reports the client disconnect, and the
+ * unread check below only catches a consumer that has stopped draining. This
+ * releases whatever both miss, so retention is bounded by the ceiling instead
+ * of by process uptime.
+ *
+ * Matches the ceiling `lib/realtime/event-stream-route.ts` already uses for the
+ * same purpose. It is deliberately far longer than the unread window: a healthy
+ * client is drained and therefore never unread, so a short ceiling would only
+ * force reconnects on the connections that are working, and every reconnect is
+ * a window in which a transient event can be missed.
  */
-export const MAX_CONNECTION_MS = 15 * 60 * 1000
+export const MAX_CONNECTION_MS = 4 * 60 * 60 * 1000
 
 /** Spreads reconnects so connections opened together do not expire together. */
 export const MAX_CONNECTION_JITTER_MS = 60_000