fix(mothership): keep a message the server never admitted instead of dropping it

Two sends were lost without an error:

- A send whose POST got no response (offline, Wi-Fi drop, waking a laptop)
  reconnected to the stream it would have opened. That stream does not exist,
  so the 404 read as "finished", the turn finalized as a success, and the
  refetched transcript no longer held the message. A queued follow-up was lost
  the same way, since it had already left the queue.
- A send refused with 409 because another turn held the chat (started in
  another tab, or one this surface lost track of) reconnected to that turn
  under the new message's bubble, then vanished when it finished.

Both now hand the message back under its id. An unreachable send is held in
the queue, so it is not redispatched into the same failure, and goes out when
the browser is back online (or when the user sends it). A send that found the
chat busy waits in the queue behind that turn, which the chat shows as
running, and goes out when it ends. Reusing the id keeps a retry deduplicated
if the server did admit the first attempt.

Author: waleedlatif1

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

@@ -1921,6 +1921,118 @@ describe('useChat remount send recovery', () => {
     })
   })
 
+  describe('a send the server never admitted', () => {
+    const idleHistory = (id: string): MothershipChatHistory => ({
+      id,
+      mode: 'agent',
+      title: 'Not admitted',
+      messages: [],
+      activeStreamId: null,
+      resources: [],
+    })
+
+    /**
+     * 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 }
+    function stubUnreachableSend() {
+      network.online = false
+      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()
+          throw new TypeError('Failed to fetch')
+        }
+        if (url.includes('/api/mothership/chat/stream')) {
+          return Response.json({ error: 'Stream not found' }, { status: 404 })
+        }
+        return fetchStub(input, init)
+      })
+    }
+
+    it('holds a message sent while offline and sends it under the same id once back online', async () => {
+      const history = idleHistory('chat-offline-send')
+      mockRequestJson.mockImplementation(() => Promise.resolve({ chat: history }))
+      stubUnreachableSend()
+      const { getResult } = renderUseChatInChat(history.id, history)
+
+      await act(async () => {
+        await getResult().sendMessage('Written while offline')
+      })
+      await waitFor(() => !getResult().isSending)
+
+      const queued = useMothershipQueueStore.getState().queues[history.id] ?? []
+      expect(queued.map((message) => message.content)).toEqual(['Written while offline'])
+      expect(queued[0].retryRequired).toBe(true)
+      expect(queued[0].resumeUserMessageId).toBe(state.postBodies[0].userMessageId)
+      expect(getResult().error).not.toBeNull()
+      expect(state.postBodies).toHaveLength(1)
+
+      network.online = true
+      await act(async () => {
+        window.dispatchEvent(new Event('online'))
+      })
+      await waitFor(() => state.postBodies.length === 2)
+
+      expect(state.postBodies[1].message).toBe('Written while offline')
+      expect(state.postBodies[1].userMessageId).toBe(state.postBodies[0].userMessageId)
+    })
+
+    it('keeps a queued follow-up whose dispatch could not reach the server', async () => {
+      const history = idleHistory('chat-offline-queue')
+      mockRequestJson.mockImplementation(() => Promise.resolve({ chat: history }))
+      stubUnreachableSend()
+      useMothershipQueueStore
+        .getState()
+        .enqueue(history.id, { id: 'queued-follow-up', content: 'Queued before the drop' })
+      renderUseChatInChat(history.id, history)
+
+      await waitFor(() => state.postBodies.length === 1)
+      await waitFor(
+        () => useMothershipQueueStore.getState().queues[history.id]?.[0]?.retryRequired === true
+      )
+
+      const queued = useMothershipQueueStore.getState().queues[history.id] ?? []
+      expect(queued.map((message) => message.content)).toEqual(['Queued before the drop'])
+      expect(queued[0].resumeUserMessageId).toBe(state.postBodies[0].userMessageId)
+      expect(state.postBodies).toHaveLength(1)
+    })
+
+    /**
+     * 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.
+     */
+    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 }
+            )
+          }
+          return emptySseResponse()
+        }
+        return fetchStub(input, init)
+      })
+      const { getResult } = renderUseChatInChat(history.id, history)
+
+      await act(async () => {
+        await getResult().sendMessage('Sent from the second tab')
+      })
+      await waitFor(() => state.postBodies.length === 2)
+
+      expect(state.postBodies[1].message).toBe('Sent from the second tab')
+      expect(state.postBodies[1].userMessageId).toBe(state.postBodies[0].userMessageId)
+    })
+  })
+
   /**
    * A withdrawn send belongs to the chat it was sent to. The cross-surface
    * lanes deliver to whatever chat is mounted next, so routing a chat-bound

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

@@ -205,13 +205,24 @@ interface FinalizeOptions {
   streamTerminal?: boolean
 }
 
+/**
+ * A send handed back to the caller instead of rendered. `userMessageId` is what
+ * a retry reuses so the server deduplicates the two attempts. An `unreachable`
+ * send is held in the queue until the browser is back online or the user sends
+ * it: dispatching it again at once would fail the same way.
+ */
+interface WithdrawnSendResult {
+  userMessageId: string
+  unreachable?: boolean
+}
+
 /**
  * `true` when the send owns the transcript (rendered, or handed to reconnect),
  * `false` when the caller should restore the queue entry, and the object form
- * when an unmount cleanup withdrew it — `userMessageId` is what a retry reuses
- * so the server deduplicates the two attempts.
+ * when the send was withdrawn: by an unmount cleanup, because it never reached
+ * the server, or because another turn held the chat.
  */
-type StartSendMessageResult = boolean | { userMessageId: string }
+type StartSendMessageResult = boolean | WithdrawnSendResult
 
 interface StartSendMessageOptions {
   /** Awaited before dispatch. Defaults to the hook's in-flight stop, if any. */
@@ -3834,6 +3845,30 @@ export function useChat(
               setError('Previous response is still shutting down; queued message was restored.')
               return false
             }
+            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. */
+              rollbackOptimisticSend()
+              if (streamGenRef.current === gen) {
+                streamGenRef.current++
+                abortController.abort('send_conflict:chat_busy')
+                abortControllerRef.current = null
+                clearActiveTurn()
+                setTransportIdle()
+              }
+              if (requestChatId) {
+                const busyChatId = requestChatId
+                upsertChatHistory(busyChatId, (current) => ({
+                  ...current,
+                  activeStreamId: conflictStreamId,
+                }))
+                void queryClient.invalidateQueries({
+                  queryKey: mothershipChatKeys.detail(busyChatId),
+                })
+              }
+              return { userMessageId }
+            }
             /* A send deduplicated against an earlier attempt comes back naming
                the chat that attempt opened. Adopting it here spares a chatless
                surface the stream-to-chat lookup and puts the user in the right
@@ -3953,6 +3988,27 @@ export function useChat(
           return consumedByTranscript
         }
 
+        if (!sendReachedServer) {
+          /* The POST got no answer, so nothing shows the server admitted it, and a
+             stream that was never opened reads as finished. Hand the message back
+             under the same id: if the server did admit it, the retry deduplicates
+             and the chat's own recovery shows the running turn. */
+          rollbackOptimisticSend()
+          if (gen !== undefined && streamGenRef.current === gen) {
+            streamGenRef.current++
+            admission?.controller.abort('send_unreachable:no_response')
+            abortControllerRef.current = null
+            clearActiveTurn()
+            setTransportIdle()
+          }
+          setError(
+            err instanceof TypeError
+              ? 'Message not sent: Sim could not be reached. It will send when you are back online.'
+              : getErrorMessage(err, 'Failed to send message')
+          )
+          return { userMessageId, unreachable: true }
+        }
+
         const activeStreamId = streamIdRef.current
         if (activeStreamId && gen !== undefined && streamGenRef.current === gen) {
           const succeeded = await retryReconnect({
@@ -4110,11 +4166,12 @@ export function useChat(
       const result = await startSendMessage(message, fileAttachments, contexts, options)
       if (typeof result !== 'object') return
 
-      /* An unmount cleanup withdrew the send. A chat-bound key is the stable
-         chat id, so re-queueing under the key this was sent to is the durable
-         retry — and keeps the message in that chat rather than following the
-         user into whichever one they opened next. Only a chatless surface,
-         whose key dies with the mount, goes to the cross-surface lanes. */
+      /* The send was withdrawn. A chat-bound key is the stable chat id, so
+         re-queueing under the key this was sent to is the durable retry, and
+         keeps the message in that chat rather than following the user into
+         whichever one they opened next. Only a send an unmount withdrew from a
+         chatless surface, whose key dies with the mount, goes to the
+         cross-surface lanes. */
       const withdrawn = {
         content: message,
         fileAttachments,
@@ -4126,24 +4183,22 @@ export function useChat(
           ? { assistantSearchLevel: options?.assistantSearchLevel }
           : {}),
       }
-      if (activeChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) {
+      if (!result.unreachable && activeChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) {
         handOffWithdrawnSend(withdrawn)
         return
       }
-      useMothershipQueueStore
-        .getState()
-        .enqueue(
-          activeChatKey,
-          createQueuedMessage(
-            message,
-            fileAttachments,
-            contexts,
-            result.userMessageId,
-            options?.requestMode,
-            options?.assistantSearch,
-            options?.assistantSearchLevel
-          )
-        )
+      useMothershipQueueStore.getState().enqueue(activeChatKey, {
+        ...createQueuedMessage(
+          message,
+          fileAttachments,
+          contexts,
+          result.userMessageId,
+          options?.requestMode,
+          options?.assistantSearch,
+          options?.assistantSearchLevel
+        ),
+        ...(result.unreachable ? { retryRequired: true, heldUntilOnline: true } : {}),
+      })
     },
     [
       workspaceId,
@@ -4696,9 +4751,10 @@ export function useChat(
       let dispatched = msg
       const restoreQueuedMessage = (
         handoff?: QueuedSendHandoffSeed,
-        withdrawnUserMessageId?: string
+        withdrawn?: WithdrawnSendResult
       ) => {
-        const withdrawnByCleanup = withdrawnUserMessageId !== undefined
+        const withdrawnUserMessageId = withdrawn?.userMessageId
+        const retriesOnItsOwn = withdrawn !== undefined && !withdrawn.unreachable
         const savedHandoff = readQueuedSendHandoffState()
         const retainedHandoff =
           savedHandoff?.id === msg.id
@@ -4714,7 +4770,7 @@ export function useChat(
         if (!removedFromQueue) {
           return
         }
-        if (options.epoch !== queueDispatchEpochRef.current && !withdrawnByCleanup) {
+        if (options.epoch !== queueDispatchEpochRef.current && !retriesOnItsOwn) {
           return
         }
         // If the user explicitly removed this message during dispatch, honor
@@ -4727,7 +4783,7 @@ export function useChat(
            restore would strand this under the dead instance's key — hand it to
            the next surface instead. A chat-bound key is the stable chat id, so
            the queue itself is the durable retry. */
-        if (withdrawnByCleanup && dispatchChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) {
+        if (withdrawn && retriesOnItsOwn && dispatchChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) {
           clearQueuedSendHandoffState(msg.id)
           handOffWithdrawnSend({
             content: dispatched.content,
@@ -4738,7 +4794,7 @@ export function useChat(
             ...(dispatched.assistantSearchLevel !== undefined
               ? { assistantSearchLevel: dispatched.assistantSearchLevel }
               : {}),
-            userMessageId: withdrawnUserMessageId,
+            userMessageId: withdrawn.userMessageId,
           })
           return
         }
@@ -4747,7 +4803,8 @@ export function useChat(
         useMothershipQueueStore.getState().insertAt(dispatchChatKey, originalIndex, {
           ...dispatched,
           ...(retainedHandoff ? { queuedSendHandoff: retainedHandoff } : {}),
-          retryRequired: !withdrawnByCleanup,
+          retryRequired: !retriesOnItsOwn,
+          ...(withdrawn?.unreachable ? { heldUntilOnline: true } : {}),
           ...(withdrawnUserMessageId ? { resumeUserMessageId: withdrawnUserMessageId } : {}),
         })
       }
@@ -4791,7 +4848,7 @@ export function useChat(
         if (sendResult !== true) {
           restoreQueuedMessage(
             activeQueuedSendHandoff,
-            typeof sendResult === 'object' ? sendResult.userMessageId : undefined
+            typeof sendResult === 'object' ? sendResult : undefined
           )
         }
       } catch {
@@ -4956,6 +5013,22 @@ export function useChat(
     }
   }, [])
 
+  /** A send held because the server could not be reached goes out once the browser is back online. */
+  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) {
+        void enqueueQueueDispatchRef.current({ type: 'send_head' })
+      }
+    }
+    window.addEventListener('online', releaseHeldSends)
+    return () => window.removeEventListener('online', releaseHeldSends)
+  }, [])
+
   /** A recovered send already in history belongs to its accepted turn, even after Stop. */
   useEffect(() => {
     if (!chatHistory || chatHistory.id !== chatKeyRef.current) return

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

@@ -93,6 +93,7 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
               queuedSendHandoff,
               resumeUserMessageId: _staleResume,
               retryRequired: _retry,
+              heldUntilOnline: _held,
               ...rest
             } = next[index]
             next[index] = {
@@ -151,6 +152,18 @@ export const useMothershipQueueStore = create<MothershipQueueState>()(
             return { queues, editing }
           }),
 
+        releaseHeldUntilOnline: (chatKey) =>
+          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) }
+          }),
+
         clearChat: (chatKey) =>
           set((state) => ({
             queues: omitKey(state.queues, chatKey),

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

@@ -13,6 +13,11 @@ export type QueuedMothershipMessage = QueuedMessage & {
   queuedSendHandoff?: QueuedSendHandoffSeed
   /** A failed dispatch remains queued until the user retries or edits it. */
   retryRequired?: boolean
+  /**
+   * The failed dispatch never reached the server, so the browser coming back
+   * online releases it for dispatch too.
+   */
+  heldUntilOnline?: boolean
   /**
    * Message id of a prior attempt at this send that an unmount cleanup
    * withdrew. Reused when the entry is dispatched so the server deduplicates
@@ -44,6 +49,8 @@ export interface MothershipQueueState {
   remove: (chatKey: string, id: string) => void
   setEditing: (chatKey: string, id: string | null) => void
   migrate: (fromKey: string, toKey: string) => void
+  /** Releases the chat's sends held for the network for dispatch. */
+  releaseHeldUntilOnline: (chatKey: string) => void
   clearChat: (chatKey: string) => void
   reset: () => void
 }