improvement(chat): coalesce replayed transcript reloads and narrow text-only message reads

apps/sim/app/workspace/[workspaceId]/home/components/mothership-view/components/resource-registry/resource-invalidation.ts

@@ -3,7 +3,7 @@ import type { MothershipResourceType } from '@/lib/mothership/resources/types'
 import { dashboardKeys } from '@/hooks/queries/dashboards'
 import { deploymentKeys, invalidateDeploymentQueries } from '@/hooks/queries/deployments'
 import { logKeys } from '@/hooks/queries/logs'
-import { mothershipChatKeys } from '@/hooks/queries/mothership-chats'
+import { mothershipChatKeys, refreshMothershipChatDetail } from '@/hooks/queries/mothership-chats'
 import { folderKeys } from '@/hooks/queries/utils/folder-keys'
 import { invalidateWorkflowLists } from '@/hooks/queries/utils/invalidate-workflow-lists'
 import { knowledgeKeys } from '@/hooks/queries/utils/knowledge-keys'
@@ -59,7 +59,7 @@ const RESOURCE_INVALIDATORS: Record<
   },
   task: (qc, wId, id) => {
     invalidate(qc, mothershipChatKeys.workspaceLists(wId))
-    if (id) invalidate(qc, mothershipChatKeys.detail(id))
+    if (id) refreshMothershipChatDetail(qc, id)
   },
   log: (qc, wId, id) => {
     invalidate(qc, [...logKeys.lists(), wId])

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.test.ts

@@ -1,3 +1,4 @@
+import { createDeferred } from '@sim/testing/helpers/deferred'
 import { QueryClient, QueryObserver } from '@tanstack/react-query'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { resourceScopeKey } from '@/lib/core/resource-scope'
@@ -26,7 +27,10 @@ vi.mock('@/hooks/queries/utils/workflow-cache', () => ({
 
 import type { PersistedStreamEventEnvelope } from '@/lib/mothership/request/session/contract'
 import { handleResourceEvent } from '@/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event'
-import type { StreamLoopContext } from '@/app/workspace/[workspaceId]/home/hooks/stream/stream-context'
+import {
+  createStreamLoopContext,
+  type StreamLoopContext,
+} from '@/app/workspace/[workspaceId]/home/hooks/stream/stream-context'
 import {
   makeStreamLoopDeps,
   ref,
@@ -197,6 +201,137 @@ describe('live resource effects', () => {
   )
 })
 
+describe('recovered resource events', () => {
+  const detailKey = mothershipChatKeys.detail('chat-1')
+  const tables = ['tbl-1', 'tbl-2', 'tbl-3'].map(
+    (id): MothershipResource => ({ type: 'table', id, title: id })
+  )
+
+  function replayed(resource: MothershipResource, seq: number): ResourceEvent {
+    return {
+      ...removeEvent('file', resource.id),
+      seq,
+      payload: { op: 'upsert', replay: true, resource },
+    } as ResourceEvent
+  }
+
+  /** A mounted chat whose every read returns the resources stored when the read began. */
+  function mountedChat(queryClient: QueryClient, { cached = true, hangFirstRead = false } = {}) {
+    const server = { stored: [] as MothershipResource[] }
+    const reads = vi.fn(
+      async (): Promise<MothershipChatHistory> => ({
+        id: 'chat-1',
+        title: null,
+        messages: [],
+        activeStreamId: 'stream-1',
+        resources: [...server.stored],
+      })
+    )
+    if (hangFirstRead) reads.mockImplementationOnce(() => new Promise(() => {}))
+    if (cached) {
+      queryClient.setQueryData(detailKey, {
+        id: 'chat-1',
+        title: null,
+        messages: [],
+        activeStreamId: 'stream-1',
+        resources: [],
+      } satisfies MothershipChatHistory)
+    }
+    const unsubscribe = new QueryObserver(queryClient, {
+      queryKey: detailKey,
+      queryFn: reads,
+      staleTime: Number.POSITIVE_INFINITY,
+    }).subscribe(() => {})
+    return { server, reads, unsubscribe }
+  }
+
+  async function settledResources(queryClient: QueryClient) {
+    await vi.waitFor(() => expect(queryClient.isFetching({ queryKey: detailKey })).toBe(0))
+    return queryClient.getQueryData<MothershipChatHistory>(detailKey)?.resources
+  }
+
+  it('reloads the transcript once for a whole batch replay, after its terminal flush', async () => {
+    const queryClient = new QueryClient()
+    const { server, reads, unsubscribe } = mountedChat(queryClient)
+    const ctx = createStreamLoopContext(
+      makeStreamLoopDeps({
+        queryClient,
+        chatIdRef: ref<string | undefined>('chat-1'),
+        options: { deferFlushes: true },
+      })
+    )
+    server.stored = tables
+
+    tables.forEach((table, index) => handleResourceEvent(ctx, replayed(table, index + 1)))
+    expect(reads).not.toHaveBeenCalled()
+    ctx.ops.forceFlush()
+
+    expect(await settledResources(queryClient)).toEqual(tables)
+    expect(reads).toHaveBeenCalledOnce()
+    unsubscribe()
+  })
+
+  it('coalesces a reconnected tail burst into one read plus one trailing read that sees the last change', async () => {
+    const queryClient = new QueryClient()
+    const { server, reads, unsubscribe } = mountedChat(queryClient)
+    const deps = makeStreamLoopDeps({ queryClient, chatIdRef: ref<string | undefined>('chat-1') })
+    const ctx = createStreamLoopContext(deps)
+    const firstReadReleased = createDeferred<void>()
+    reads.mockImplementationOnce(async () => {
+      const resources = [...server.stored]
+      await firstReadReleased.promise
+      return { id: 'chat-1', title: null, messages: [], activeStreamId: 'stream-1', resources }
+    })
+
+    server.stored = tables.slice(0, 1)
+    handleResourceEvent(ctx, replayed(tables[0], 1))
+    server.stored = tables
+    handleResourceEvent(ctx, replayed(tables[1], 2))
+    handleResourceEvent(ctx, replayed(tables[2], 3))
+    firstReadReleased.resolve()
+
+    await vi.waitFor(() => expect(reads).toHaveBeenCalledTimes(2))
+    expect(await settledResources(queryClient)).toEqual(tables)
+    expect(reads).toHaveBeenCalledTimes(2)
+    unsubscribe()
+  })
+
+  it.each([
+    { name: 'a loaded transcript', cached: true },
+    { name: 'a transcript still on its first load', cached: false },
+  ])(
+    'stops holding reloads of $name behind one that never settles once 30 seconds pass',
+    async ({ cached }) => {
+      const queryClient = new QueryClient()
+      const { server, reads, unsubscribe } = mountedChat(queryClient, {
+        cached,
+        hangFirstRead: true,
+      })
+      const ctx = createStreamLoopContext(
+        makeStreamLoopDeps({ queryClient, chatIdRef: ref<string | undefined>('chat-1') })
+      )
+      vi.useFakeTimers({ toFake: ['Date'] })
+      try {
+        handleResourceEvent(ctx, replayed(tables[0], 1))
+        await vi.waitFor(() => expect(reads).toHaveBeenCalledOnce())
+        handleResourceEvent(ctx, replayed(tables[1], 2))
+        expect(reads).toHaveBeenCalledOnce()
+
+        server.stored = tables
+        vi.setSystemTime(Date.now() + 30_000)
+        handleResourceEvent(ctx, replayed(tables[2], 3))
+
+        await vi.waitFor(() => expect(reads).toHaveBeenCalledTimes(2))
+        expect(await settledResources(queryClient)).toEqual(tables)
+        expect(reads).toHaveBeenCalledTimes(2)
+      } finally {
+        vi.useRealTimers()
+        unsubscribe()
+      }
+    }
+  )
+})
+
 describe('handleResourceEvent saved-view pins', () => {
   beforeEach(() => {
     useTableViewPinStore.getState().reset()

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.ts

@@ -24,7 +24,11 @@ import type {
   MothershipResource,
   MothershipResourceType,
 } from '@/app/workspace/[workspaceId]/home/types'
-import { type MothershipChatHistory, mothershipChatKeys } from '@/hooks/queries/mothership-chats'
+import {
+  type MothershipChatHistory,
+  mothershipChatKeys,
+  refreshMothershipChatDetail,
+} from '@/hooks/queries/mothership-chats'
 import { knowledgeKeys } from '@/hooks/queries/utils/knowledge-keys'
 import { removeWorkflowFromActiveCache } from '@/hooks/queries/utils/workflow-cache'
 import { useTableViewPinStore } from '@/stores/table/view-pin/store'
@@ -63,6 +67,18 @@ function applyResourceEffect(
   }
 }
 
+/**
+ * Recovery reloads the current panel state from the stored chat instead of
+ * repeating historical focus commands. A batch replay reloads once at its
+ * terminal flush; a reconnected live tail coalesces its reloads.
+ */
+function refreshRecoveredChatResources(ctx: StreamLoopContext): void {
+  const chatId = ctx.deps.chatIdRef.current
+  if (!chatId) return
+  if (ctx.deps.options.deferFlushes) ctx.state.deferredChatDetailRefreshes.add(chatId)
+  else refreshMothershipChatDetail(ctx.deps.queryClient, chatId)
+}
+
 /**
  * Applies a streamed resource upsert/remove to the mothership resource list,
  * reconciling it with any in-flight or just-completed file-preview handoff so a
@@ -159,9 +175,7 @@ export function handleResourceEvent(ctx: StreamLoopContext, parsed: ResourceEven
       }
     }
     if (payload.replay || ctx.deps.options.deferFlushes) {
-      const chatId = ctx.deps.chatIdRef.current
-      if (chatId)
-        void queryClient.invalidateQueries({ queryKey: mothershipChatKeys.detail(chatId) })
+      refreshRecoveredChatResources(ctx)
       return
     }
     if (ctx.deps.citedSourcesEnabled) return
@@ -190,9 +204,7 @@ export function handleResourceEvent(ctx: StreamLoopContext, parsed: ResourceEven
   }
   if (payload.op === 'refresh') return
   if (payload.replay || ctx.deps.options.deferFlushes) {
-    /** Recovery refetches current panel state instead of repeating historical focus commands. */
-    const chatId = ctx.deps.chatIdRef.current
-    if (chatId) void queryClient.invalidateQueries({ queryKey: mothershipChatKeys.detail(chatId) })
+    refreshRecoveredChatResources(ctx)
     return
   }
   if (payload.op === 'clear_view') {

apps/sim/app/workspace/[workspaceId]/home/hooks/stream/stream-context.ts

@@ -22,7 +22,10 @@ import type {
   MothershipResource,
   MothershipResourceType,
 } from '@/app/workspace/[workspaceId]/home/types'
-import type { MothershipChatHistory } from '@/hooks/queries/mothership-chats'
+import {
+  type MothershipChatHistory,
+  refreshMothershipChatDetail,
+} from '@/hooks/queries/mothership-chats'
 
 /** Minimum spacing between text-driven snapshot flushes (see flushText). */
 const MIN_TEXT_FLUSH_INTERVAL_MS = 50
@@ -83,6 +86,8 @@ export interface StreamLoopState {
   scheduledTextFlushFrame: number | null
   /** Trailing timer for the min-interval text-flush gate (see flushText). */
   scheduledTextFlushTimer: ReturnType<typeof setTimeout> | null
+  /** Chats a batch replay recovered resources for, reloaded once by its terminal flush. */
+  deferredChatDetailRefreshes: Set<string>
 }
 
 export interface StreamEventScope {
@@ -198,7 +203,10 @@ export interface StreamLoopOps {
   isStale: () => boolean
   flush: () => void
   flushText: () => void
-  /** Real flush that bypasses `deferFlushes` — the batch-replay terminal flush. */
+  /**
+   * Real flush that bypasses `deferFlushes` — the batch-replay terminal flush — then
+   * reloads each chat the replay recovered resources for.
+   */
   forceFlush: () => void
 }
 
@@ -234,6 +242,7 @@ export function createStreamLoopContext(deps: StreamLoopDeps): StreamLoopContext
     browserAgentRunIds: new Set(),
     scheduledTextFlushFrame: null,
     scheduledTextFlushTimer: null,
+    deferredChatDetailRefreshes: new Set(),
   }
 
   if (preserveState) {
@@ -371,7 +380,13 @@ export function createStreamLoopContext(deps: StreamLoopDeps): StreamLoopContext
       flushAndStamp()
     },
     flushText,
-    forceFlush: flush,
+    forceFlush: () => {
+      flush()
+      for (const chatId of state.deferredChatDetailRefreshes) {
+        refreshMothershipChatDetail(deps.queryClient, chatId)
+      }
+      state.deferredChatDetailRefreshes.clear()
+    },
   }
 
   return { state, ops, deps }

apps/sim/hooks/queries/mothership-chats.ts

@@ -109,6 +109,53 @@ export const mothershipChatKeys = {
   detail: (chatId: string | undefined) => [...mothershipChatKeys.details(), chatId ?? ''] as const,
 }
 
+/**
+ * How long a chat's reload in flight holds back later requests. Past it a hung fetch no longer
+ * blocks the chat: the next request starts its own reload, which replaces the stuck one.
+ */
+const CHAT_DETAIL_REFRESH_HOLD_MS = 30 * 1000
+
+interface ChatDetailRefresh {
+  startedAt: number
+  requestedAgain: boolean
+}
+
+/** Per client, the chats with a reload in flight. */
+const chatDetailRefreshes = new WeakMap<QueryClient, Map<string, ChatDetailRefresh>>()
+
+/**
+ * Invalidates a chat's detail so a mounted transcript reloads, coalescing a burst of
+ * requests (a replayed or reconnected stream re-announces every resource it touched)
+ * into one reload in flight plus one trailing reload. The trailing reload starts after
+ * the last request, so the final read reflects every change requested meanwhile; a read
+ * already in flight may predate a request and is never trusted to include it. A reload
+ * in flight longer than {@link CHAT_DETAIL_REFRESH_HOLD_MS} stops holding requests back.
+ */
+export function refreshMothershipChatDetail(queryClient: QueryClient, chatId: string): void {
+  const refreshes = chatDetailRefreshes.get(queryClient) ?? new Map<string, ChatDetailRefresh>()
+  chatDetailRefreshes.set(queryClient, refreshes)
+  const inFlight = refreshes.get(chatId)
+  if (inFlight && Date.now() - inFlight.startedAt < CHAT_DETAIL_REFRESH_HOLD_MS) {
+    inFlight.requestedAgain = true
+    return
+  }
+  const refresh: ChatDetailRefresh = { startedAt: Date.now(), requestedAgain: false }
+  refreshes.set(chatId, refresh)
+  const queryKey = mothershipChatKeys.detail(chatId)
+  // A query with no data yet joins its fetch in flight instead of replacing it, so a stuck
+  // one is cancelled first.
+  const reload = inFlight
+    ? queryClient
+        .cancelQueries({ queryKey, exact: true })
+        .then(() => queryClient.invalidateQueries({ queryKey }))
+    : queryClient.invalidateQueries({ queryKey })
+  void reload.finally(() => {
+    if (refreshes.get(chatId) !== refresh) return
+    refreshes.delete(chatId)
+    if (refresh.requestedAgain) refreshMothershipChatDetail(queryClient, chatId)
+  })
+}
+
 /** Shared by the `useMothershipChats` hook and the workspace sidebar prefetch. */
 export const MOTHERSHIP_CHAT_LIST_STALE_TIME = 60 * 1000
 export const MOTHERSHIP_CHAT_HISTORY_STALE_TIME = 30 * 1000