fix(mothership): a pure resend verdict, and Send-now drops a message the server already has (#8744)

* refactor(mothership): a pure resend verdict, one discard helper, and Send-now honours it

- resendVerdict(entry, history | null) returns send, wait or drop, and the
  queue drain acts on it, instead of a boolean check with side effects
- discardQueuedSend replaces the three copies of clear handoff, clear claim,
  remove
- Send-now checks the history too and drops a message the server already
  accepted instead of resending it (G5); a history it cannot read does not
  hold back a send the user asked for
- the own-id conflict test reaches that branch through a restored Send-now,
  since a message the history shows accepted is now dropped first

* test(mothership): lock the queue upgrade path and tighten held waits

- rehydrate rows for the production shape (retryRequired: false, no hold)
  and for a queue already in the new shape
- the offline-hold waits check hold === 'online' instead of any hold
- the one-lookup comment names where the one-hop invariant is enforced

* fix(mothership): don't stop the running turn for a Send-now gone during its history read

Send-now reads the chat's history before stopping the running turn. A
message removed, edited or dispatched in that time, or a view that moved on,
still stopped the turn. It now re-reads the queue and checks the chat and
mount are current before going on.

* test(mothership): cover Send-now's re-checks after its history read

Send-now no longer stops a turn in a chat the user moved to during the
read, or from a surface that unmounted during it, and leaves a message
the drain dispatched meanwhile to that dispatch. applyHeldResend is
renamed applyResendVerdict.

Author: waleedlatif1

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

@@ -1,9 +1,11 @@
 import { describe, expect, it } from 'vitest'
 import {
   requeuedFields,
+  resendVerdict,
   sendPayload,
   withoutRequeueFields,
 } from '@/app/workspace/[workspaceId]/home/hooks/send-queue-policy'
+import type { MothershipChatHistory } from '@/hooks/queries/mothership-chats'
 
 describe('requeuedFields', () => {
   it('holds an offline send for the network, on its chatless surface', () => {
@@ -72,3 +74,44 @@ describe('sendPayload', () => {
     ).toEqual({ content: 'hello', requestMode: 'assistant' })
   })
 })
+
+describe('resendVerdict', () => {
+  const history = (accepted: string[] = [], activeStreamId: string | null = null) =>
+    ({
+      id: 'chat-A',
+      mode: 'agent',
+      title: 'A',
+      messages: accepted.map((id) => ({
+        id,
+        role: 'user' as const,
+        content: 'sent',
+        timestamp: new Date(0).toISOString(),
+      })),
+      activeStreamId,
+      resources: [],
+    }) satisfies MothershipChatHistory
+  const resumed = {
+    id: 'm1',
+    content: 'hello',
+    resumeUserMessageId: 'attempt-1',
+    admissionUnknown: true,
+  }
+
+  it('sends a message the server cannot already hold, without reading history', () => {
+    expect(resendVerdict({ id: 'm1', content: 'hello' }, null)).toBe('send')
+    expect(resendVerdict({ ...resumed, admissionUnknown: false }, null)).toBe('send')
+  })
+
+  it('drops a message the history shows accepted, as a message or as the running turn', () => {
+    expect(resendVerdict(resumed, history(['attempt-1']))).toBe('drop')
+    expect(resendVerdict(resumed, history([], 'attempt-1'))).toBe('drop')
+  })
+
+  it('sends a message the history does not show', () => {
+    expect(resendVerdict(resumed, history(['other']))).toBe('send')
+  })
+
+  it('waits when the history could not be read', () => {
+    expect(resendVerdict(resumed, null)).toBe('wait')
+  })
+})

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

@@ -1,5 +1,7 @@
 import { backoffWithJitter } from '@sim/utils/retry'
 import type { SendPayload } from '@/app/workspace/[workspaceId]/home/types'
+import type { MothershipChatHistory } from '@/hooks/queries/mothership-chats'
+import { reusedRequestId } from '@/stores/mothership-queue/store'
 import type { QueuedMothershipMessage, SendRetry } from '@/stores/mothership-queue/types'
 
 /**
@@ -87,3 +89,40 @@ export function sendPayload(source: SendPayload): SendPayload {
       : {}),
   }
 }
+
+/** Ids of the sends a chat's history shows the server accepted: its user messages and running turn. */
+export function acceptedMessageIds(history: MothershipChatHistory): Set<string> {
+  const ids = new Set(
+    history.messages.filter((message) => message.role === 'user').map((message) => message.id)
+  )
+  if (history.activeStreamId) ids.add(history.activeStreamId)
+  return ids
+}
+
+/**
+ * Whether a queued message must be checked against its chat's history before it
+ * goes out: it may already be a turn on the server, under the id it reuses.
+ */
+export function needsResendCheck(entry: QueuedMothershipMessage): boolean {
+  return entry.admissionUnknown === true && reusedRequestId(entry) !== undefined
+}
+
+/** What to do with a queued message about to go out. */
+export type ResendVerdict = 'send' | 'wait' | 'drop'
+
+/**
+ * Whether a queued message may go out, given its chat's history read fresh
+ * (`null` when the read failed). The server deduplicates a resend only while the
+ * earlier attempt's claim lasts, which a long outage outlives, so a message the
+ * history shows accepted is dropped (`drop`). One whose history cannot be read
+ * waits (`wait`): the read failing says nothing about whether it ran.
+ */
+export function resendVerdict(
+  entry: QueuedMothershipMessage,
+  history: MothershipChatHistory | null
+): ResendVerdict {
+  const requestId = reusedRequestId(entry)
+  if (!needsResendCheck(entry) || requestId === undefined) return 'send'
+  if (!history) return 'wait'
+  return acceptedMessageIds(history).has(requestId) ? 'drop' : 'send'
+}