fix(mothership): close the remaining ways a chat send is lost, resent hot, or polled after unmount (#8695)

* fix(mothership): close the remaining ways a chat send is lost, resent hot, or polled after unmount

- A deleted chat's queue could come back through `enqueue`: a direct send to a
  chat deleted while its POST was failing (unreachable or busy) was re-queued,
  invisible, and would go out if the chat were restored. A cleared chat now
  refuses every queue write until it is restored from Recently Deleted.
- A busy refusal re-queued its message for immediate redispatch. With Redis
  down every send is refused as busy without naming a turn, so the message
  was resent on every refusal (or stalled in the queue). Busy re-queues now
  wait out a jittered, growing delay (`backoffWithJitter`, 1s to 30s) before
  the drain sends them again under the same id.
- A send waiting on the server's chat lock behind another tab's turn was
  aborted by a return-triggered recovery, and its message was lost. Recovery
  now leaves any POST that has not answered alone unless the chat already
  holds its message. When the busy refusal then arrives, the chat attaches to
  the turn holding it explicitly: the history read it relies on can repeat one
  the surface skipped while the POST was pending, so nothing else re-ran to
  attach, and the queued message waited behind a turn the tab never showed.
- A retry told "already sent" while the server's earlier attempt with that id
  was still in flight (it had opened no stream yet) reattached to the missing
  stream, read the 404 as a finished turn, and dropped the message. Seen in
  the browser harness when Redis came back during a busy-refusal retry. Such a
  refusal is now retried later like a busy one.
- A Send-now message whose Stop handoff failed (for example because the user
  switched chats while the Stop settled) returned a plain failure that the
  dispatch epoch check discarded. It is now handed back held, so it stays in
  its own chat's queue for the user to send; a surface that unmounted keeps
  resuming it from its stored handoff as before.
- The saved-turn re-read kept refetching an unobserved chat for up to two
  minutes after the chat view unmounted. It now stops with the view.

* fix(mothership): keep deleted chats empty across tabs and back off busy first messages

- migrate no longer moves a new-chat queue into a chat deleted meanwhile
- another tab's delete clears and tombstones the chat's queue; a restore
  (published as created) or the chat loading lifts it
- a busy refusal on the new-chat surface waits out the backoff in its own
  queue instead of being handed to the surface's listener and resent at once
- a deduplicated send whose stream check resolves after the user moved on
  leaves Stop on the new turn

* fix(mothership): lift a chat's delete only when a later server read returns it

The chat history effect also ran on history written locally, such as the
rollback after a busy refusal, so a chat deleted in another tab could take
queued sends again before any restore. The delete is now lifted only by a
server read that began after it.

* fix(mothership): apply the restore check to every server read of a chat

Recovery on a return to the tab reads the chat directly, not through the
history query, so a restore it found left the chat refusing follow-ups.
The check now lives in fetchMothershipChatHistory itself.

* test(mothership): use the central request mock in the restore read test

* fix(mothership): lift only the chat delete a read or restore saw

Each delete now gets a token. A history read or a restore lifts the delete
it saw when it began, so a slow answer cannot reopen a chat deleted again
while it was in flight.

* fix(mothership): keep a send refused as busy after the user switched chats

A chat switch detaches the view but leaves a send waiting on the chat lock
running. Its busy refusal was dropped as a stale answer, so the message was
lost. A conflict is now handled after a switch too, re-queuing the message
in its own chat. The busy branch attaches through recovery's own history
read instead of reading the chat twice.

* fix(mothership): check a deduplicated send's stream before adopting its chat

On the new-chat surface, an "already sent" answer names the chat the
earlier attempt opened. Adopting it before finding no stream moved the
surface to that chat while the retried message was queued under the
new-chat key, where nothing sent it. The stream is now checked first.

* fix(mothership): retry a deduplicated send whose stream lookup failed

A lookup that failed with a network error or 5xx was read as proof that the
earlier attempt's stream existed, so a send that never started was
reattached and finalized instead of re-sent. Only a lookup this send aborted
skips the retry now; the server deduplicates the retry by id.

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.dom.test.tsx

@@ -0,0 +1,119 @@
+import {
+  apiClientRequestMock,
+  apiClientRequestMockFns,
+} from '@sim/testing/mocks/api-client-request.mock'
+import { reactQueryMock } from '@sim/testing/mocks/react-query.mock'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+vi.mock('@/lib/api/client/request', () => apiClientRequestMock)
+vi.mock('@tanstack/react-query', () => reactQueryMock)
+
+import {
+  fetchMothershipChatHistory,
+  type MothershipChatHistory,
+  useRestoreMothershipChat,
+} from '@/hooks/queries/mothership-chats'
+import { useMothershipQueueStore } from '@/stores/mothership-queue/store'
+
+const mockRequestJson = apiClientRequestMockFns.mockRequestJson
+
+const history: MothershipChatHistory = {
+  id: 'chat-1',
+  mode: 'agent',
+  title: 'Restored',
+  messages: [],
+  activeStreamId: null,
+  resources: [],
+}
+
+/** Whether the queue store takes a send for the chat, i.e. whether its delete still holds. */
+function takesSends(chatId: string): boolean {
+  useMothershipQueueStore.getState().enqueue(chatId, { id: 'probe', content: 'probe' })
+  return useMothershipQueueStore.getState().queues[chatId] !== undefined
+}
+
+/** A server answer the test releases when it chooses. */
+function deferredAnswer(value: unknown): () => void {
+  let answer!: () => void
+  mockRequestJson.mockReturnValue(
+    new Promise((resolve) => {
+      answer = () => resolve(value)
+    })
+  )
+  return answer
+}
+
+/** The options `useRestoreMothershipChat` hands to `useMutation` (the mock returns them). */
+interface RestoreMutation {
+  mutationFn: (chatId: string) => Promise<void>
+  onMutate?: (chatId: string) => { deleteSeen?: number }
+  onSuccess: (data: undefined, chatId: string, context?: { deleteSeen?: number }) => void
+}
+
+async function restore(chatId: string, whileInFlight: () => void = () => {}) {
+  const mutation = useRestoreMothershipChat() as unknown as RestoreMutation
+  const context = mutation.onMutate?.(chatId)
+  const answer = deferredAnswer({ success: true })
+  const done = mutation.mutationFn(chatId)
+  whileInFlight()
+  answer()
+  await done
+  mutation.onSuccess(undefined, chatId, context)
+}
+
+describe('lifting a chat delete', () => {
+  beforeEach(() => {
+    useMothershipQueueStore.getState().reset()
+    mockRequestJson.mockReset()
+  })
+
+  it('reopens a chat this tab saw deleted once the server returns it again', async () => {
+    useMothershipQueueStore.getState().clearChat(history.id)
+    mockRequestJson.mockResolvedValue({ chat: history })
+
+    await fetchMothershipChatHistory(history.id)
+
+    expect(takesSends(history.id)).toBe(true)
+  })
+
+  it('keeps the delete when the read returning the chat began before it', async () => {
+    const answer = deferredAnswer({ chat: history })
+    const read = fetchMothershipChatHistory(history.id)
+    useMothershipQueueStore.getState().clearChat(history.id)
+    answer()
+    await read
+
+    expect(takesSends(history.id)).toBe(false)
+  })
+
+  it('keeps a newer delete that lands while a read after an earlier one is in flight', async () => {
+    useMothershipQueueStore.getState().clearChat(history.id)
+    const answer = deferredAnswer({ chat: history })
+    const read = fetchMothershipChatHistory(history.id)
+    useMothershipQueueStore.getState().reopenChat(history.id)
+    useMothershipQueueStore.getState().clearChat(history.id)
+    answer()
+    await read
+
+    expect(takesSends(history.id)).toBe(false)
+  })
+
+  it('reopens a chat restored from Recently Deleted', async () => {
+    useMothershipQueueStore.getState().clearChat(history.id)
+
+    await restore(history.id)
+
+    expect(takesSends(history.id)).toBe(true)
+  })
+
+  it('keeps a delete that lands while the restore is in flight', async () => {
+    useMothershipQueueStore.getState().clearChat(history.id)
+
+    await restore(history.id, () => {
+      useMothershipQueueStore.getState().reopenChat(history.id)
+      useMothershipQueueStore.getState().clearChat(history.id)
+    })
+
+    expect(takesSends(history.id)).toBe(false)
+  })
+})

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

@@ -281,7 +281,7 @@ export function useOrganizationMothershipChats(
   })
 }
 
-export async function fetchMothershipChatHistory(
+async function readMothershipChatHistory(
   chatId: string,
   signal?: AbortSignal
 ): Promise<MothershipChatHistory> {
@@ -309,6 +309,22 @@ export async function fetchMothershipChatHistory(
   return parseChatHistory(await copilotRes.json())
 }
 
+/**
+ * Reads a chat from the server. A chat this tab saw deleted that the server
+ * returns again was restored, so it takes queued sends again. Only a read that
+ * began after the delete counts: one already in flight can return the chat from
+ * before it.
+ */
+export async function fetchMothershipChatHistory(
+  chatId: string,
+  signal?: AbortSignal
+): Promise<MothershipChatHistory> {
+  const deleteSeen = useMothershipQueueStore.getState().cleared[chatId]
+  const history = await readMothershipChatHistory(chatId, signal)
+  if (deleteSeen !== undefined) useMothershipQueueStore.getState().reopenChat(chatId, deleteSeen)
+  return history
+}
+
 export function mothershipChatHistoryQueryOptions(chatId: string | undefined) {
   return queryOptions({
     queryKey: mothershipChatKeys.detail(chatId),
@@ -365,6 +381,12 @@ export function useRestoreMothershipChat(owner?: MothershipChatOwner) {
   const queryClient = useQueryClient()
   return useMutation({
     mutationFn: restoreChat,
+    /** The delete this restore undoes; one that lands while it is in flight stays. */
+    onMutate: (chatId) => ({ deleteSeen: useMothershipQueueStore.getState().cleared[chatId] }),
+    onSuccess: (_data, chatId, context) => {
+      if (context?.deleteSeen === undefined) return
+      useMothershipQueueStore.getState().reopenChat(chatId, context.deleteSeen)
+    },
     onSettled: () => {
       queryClient.invalidateQueries({ queryKey: mothershipChatKeys.ownerLists(owner) })
     },

apps/sim/hooks/queries/mothership-chat-history-restore.test.ts

@@ -15,6 +15,7 @@ import {
   handleMothershipChatStatusEvent,
   resyncMothershipChatCaches,
 } from '@/hooks/use-mothership-chat-events'
+import { useMothershipQueueStore } from '@/stores/mothership-queue/store'
 
 describe('handleMothershipChatStatusEvent', () => {
   const queryClient = {
@@ -136,6 +137,27 @@ describe('handleMothershipChatStatusEvent', () => {
     expect(suspendTerminalScope).toHaveBeenCalledWith('chat-1')
   })
 
+  it('drops the queue of a chat deleted elsewhere and takes sends again once it is restored', () => {
+    useMothershipQueueStore.getState().reset()
+    const queued = { id: 'm1', content: 'follow-up' }
+    const publish = (type: 'deleted' | 'created') =>
+      handleMothershipChatStatusEvent(
+        queryClient,
+        'ws-1',
+        JSON.stringify({ chatId: 'chat-1', type, timestamp: Date.now() })
+      )
+
+    useMothershipQueueStore.getState().enqueue('chat-1', queued)
+    publish('deleted')
+    expect(useMothershipQueueStore.getState().queues['chat-1']).toBeUndefined()
+    useMothershipQueueStore.getState().enqueue('chat-1', queued)
+    expect(useMothershipQueueStore.getState().queues['chat-1']).toBeUndefined()
+
+    publish('created')
+    useMothershipQueueStore.getState().enqueue('chat-1', queued)
+    expect(useMothershipQueueStore.getState().queues['chat-1']?.map((m) => m.id)).toEqual(['m1'])
+  })
+
   it('keeps started task detail when a stale started stream is older than the active stream', () => {
     queryClient.getQueryData.mockReturnValue({
       id: 'chat-1',

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

@@ -10,6 +10,7 @@ import {
   type MothershipChatOwner,
   mothershipChatKeys,
 } from '@/hooks/queries/mothership-chats'
+import { useMothershipQueueStore } from '@/stores/mothership-queue/store'
 
 const logger = createLogger('MothershipChatEvents')
 
@@ -119,8 +120,12 @@ export function handleMothershipChatStatusEvent(
     // mutation would leave pages and PTYs running indefinitely.
     void suspendDesktopChatScopes(payload.chatId)
     queryClient.removeQueries({ queryKey: mothershipChatKeys.detail(payload.chatId) })
+    /** This tab's queue for the chat goes too, and no later send may bring it back. */
+    useMothershipQueueStore.getState().clearChat(payload.chatId)
     return
   }
+  /** A restore is published as `created`; the chat takes queued sends again. */
+  if (payload.type === 'created') useMothershipQueueStore.getState().reopenChat(payload.chatId)
   if (payload.type === 'renamed') {
     /**
      * The lists invalidated above carry the title every surface renders; the

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

@@ -88,5 +88,32 @@ describe('useMothershipQueueStore', () => {
       ])
       expect(useMothershipQueueStore.getState().queues['pending::abc']).toBeUndefined()
     })
+
+    it('lifts only the delete a restore saw, never a later one', () => {
+      useMothershipQueueStore.getState().clearChat('chat-X')
+      const seen = useMothershipQueueStore.getState().cleared['chat-X']
+      useMothershipQueueStore.getState().reopenChat('chat-X')
+      useMothershipQueueStore.getState().clearChat('chat-X')
+
+      useMothershipQueueStore.getState().reopenChat('chat-X', seen)
+      useMothershipQueueStore.getState().enqueue('chat-X', message('after-stale-restore'))
+      expect(useMothershipQueueStore.getState().queues['chat-X']).toBeUndefined()
+
+      const latest = useMothershipQueueStore.getState().cleared['chat-X']
+      useMothershipQueueStore.getState().reopenChat('chat-X', latest)
+      useMothershipQueueStore.getState().enqueue('chat-X', message('after-restore'))
+      expect(useMothershipQueueStore.getState().queues['chat-X']?.map((m) => m.id)).toEqual([
+        'after-restore',
+      ])
+    })
+
+    it('does not move a new chat surface queue into a chat deleted meanwhile', () => {
+      useMothershipQueueStore.getState().enqueue('pending::abc', message('pending-1'))
+      useMothershipQueueStore.getState().clearChat('chat-X')
+      useMothershipQueueStore.getState().migrate('pending::abc', 'chat-X')
+      const state = useMothershipQueueStore.getState()
+      expect(state.queues['chat-X']).toBeUndefined()
+      expect(state.queues['pending::abc']).toBeUndefined()
+    })
   })
 })

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

@@ -41,10 +41,13 @@ const sessionStorageAdapter = {
   },
 }
 
+/** Numbers each delete, so a restore or read can tell the delete it saw from a later one. */
+let deleteCount = 0
+
 const initialState = {
   queues: {} as Record<string, QueuedMothershipMessage[]>,
   editing: {} as Record<string, string>,
-  cleared: {} as Record<string, true>,
+  cleared: {} as Record<string, number>,
 }
 
 const omitKey = <V>(record: Record<string, V>, key: string): Record<string, V> => {
@@ -67,13 +70,15 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
         ...initialState,
 
         enqueue: (chatKey, message) =>
-          set((state) => ({
-            cleared: omitKey(state.cleared, chatKey),
-            queues: setQueueForChat(state.queues, chatKey, [
-              ...(state.queues[chatKey] ?? []),
-              message,
-            ]),
-          })),
+          set((state) => {
+            if (state.cleared[chatKey]) return state
+            return {
+              queues: setQueueForChat(state.queues, chatKey, [
+                ...(state.queues[chatKey] ?? []),
+                message,
+              ]),
+            }
+          }),
 
         insertAt: (chatKey, index, message) =>
           set((state) => {
@@ -99,6 +104,8 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
               retryRequired: _retry,
               heldUntilOnline: _held,
               heldSurface: _surface,
+              busyRetries: _busyRetries,
+              notBefore: _notBefore,
               ...rest
             } = next[index]
             next[index] = {
@@ -144,7 +151,8 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
             if (!fromQueue && fromEditing === undefined) return state
 
             const queues = omitKey(state.queues, fromKey)
-            if (fromQueue && fromQueue.length > 0) {
+            /** A chat deleted meanwhile takes nothing: its queue is gone with it. */
+            if (fromQueue && fromQueue.length > 0 && !state.cleared[toKey]) {
               // Merge defensively in case a stale bucket survived in
               // sessionStorage. FIFO: existing first, then the resolved stream.
               const existing = state.queues[toKey] ?? []
@@ -201,9 +209,17 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
           set((state) => ({
             queues: omitKey(state.queues, chatKey),
             editing: omitKey(state.editing, chatKey),
-            cleared: { ...state.cleared, [chatKey]: true },
+            cleared: { ...state.cleared, [chatKey]: ++deleteCount },
           })),
 
+        reopenChat: (chatKey, deleteToken) =>
+          set((state) => {
+            const current = state.cleared[chatKey]
+            if (current === undefined) return state
+            if (deleteToken !== undefined && deleteToken !== current) return state
+            return { cleared: omitKey(state.cleared, chatKey) }
+          }),
+
         reset: () => set(initialState),
       }),
       {

apps/sim/stores/mothership-queue/store.test.ts

@@ -24,6 +24,10 @@ export type QueuedMothershipMessage = QueuedMessage & {
    * mount: the next chatless surface for the same owner and workflow adopts it.
    */
   heldSurface?: string
+  /** Busy refusals so far; paces the next retry. */
+  busyRetries?: number
+  /** Epoch ms before which a busy-refused message is not sent again. */
+  notBefore?: number
   /**
    * Message id of a prior attempt at this send that an unmount cleanup
    * withdrew. Reused when the entry is dispatched so the server deduplicates
@@ -49,10 +53,11 @@ export interface MothershipQueueState {
   queues: Record<string, QueuedMothershipMessage[]>
   editing: Record<string, string>
   /**
-   * Chats cleared this session (deleted). A late restore of a send dispatched
-   * before the clear does not recreate their queue; a new enqueue lifts it.
+   * Chats cleared this session (deleted), each with the token of its latest
+   * delete. No write recreates their queue (a late restore, or a failed send
+   * handed back); restoring the chat lifts it.
    */
-  cleared: Record<string, true>
+  cleared: Record<string, number>
 
   enqueue: (chatKey: string, message: QueuedMothershipMessage) => void
   insertAt: (chatKey: string, index: number, message: QueuedMothershipMessage) => void
@@ -65,5 +70,10 @@ export interface MothershipQueueState {
   /** Moves the sends a dead chatless mount of `surface` held onto `toKey`. */
   adoptHeldSends: (toKey: string, surface: string) => void
   clearChat: (chatKey: string) => void
+  /**
+   * Lifts `cleared` for a restored chat. Given the delete token an operation
+   * saw when it began, lifts only that delete, never one that landed after it.
+   */
+  reopenChat: (chatKey: string, deleteToken?: number) => void
   reset: () => void
 }