fix(mothership): keep held and busy-refused sends exactly once across remounts

- A first message held offline on the new-chat page sat under that mount's
  queue key, which dies with the mount, so a reload or remount before the
  network returned stranded it. Held sends on a chatless surface now carry the
  surface they belong to, and the next chatless mount of that surface adopts
  them.
- Held sends are released for every chat when the browser comes back online,
  and on mount when it already is, so a send held in a chat the user is not
  viewing (or one whose `online` event fired with no surface mounted) still
  goes out.
- A send refused because the chat is busy is handed back only after the chat's
  running turn has been read, so the queue cannot redispatch it before that
  turn ends. A busy refusal that does not name the running turn no longer reads
  as a deduplicated send, which reconnected to a stream that never existed and
  lost the message.

Author: waleedlatif1

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

@@ -1935,14 +1935,18 @@ describe('useChat remount send recovery', () => {
      * The POST fails at the network layer until the network is back, and the
      * stream it would have opened does not exist.
      */
-    const network = { online: false }
+    const network = { online: false, acceptedPosts: 0 }
     function stubUnreachableSend() {
       network.online = false
+      network.acceptedPosts = 0
       vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
         const url = String(input)
         if (url === '/api/mothership/chat' && init?.method === 'POST') {
           state.postBodies.push(JSON.parse(String(init.body)))
-          if (network.online) return emptySseResponse()
+          if (network.online) {
+            network.acceptedPosts++
+            return emptySseResponse()
+          }
           throw new TypeError('Failed to fetch')
         }
         if (url.includes('/api/mothership/chat/stream')) {
@@ -2002,33 +2006,133 @@ describe('useChat remount send recovery', () => {
 
     /**
      * Another tab's turn holds the chat (this one missed its start). The server
-     * refuses the send naming that turn; the message must wait for it rather than
-     * render that turn's answer under itself and then vanish.
+     * refuses the send, naming that turn or, when its stream id is unreadable,
+     * nothing. The message must go out exactly once, under its id, after that turn
+     * ends: never rendered under the other turn's answer, never lost, never resent
+     * while the turn still runs.
      */
-    it('sends a message again after the turn that held the chat ends', async () => {
-      const history = idleHistory('chat-busy-elsewhere')
-      mockRequestJson.mockImplementation(() => Promise.resolve({ chat: history }))
-      vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
-        if (String(input) === '/api/mothership/chat' && init?.method === 'POST') {
-          state.postBodies.push(JSON.parse(String(init.body)))
-          if (state.postBodies.length === 1) {
-            return Response.json(
-              { error: 'A response is already running', activeStreamId: 'turn-from-another-tab' },
-              { status: 409 }
-            )
+    it.each([
+      ['names', 'turn-from-another-tab'],
+      ['does not name', undefined],
+    ] as const)(
+      'sends a message once, after the turn that held the chat ends, when the refusal %s it',
+      async (_names, refusalStreamId) => {
+        const history = idleHistory(`chat-busy-${refusalStreamId ?? 'unnamed'}`)
+        let otherTurnRunning = true
+        mockRequestJson.mockImplementation(() =>
+          Promise.resolve({
+            chat: {
+              ...history,
+              activeStreamId: otherTurnRunning ? 'turn-from-another-tab' : null,
+            },
+          })
+        )
+        vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
+          const url = String(input)
+          if (url === '/api/mothership/chat' && init?.method === 'POST') {
+            state.postBodies.push(JSON.parse(String(init.body)))
+            if (otherTurnRunning) {
+              return Response.json(
+                {
+                  error: 'A response is already in progress for this chat.',
+                  ...(refusalStreamId ? { activeStreamId: refusalStreamId } : {}),
+                },
+                { status: 409 }
+              )
+            }
+            return emptySseResponse()
           }
-          return emptySseResponse()
-        }
-        return fetchStub(input, init)
+          if (url.includes('/api/mothership/chat/stream')) {
+            if (url.includes('batch=true')) {
+              return Response.json({
+                success: true,
+                events: [],
+                status: otherTurnRunning ? 'streaming' : 'complete',
+              })
+            }
+            return emptySseResponse()
+          }
+          return fetchStub(input, init)
+        })
+        const { getResult } = renderUseChatInChat(history.id, history)
+
+        await act(async () => {
+          await getResult().sendMessage('Sent from the second tab')
+        })
+        await act(async () => {
+          await sleep(1500)
+        })
+        expect(state.postBodies).toHaveLength(1)
+        expect(useMothershipQueueStore.getState().queues[history.id]?.[0]?.content).toBe(
+          'Sent from the second tab'
+        )
+
+        otherTurnRunning = false
+        await waitFor(() => state.postBodies.length === 2, 5000)
+        await act(async () => {
+          await sleep(500)
+        })
+
+        expect(state.postBodies).toHaveLength(2)
+        expect(state.postBodies[1].message).toBe('Sent from the second tab')
+        expect(state.postBodies[1].userMessageId).toBe(state.postBodies[0].userMessageId)
+      }
+    )
+
+    /**
+     * A chatless surface keys its queue by mount, so a held first message would
+     * be stranded by a reload or remount before the network returns. The next
+     * chatless mount of the same surface adopts it.
+     */
+    it('carries a first message held offline over to the next new-chat surface', async () => {
+      stubUnreachableSend()
+      const first = renderUseChat()
+      await act(async () => {
+        await first.getResult().sendMessage('First message, sent offline')
       })
-      const { getResult } = renderUseChatInChat(history.id, history)
+      await waitFor(() => allQueuedMessages().some((message) => message.retryRequired === true))
+      first.unmount()
+
+      const second = renderUseChat()
+      await waitFor(() =>
+        second
+          .getResult()
+          .messageQueue.some((message) => message.content === 'First message, sent offline')
+      )
 
+      network.online = true
       await act(async () => {
-        await getResult().sendMessage('Sent from the second tab')
+        window.dispatchEvent(new Event('online'))
       })
+      await waitFor(() => network.acceptedPosts === 1)
+      await act(async () => {
+        await sleep(300)
+      })
+
+      expect(network.acceptedPosts).toBe(1)
+      expect(new Set(state.postBodies.map((body) => body.userMessageId)).size).toBe(1)
+      expect(state.postBodies.at(-1)?.message).toBe('First message, sent offline')
+    })
+
+    /** The `online` event can fire while no surface for the chat is mounted. */
+    it('sends a held message when its chat mounts after the network came back', async () => {
+      const history = idleHistory('chat-held-while-away')
+      mockRequestJson.mockImplementation(() => Promise.resolve({ chat: history }))
+      stubUnreachableSend()
+      const first = renderUseChatInChat(history.id, history)
+      await act(async () => {
+        await first.getResult().sendMessage('Held while I was elsewhere')
+      })
+      await waitFor(
+        () => useMothershipQueueStore.getState().queues[history.id]?.[0]?.retryRequired === true
+      )
+      first.unmount()
+
+      network.online = true
+      renderUseChatInChat(history.id, history)
       await waitFor(() => state.postBodies.length === 2)
 
-      expect(state.postBodies[1].message).toBe('Sent from the second tab')
+      expect(state.postBodies[1].message).toBe('Held while I was elsewhere')
       expect(state.postBodies[1].userMessageId).toBe(state.postBodies[0].userMessageId)
     })
   })

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

@@ -729,6 +729,8 @@ export function useChat(
   const pendingStopModeRef = useRef<StopGenerationMode | null>(null)
   const workflowIdRef = useRef(options?.workflowId)
   workflowIdRef.current = options?.workflowId
+  /** Identifies this chatless surface across mounts, for the sends it holds. */
+  const heldSendSurface = `${scopeKey}:${options?.workflowId ?? 'home'}`
   const onToolResultRef = useRef(options?.onToolResult)
   onToolResultRef.current = options?.onToolResult
   const onTitleUpdateRef = useRef(options?.onTitleUpdate)
@@ -3834,10 +3836,10 @@ export function useChat(
         if (!response.ok) {
           const errorData = await response.json().catch(() => ({}))
           if (response.status === 409) {
+            /* A deduplicated send always names itself; a chat busy with another
+               turn names that turn, or nothing when its stream id is unreadable. */
             const conflictStreamId =
-              typeof errorData.activeStreamId === 'string'
-                ? errorData.activeStreamId
-                : userMessageId
+              typeof errorData.activeStreamId === 'string' ? errorData.activeStreamId : undefined
             const supersededStreamId = queuedSendHandoff?.supersededStreamId ?? pendingStopStreamId
             if (supersededStreamId && conflictStreamId === supersededStreamId) {
               rollbackOptimisticSend()
@@ -3853,8 +3855,11 @@ export function useChat(
             }
             if (conflictStreamId !== userMessageId) {
               /* Another turn holds the chat: one started in another tab, or one this
-                 surface lost track of. This message was not admitted, so it waits in
-                 the queue behind that turn, which the chat now shows as running. */
+                 surface lost track of. This message was not admitted (the server
+                 released its id), so it goes back to the queue under the same id. The
+                 queue drains only while the chat is idle, so the chat's running turn is
+                 read before the message is handed back: the chat then attaches to that
+                 turn and the message goes out once, after it ends. */
               rollbackOptimisticSend()
               if (streamGenRef.current === gen) {
                 streamGenRef.current++
@@ -3865,13 +3870,15 @@ export function useChat(
               }
               if (requestChatId) {
                 const busyChatId = requestChatId
-                upsertChatHistory(busyChatId, (current) => ({
-                  ...current,
-                  activeStreamId: conflictStreamId,
-                }))
-                void queryClient.invalidateQueries({
-                  queryKey: mothershipChatKeys.detail(busyChatId),
-                })
+                if (conflictStreamId) {
+                  upsertChatHistory(busyChatId, (current) => ({
+                    ...current,
+                    activeStreamId: conflictStreamId,
+                  }))
+                }
+                await queryClient
+                  .refetchQueries({ queryKey: mothershipChatKeys.detail(busyChatId), exact: true })
+                  .catch(() => {})
               }
               return { userMessageId }
             }
@@ -4203,14 +4210,23 @@ export function useChat(
           options?.assistantSearch,
           options?.assistantSearchLevel
         ),
-        ...(result.unreachable ? { retryRequired: true, heldUntilOnline: true } : {}),
+        ...(result.unreachable
+          ? {
+              retryRequired: true,
+              heldUntilOnline: true,
+              ...(activeChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)
+                ? { heldSurface: heldSendSurface }
+                : {}),
+            }
+          : {}),
       })
     },
     [
       workspaceId,
       createQueuedMessage,
       startSendMessage,
       handOffWithdrawnSend,
+      heldSendSurface,
       hasPendingChatAdmission,
     ]
   )
@@ -4810,7 +4826,14 @@ export function useChat(
           ...dispatched,
           ...(retainedHandoff ? { queuedSendHandoff: retainedHandoff } : {}),
           retryRequired: !retriesOnItsOwn,
-          ...(withdrawn?.unreachable ? { heldUntilOnline: true } : {}),
+          ...(withdrawn?.unreachable
+            ? {
+                heldUntilOnline: true,
+                ...(dispatchChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)
+                  ? { heldSurface: heldSendSurface }
+                  : {}),
+              }
+            : {}),
           ...(withdrawnUserMessageId ? { resumeUserMessageId: withdrawnUserMessageId } : {}),
         })
       }
@@ -4865,7 +4888,7 @@ export function useChat(
         userRemovedDuringDispatch.delete(msg.id)
       }
     },
-    [startSendMessage, handOffWithdrawnSend]
+    [startSendMessage, handOffWithdrawnSend, heldSendSurface]
   )
 
   const runQueueDispatchLoop = useCallback(async () => {
@@ -5019,21 +5042,31 @@ export function useChat(
     }
   }, [])
 
-  /** A send held because the server could not be reached goes out once the browser is back online. */
+  /**
+   * Sends held because the server could not be reached go out once the browser is
+   * online: on the `online` event, and on mount in case it fired while no chat
+   * surface was listening. A chatless surface first adopts what a dead mount of the
+   * same surface held, since that mount's queue key died with it.
+   */
   useEffect(() => {
     if (typeof window === 'undefined') return
     const releaseHeldSends = () => {
-      const chatKey = chatKeyRef.current
-      const queue = useMothershipQueueStore.getState().queues[chatKey]
-      if (!queue?.some((message) => message.heldUntilOnline)) return
-      useMothershipQueueStore.getState().releaseHeldUntilOnline(chatKey)
-      if (!sendingRef.current && !pendingStopPromiseRef.current) {
+      useMothershipQueueStore.getState().releaseHeldUntilOnline()
+      if (
+        useMothershipQueueStore.getState().queues[chatKeyRef.current]?.length &&
+        !sendingRef.current &&
+        !pendingStopPromiseRef.current
+      ) {
         void enqueueQueueDispatchRef.current({ type: 'send_head' })
       }
     }
+    if (chatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) {
+      useMothershipQueueStore.getState().adoptHeldSends(chatKey, heldSendSurface)
+    }
+    if (navigator.onLine) releaseHeldSends()
     window.addEventListener('online', releaseHeldSends)
     return () => window.removeEventListener('online', releaseHeldSends)
-  }, [])
+  }, [chatKey, heldSendSurface])
 
   /** A recovered send already in history belongs to its accepted turn, even after Stop. */
   useEffect(() => {

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

@@ -94,6 +94,7 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
               resumeUserMessageId: _staleResume,
               retryRequired: _retry,
               heldUntilOnline: _held,
+              heldSurface: _surface,
               ...rest
             } = next[index]
             next[index] = {
@@ -143,7 +144,11 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
               // Merge defensively in case a stale bucket survived in
               // sessionStorage. FIFO: existing first, then the resolved stream.
               const existing = state.queues[toKey] ?? []
-              queues[toKey] = [...existing, ...fromQueue]
+              /** A chat-bound key is stable, so its messages no longer need a surface to adopt them. */
+              queues[toKey] = [
+                ...existing,
+                ...fromQueue.map(({ heldSurface: _surface, ...message }) => message),
+              ]
             }
             const editing = omitKey(state.editing, fromKey)
             if (fromEditing !== undefined) {
@@ -152,16 +157,40 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
             return { queues, editing }
           }),
 
-        releaseHeldUntilOnline: (chatKey) =>
+        releaseHeldUntilOnline: () =>
           set((state) => {
-            const current = state.queues[chatKey] ?? []
-            if (!current.some((m) => m.heldUntilOnline)) return state
-            const next = current.map((message) => {
-              if (!message.heldUntilOnline) return message
-              const { retryRequired: _retry, heldUntilOnline: _held, ...rest } = message
-              return rest
-            })
-            return { queues: setQueueForChat(state.queues, chatKey, next) }
+            let released = false
+            const queues: Record<string, QueuedMothershipMessage[]> = {}
+            for (const [chatKey, queue] of Object.entries(state.queues)) {
+              queues[chatKey] = queue.map((message) => {
+                if (!message.heldUntilOnline) return message
+                released = true
+                const { retryRequired: _retry, heldUntilOnline: _held, ...rest } = message
+                return rest
+              })
+            }
+            return released ? { queues } : state
+          }),
+
+        adoptHeldSends: (toKey, surface) =>
+          set((state) => {
+            const adopted: QueuedMothershipMessage[] = []
+            let queues = state.queues
+            for (const [chatKey, queue] of Object.entries(state.queues)) {
+              if (chatKey === toKey) continue
+              const held = queue.filter((message) => message.heldSurface === surface)
+              if (held.length === 0) continue
+              adopted.push(...held)
+              queues = setQueueForChat(
+                queues,
+                chatKey,
+                queue.filter((message) => message.heldSurface !== surface)
+              )
+            }
+            if (adopted.length === 0) return state
+            return {
+              queues: setQueueForChat(queues, toKey, [...(queues[toKey] ?? []), ...adopted]),
+            }
           }),
 
         clearChat: (chatKey) =>