refactor(mothership): one re-queue policy and one send payload, and keep a Stop-failed send waiting for the user (#8740)

* refactor(mothership): one re-queue policy for a withdrawn send

A withdrawn send's outcome is one reason (withdrawn, offline, unreachable,
busy, stop-failed) instead of five exclusive booleans, and a pure
requeuedFields(reason, previousAttempts, chatlessSurface) is the policy both
re-queue sites apply. The restore strips the hold, retry and surface fields
an earlier outcome left before applying it, so a stale heldUntilOnline can no
longer let the browser coming online send a message waiting for the user.

* refactor(mothership): pass a chat send as one SendPayload

A send's content, attachments, contexts, mode and search settings travel as
one SendPayload instead of positional parameters (createQueuedMessage,
sendMothershipMessage) and about ten hand-spread copies; sendPayload() is the
one place that drops the fields a send does not set.

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/home/home.tsx

@@ -25,6 +25,7 @@ import { ChatResourcePanel } from '@/app/workspace/[workspaceId]/home/components
 import { RESOURCE_HEADER_CLASSES } from '@/app/workspace/[workspaceId]/home/components/mothership-view/components/resource-tabs/resource-tab-controls'
 import { SuggestedActions } from '@/app/workspace/[workspaceId]/home/components/suggested-actions'
 import { HomeFallback } from '@/app/workspace/[workspaceId]/home/home-fallback'
+import { sendPayload } from '@/app/workspace/[workspaceId]/home/hooks/send-queue-policy'
 import {
   useChatResourcePanel,
   useResourcePanelController,
@@ -242,13 +243,13 @@ function HomeContent({ chatId, userName, userId }: HomeProps) {
       if (!detail?.message) return
       e.preventDefault()
       prepareResourceViewForAgentTurn()
-      sendMessage(detail.message, detail.fileAttachments, detail.contexts, {
+      const { content, fileAttachments, contexts, ...sendOptions } = sendPayload({
+        ...detail,
+        content: detail.message,
+      })
+      sendMessage(content, fileAttachments, contexts, {
+        ...sendOptions,
         ...(detail.resumeUserMessageId ? { resumeUserMessageId: detail.resumeUserMessageId } : {}),
-        ...(detail.requestMode ? { requestMode: detail.requestMode } : {}),
-        ...(detail.assistantSearch ? { assistantSearch: detail.assistantSearch } : {}),
-        ...(detail.assistantSearchLevel !== undefined
-          ? { assistantSearchLevel: detail.assistantSearchLevel }
-          : {}),
       })
     }
     window.addEventListener(MOTHERSHIP_SEND_MESSAGE_EVENT, handler)
@@ -279,15 +280,15 @@ function HomeContent({ chatId, userName, userId }: HomeProps) {
     if (!handoff) return
     if (handoff.message) {
       prepareResourceViewForAgentTurn()
-      sendMessage(handoff.message, handoff.fileAttachments, handoff.contexts, {
+      const { content, fileAttachments, contexts, ...sendOptions } = sendPayload({
+        ...handoff,
+        content: handoff.message,
+      })
+      sendMessage(content, fileAttachments, contexts, {
+        ...sendOptions,
         ...(handoff.resumeUserMessageId
           ? { resumeUserMessageId: handoff.resumeUserMessageId }
           : {}),
-        ...(handoff.requestMode ? { requestMode: handoff.requestMode } : {}),
-        ...(handoff.assistantSearch ? { assistantSearch: handoff.assistantSearch } : {}),
-        ...(handoff.assistantSearchLevel !== undefined
-          ? { assistantSearchLevel: handoff.assistantSearchLevel }
-          : {}),
       })
       return
     }

apps/sim/app/workspace/[workspaceId]/home/hooks/send-queue-policy.test.ts

@@ -0,0 +1,73 @@
+import { describe, expect, it } from 'vitest'
+import {
+  requeuedFields,
+  sendPayload,
+  withoutRequeueFields,
+} from '@/app/workspace/[workspaceId]/home/hooks/send-queue-policy'
+
+describe('requeuedFields', () => {
+  it('holds an offline send for the network, on its chatless surface', () => {
+    expect(requeuedFields('offline', 0, 'ws-1:home')).toEqual({
+      retryRequired: true,
+      heldUntilOnline: true,
+      heldSurface: 'ws-1:home',
+    })
+  })
+
+  it.each(['unreachable', 'busy'] as const)('retries a %s send on a growing delay', (reason) => {
+    const before = Date.now()
+    const fields = requeuedFields(reason, 2, undefined)
+
+    expect(fields.sendRetries).toBe(3)
+    expect(fields.notBefore).toBeGreaterThan(before)
+    expect(fields.retryRequired).toBeUndefined()
+    expect(fields.heldSurface).toBeUndefined()
+  })
+
+  it.each(['stop-failed', 'failed'] as const)(
+    'leaves a %s send for the user, on any surface',
+    (reason) => {
+      expect(requeuedFields(reason, 4, 'ws-1:home')).toEqual({ retryRequired: true })
+    }
+  )
+
+  it('sends a withdrawn message again as soon as the queue drains', () => {
+    expect(requeuedFields('withdrawn', 1, undefined)).toEqual({})
+  })
+})
+
+describe('withoutRequeueFields', () => {
+  it('drops every hold, retry and surface field and keeps the message itself', () => {
+    expect(
+      withoutRequeueFields({
+        id: 'm1',
+        content: 'hello',
+        resumeUserMessageId: 'attempt-1',
+        admissionUnknown: true,
+        retryRequired: true,
+        heldUntilOnline: true,
+        sendRetries: 2,
+        notBefore: 123,
+        heldSurface: 'ws-1:home',
+      })
+    ).toEqual({
+      id: 'm1',
+      content: 'hello',
+      resumeUserMessageId: 'attempt-1',
+      admissionUnknown: true,
+    })
+  })
+})
+
+describe('sendPayload', () => {
+  it('keeps only the fields a send sets', () => {
+    expect(
+      sendPayload({
+        content: 'hello',
+        fileAttachments: undefined,
+        requestMode: 'assistant',
+        assistantSearchLevel: undefined,
+      })
+    ).toEqual({ content: 'hello', requestMode: 'assistant' })
+  })
+})

apps/sim/app/workspace/[workspaceId]/home/hooks/send-queue-policy.ts

@@ -0,0 +1,98 @@
+import { backoffWithJitter } from '@sim/utils/retry'
+import type { SendPayload } from '@/app/workspace/[workspaceId]/home/types'
+import type { QueuedMothershipMessage, ScheduledRetry } from '@/stores/mothership-queue/types'
+
+/**
+ * Why a send came back to its caller instead of going out:
+ * - `withdrawn`: an unmount cleanup withdrew it before the server answered;
+ * - `offline`: its POST got no answer and the browser is offline;
+ * - `unreachable`: its POST failed at the network while the browser reports
+ *   itself online (a dropped connection), so no `online` event will come;
+ * - `busy`: the server refused it because another turn held the chat;
+ * - `stop-failed`: the Stop it waited on did not settle, so it was not sent.
+ */
+export type WithdrawalReason = 'withdrawn' | 'offline' | 'unreachable' | 'busy' | 'stop-failed'
+
+/** A withdrawal, or `failed`: the send failed outright and the user decides what next. */
+export type RequeueReason = WithdrawalReason | 'failed'
+
+/** The queue fields that say when, and on which surface, a re-queued message goes out. */
+type RequeueFields = Pick<
+  QueuedMothershipMessage,
+  'retryRequired' | 'heldUntilOnline' | 'sendRetries' | 'notBefore' | 'heldSurface'
+>
+
+const SEND_RETRY_BASE_MS = 1_000
+const SEND_RETRY_MAX_MS = 30_000
+
+/** Queue fields for the `attempt`th automatic retry of a message: when it may be sent again. */
+export function sendRetry(attempt: number): ScheduledRetry {
+  return {
+    sendRetries: attempt,
+    notBefore:
+      Date.now() +
+      backoffWithJitter(attempt, null, { baseMs: SEND_RETRY_BASE_MS, maxMs: SEND_RETRY_MAX_MS }),
+  }
+}
+
+/**
+ * The one re-queue policy, for a message going back to its queue:
+ * - `offline` waits for the browser to come back online (or the user);
+ * - `unreachable` and `busy` retry on a growing delay after `previousAttempts`;
+ * - `stop-failed` and `failed` wait for the user;
+ * - `withdrawn` goes out again as soon as the queue drains.
+ *
+ * A message held by a chatless surface carries that surface (`chatlessSurface`),
+ * whose queue key dies with its mount, so the next mount of it adopts the
+ * message. Only sends that wait on the network or the server are held that way.
+ */
+export function requeuedFields(
+  reason: RequeueReason,
+  previousAttempts: number,
+  chatlessSurface: string | undefined
+): RequeueFields {
+  const surface = chatlessSurface ? { heldSurface: chatlessSurface } : {}
+  switch (reason) {
+    case 'offline':
+      return { retryRequired: true, heldUntilOnline: true, ...surface }
+    case 'unreachable':
+    case 'busy':
+      return { ...sendRetry(previousAttempts + 1), ...surface }
+    case 'stop-failed':
+    case 'failed':
+      return { retryRequired: true }
+    case 'withdrawn':
+      return {}
+  }
+}
+
+/**
+ * A queue entry without the fields an earlier outcome set, so a re-queue applies
+ * only the policy for the outcome it is handling. A stale `heldUntilOnline`, for
+ * one, would let the browser coming online send a message waiting for the user.
+ */
+export function withoutRequeueFields(entry: QueuedMothershipMessage): QueuedMothershipMessage {
+  const {
+    retryRequired: _retryRequired,
+    heldUntilOnline: _heldUntilOnline,
+    sendRetries: _sendRetries,
+    notBefore: _notBefore,
+    heldSurface: _heldSurface,
+    ...rest
+  } = entry
+  return rest
+}
+
+/** The payload of a send, without fields it does not set. */
+export function sendPayload(source: SendPayload): SendPayload {
+  return {
+    content: source.content,
+    ...(source.fileAttachments ? { fileAttachments: source.fileAttachments } : {}),
+    ...(source.contexts ? { contexts: source.contexts } : {}),
+    ...(source.requestMode ? { requestMode: source.requestMode } : {}),
+    ...(source.assistantSearch ? { assistantSearch: source.assistantSearch } : {}),
+    ...(source.assistantSearchLevel !== undefined
+      ? { assistantSearchLevel: source.assistantSearchLevel }
+      : {}),
+  }
+}

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

@@ -2341,6 +2341,44 @@ describe('useChat remount send recovery', () => {
    * settle sends nothing. That says nothing about the earlier attempt the
    * message resumes, so it must stay uneditable.
    */
+  /**
+   * A message held for the network that the user then sends by hand, over a turn
+   * whose Stop does not settle, goes back waiting for the user. The hold it had
+   * before must not outlive that: the browser coming online must not send it.
+   */
+  it('keeps a Send-now whose Stop failed waiting for the user when the browser comes online', async () => {
+    state.abortSettlements = [false, false, false, false]
+    const { getResult } = renderUseChatInChat('chat-a')
+    await act(async () => {
+      void getResult().sendMessage('Original request')
+    })
+    await waitFor(() => state.postBodies.length === 1 && getResult().isSending)
+    useMothershipQueueStore.getState().enqueue('chat-a', {
+      id: 'held-offline',
+      content: 'written while offline',
+      resumeUserMessageId: 'offline-attempt',
+      admissionUnknown: true,
+      retryRequired: true,
+      heldUntilOnline: true,
+    })
+
+    await act(async () => {
+      await getResult()
+        .sendNow('held-offline')
+        .catch(() => {})
+      await sleep(200)
+    })
+    await act(async () => {
+      window.dispatchEvent(new Event('online'))
+      await sleep(100)
+    })
+
+    expect(state.postBodies).toHaveLength(1)
+    const queued = useMothershipQueueStore.getState().queues['chat-a']?.[0]
+    expect(queued).toMatchObject({ id: 'held-offline', retryRequired: true })
+    expect(queued?.heldUntilOnline).toBeUndefined()
+  })
+
   it('keeps a resumed message uneditable when its Send-now Stop does not settle', async () => {
     state.abortSettlements = [false, false, false, false]
     const { getResult } = renderUseChatInChat('chat-a')