fix(chat): deduplicate chat sends server-side instead of probing for them

A client cannot tell whether a request it aborted reached the server: the chat
route never reads `request.signal`, so an accepted one still opens the chat,
persists the user message, and bills the turn after the socket drops. #6525
answered that by polling the orphaned stream before retrying — a 2.5s guess
that had to distinguish "no such stream" from "we stopped looking", and still
left a window open.

The codebase already owns the right tool. `IdempotencyService` backs webhook,
polling, and billing dedup, and `billingIdempotency` exists for exactly this
hazard: "a retry would double-record usage — real money". Chat sends now claim
the same way, keyed on the client-generated `userMessageId` and scoped to the
caller so nobody can probe another user's sends. A repeat gets 409 naming the
chat the first attempt opened — deliberately the shape the pending-stream lock
already returns, so the client's existing conflict handler reattaches instead
of starting a turn, with only the chat-adoption line added.

The claim fails open at every step. Deduplication saves a duplicate chat; the
send IS the user's message, so an unreachable bookkeeping store degrades chat
rather than taking it down. It is released when a send fails before recording a
chat, and deliberately kept once recorded.

Retrying now just reuses the id, which deletes the probe outright: the poll and
its two constants, the three-state result, the epoch plumbing that kept a
superseded poll from re-sending, and the chat-adoption branch it needed. The
client hook nets 67 lines smaller.

Idle sends go back to calling `startSendMessage` directly. #6525 routed them
through the durable queue so recovery had a backing entry, which put every
message in the product through the queue store, sessionStorage, and the
dispatch loop for the sake of a rare path — and the recovery never needed it,
since the message, attachments, contexts, and id are all in scope at the abort.
Both callers now share one `handOffWithdrawnSend`.

`startSendMessage` takes its optional tail as an options object; it was at six
positional parameters and the retry id would have been a seventh.

Tests cover both halves: the server dedups, scopes the key per user, records
the chat, and still sends when the claim store is down; the client reuses the
original id on retry and adopts the chat a deduplicated retry names. Each was
confirmed red without its fix.

Author: waleedlatif1

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

@@ -342,7 +342,7 @@ export function Home({ chatId, userName, userId, tableViewsEnabled }: HomeProps)
       if (!detail?.message) return
       e.preventDefault()
       sendMessage(detail.message, detail.fileAttachments, detail.contexts, {
-        ...(detail.recoverStreamId ? { recoverStreamId: detail.recoverStreamId } : {}),
+        ...(detail.resumeUserMessageId ? { resumeUserMessageId: detail.resumeUserMessageId } : {}),
       })
     }
     window.addEventListener(MOTHERSHIP_SEND_MESSAGE_EVENT, handler)
@@ -373,7 +373,9 @@ export function Home({ chatId, userName, userId, tableViewsEnabled }: HomeProps)
     if (!handoff) return
     if (handoff.message) {
       sendMessage(handoff.message, handoff.fileAttachments, handoff.contexts, {
-        ...(handoff.recoverStreamId ? { recoverStreamId: handoff.recoverStreamId } : {}),
+        ...(handoff.resumeUserMessageId
+          ? { resumeUserMessageId: handoff.resumeUserMessageId }
+          : {}),
       })
       return
     }

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

@@ -490,7 +490,7 @@ export const Panel = memo(function Panel() {
       e.preventDefault()
       setActiveTab('copilot')
       copilotSendMessage(detail.message, detail.fileAttachments, detail.contexts, {
-        ...(detail.recoverStreamId ? { recoverStreamId: detail.recoverStreamId } : {}),
+        ...(detail.resumeUserMessageId ? { resumeUserMessageId: detail.resumeUserMessageId } : {}),
       })
     }
     window.addEventListener(MOTHERSHIP_SEND_MESSAGE_EVENT, handler)

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

@@ -36,6 +36,9 @@ const {
   appendCopilotChatMessages,
   persistChatResources,
   mockPublishStatusChanged,
+  atomicallyClaimChatSend,
+  storeChatSendResult,
+  releaseChatSendClaim,
 } = vi.hoisted(() => ({
   generateWorkspaceSnapshot: vi.fn(),
   processContextsServer: vi.fn(),
@@ -51,6 +54,9 @@ const {
   appendCopilotChatMessages: vi.fn(),
   persistChatResources: vi.fn(),
   mockPublishStatusChanged: vi.fn(),
+  atomicallyClaimChatSend: vi.fn(),
+  storeChatSendResult: vi.fn(),
+  releaseChatSendClaim: vi.fn(),
 }))
 
 const getSession = authMockFns.mockGetSession
@@ -103,6 +109,14 @@ vi.mock('@/lib/copilot/chat/lifecycle', () => ({
   resolveOrCreateChat,
 }))
 
+vi.mock('@/lib/core/idempotency', () => ({
+  chatSendIdempotency: {
+    atomicallyClaim: atomicallyClaimChatSend,
+    storeResult: storeChatSendResult,
+    release: releaseChatSendClaim,
+  },
+}))
+
 vi.mock('@/lib/copilot/chat/terminal-state', () => ({
   finalizeAssistantTurn,
 }))
@@ -132,6 +146,14 @@ describe('handleUnifiedChatPost', () => {
   beforeEach(() => {
     vi.clearAllMocks()
     resetDbChainMock()
+    atomicallyClaimChatSend.mockResolvedValue({
+      claimed: true,
+      normalizedKey: 'chat-send:mothership:msg-1:userId=user-1',
+      storageMethod: 'database',
+      claimToken: 'claim-1',
+    })
+    storeChatSendResult.mockResolvedValue(true)
+    releaseChatSendClaim.mockResolvedValue(undefined)
     getSession.mockResolvedValue({ user: { id: 'user-1' } })
     resolveWorkflowIdForUser.mockResolvedValue({
       status: 'resolved',
@@ -725,4 +747,105 @@ describe('handleUnifiedChatPost', () => {
       error: 'workspaceId is required when workflowId is not provided',
     })
   })
+
+  describe('deduplicating a repeated send', () => {
+    /**
+     * The client cannot tell whether a request it aborted reached the server —
+     * the route never reads `request.signal`, so an accepted one runs to
+     * completion regardless. Recovering such a send therefore retries it under
+     * the original `userMessageId`, and this is what makes that safe.
+     */
+    it('answers an already-claimed send with the chat the first attempt opened', async () => {
+      atomicallyClaimChatSend.mockResolvedValue({
+        claimed: false,
+        normalizedKey: 'chat-send:mothership:msg-1:userId=user-1',
+        storageMethod: 'database',
+        existingResult: { success: true, status: 'completed', result: { chatId: 'chat-first' } },
+      })
+
+      const response = await handleUnifiedChatPost(
+        new NextRequest('http://localhost/api/mothership/chat', {
+          method: 'POST',
+          body: JSON.stringify({
+            message: 'Hello',
+            workspaceId: 'ws-1',
+            userMessageId: 'msg-1',
+            createNewChat: true,
+          }),
+        })
+      )
+
+      expect(response.status).toBe(409)
+      await expect(response.json()).resolves.toMatchObject({
+        activeStreamId: 'msg-1',
+        chatId: 'chat-first',
+      })
+      // The whole point: no second chat, no second billed turn.
+      expect(resolveOrCreateChat).not.toHaveBeenCalled()
+      expect(createSSEStream).not.toHaveBeenCalled()
+    })
+
+    it('scopes the claim to the caller so one user cannot probe another', async () => {
+      await handleUnifiedChatPost(
+        new NextRequest('http://localhost/api/mothership/chat', {
+          method: 'POST',
+          body: JSON.stringify({
+            message: 'Hello',
+            workspaceId: 'ws-1',
+            userMessageId: 'msg-1',
+            createNewChat: true,
+          }),
+        })
+      )
+
+      expect(atomicallyClaimChatSend).toHaveBeenCalledWith('mothership', 'msg-1', {
+        userId: 'user-1',
+      })
+    })
+
+    it('records the chat against the send so a retry resolves to it', async () => {
+      await handleUnifiedChatPost(
+        new NextRequest('http://localhost/api/mothership/chat', {
+          method: 'POST',
+          body: JSON.stringify({
+            message: 'Hello',
+            workspaceId: 'ws-1',
+            userMessageId: 'msg-1',
+            createNewChat: true,
+          }),
+        })
+      )
+
+      expect(storeChatSendResult).toHaveBeenCalledWith(
+        'chat-send:mothership:msg-1:userId=user-1',
+        expect.objectContaining({ result: { chatId: 'chat-1' } }),
+        'database',
+        'claim-1'
+      )
+    })
+
+    /**
+     * Deduplication saves a duplicate chat; the send IS the user's message.
+     * An unreachable bookkeeping store must degrade chat, never take it down.
+     */
+    it('sends normally when the claim store is unavailable', async () => {
+      atomicallyClaimChatSend.mockRejectedValue(new Error('idempotency store down'))
+
+      const response = await handleUnifiedChatPost(
+        new NextRequest('http://localhost/api/mothership/chat', {
+          method: 'POST',
+          body: JSON.stringify({
+            message: 'Hello',
+            workspaceId: 'ws-1',
+            userMessageId: 'msg-1',
+            createNewChat: true,
+          }),
+        })
+      )
+
+      expect(response.status).toBe(200)
+      expect(createSSEStream).toHaveBeenCalled()
+      expect(storeChatSendResult).not.toHaveBeenCalled()
+    })
+  })
 })

apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/panel.tsx

@@ -58,6 +58,8 @@ import {
   sanitizeChatResources,
 } from '@/lib/copilot/resources/types'
 import { prepareExecutionContext } from '@/lib/copilot/tools/handlers/context'
+import type { AtomicClaimResult } from '@/lib/core/idempotency'
+import { chatSendIdempotency } from '@/lib/core/idempotency'
 import { captureServerEvent } from '@/lib/posthog/server'
 import { resolveWorkflowIdForUser } from '@/lib/workflows/utils'
 import {
@@ -943,10 +945,86 @@ async function resolveBranch(params: {
   }
 }
 
+/** Namespace-local provider segment for {@link chatSendIdempotency} keys. */
+const CHAT_SEND_IDEMPOTENCY_PROVIDER = 'mothership'
+
+/**
+ * Claims this send so a retry of it can be recognised.
+ *
+ * Fails open. Deduplication is an optimization here — it saves a duplicate chat
+ * and a duplicate billed turn — whereas the send itself is the user's message,
+ * and refusing it because a bookkeeping store is unreachable would take chat
+ * down for everyone rather than degrade it. An unavailable store therefore
+ * returns `undefined`, which sends normally with no claim to finalize.
+ *
+ * The key is scoped to the caller: `userMessageId` is client-supplied, so an
+ * unscoped one would let a user probe another's sends and read back their chat
+ * id.
+ */
+async function claimChatSend(
+  userMessageId: string,
+  userId: string,
+  requestId: string
+): Promise<AtomicClaimResult | undefined> {
+  try {
+    return await chatSendIdempotency.atomicallyClaim(
+      CHAT_SEND_IDEMPOTENCY_PROVIDER,
+      userMessageId,
+      { userId }
+    )
+  } catch (error) {
+    logger.warn(`[${requestId}] Could not claim chat send; proceeding without deduplication`, {
+      userMessageId,
+      error: getErrorMessage(error, 'Unknown error'),
+    })
+    return undefined
+  }
+}
+
+/** Chat a previously-claimed send resolved to, when it got that far. */
+function claimedChatId(claim: AtomicClaimResult): string | undefined {
+  const chatId = claim.existingResult?.result?.chatId
+  return typeof chatId === 'string' && chatId ? chatId : undefined
+}
+
+/**
+ * Answers a send whose `userMessageId` was already claimed.
+ *
+ * Deliberately the same 409 shape the pending-stream lock returns, because the
+ * client's conflict handler already knows how to reattach to `activeStreamId`
+ * instead of starting a turn — a duplicate send and a send that collided with
+ * an in-flight one want exactly the same thing. `chatId` rides along when the
+ * first attempt got far enough to resolve one, letting a chatless client adopt
+ * it without a stream-to-chat lookup.
+ */
+function duplicateChatSendResponse(
+  claim: AtomicClaimResult,
+  userMessageId: string,
+  requestId: string
+): NextResponse {
+  const chatId = claimedChatId(claim)
+  logger.info(`[${requestId}] Deduplicated a repeated chat send`, { userMessageId, chatId })
+  return NextResponse.json(
+    {
+      error: 'This message was already sent.',
+      activeStreamId: userMessageId,
+      ...(chatId ? { chatId } : {}),
+    },
+    { status: 409 }
+  )
+}
+
 export async function handleUnifiedChatPost(req: NextRequest) {
   let actualChatId: string | undefined
   let userMessageId = ''
   let chatStreamLockAcquired = false
+  /* Held from the moment this send is claimed until the chat it belongs to is
+     recorded. Released on the error path so a failed send can be retried, and
+     deliberately NOT released once recorded — by then the chat exists and the
+     user message is persisted, so a retry must resolve to it rather than open
+     a second chat. */
+  let sendClaim: AtomicClaimResult | undefined
+  let sendClaimRecorded = false
   // Started once we've parsed the body (need userMessageId to stamp as
   // streamId). Every subsequent span (persistUserMessage,
   // createRunSegment, the whole SSE stream, etc.) nests under this
@@ -981,6 +1059,11 @@ export async function handleUnifiedChatPost(req: NextRequest) {
     const normalizedContexts = normalizeContexts(body.contexts) ?? []
     userMessageId = body.userMessageId || generateId()
 
+    sendClaim = await claimChatSend(userMessageId, authenticatedUserId, requestId)
+    if (sendClaim?.claimed === false) {
+      return duplicateChatSendResponse(sendClaim, userMessageId, requestId)
+    }
+
     otelRoot = startCopilotOtelRoot({
       streamId: userMessageId,
       executionId,
@@ -1073,6 +1156,28 @@ export async function handleUnifiedChatPost(req: NextRequest) {
         }
       }
 
+      /* Record the chat against this send as soon as it is known, the earliest
+         a retry can be answered with somewhere to go. From here the claim is
+         permanent: the user message lands moments later, so a retry must
+         resolve to this chat rather than open another. Failing to record only
+         costs deduplication, so it must not fail the send. */
+      if (sendClaim?.claimToken && actualChatId) {
+        sendClaimRecorded = await chatSendIdempotency
+          .storeResult(
+            sendClaim.normalizedKey,
+            { success: true, status: 'completed', result: { chatId: actualChatId } },
+            sendClaim.storageMethod,
+            sendClaim.claimToken
+          )
+          .catch((error) => {
+            logger.warn(`[${requestId}] Could not record the chat for this send`, {
+              userMessageId,
+              error: getErrorMessage(error, 'Unknown error'),
+            })
+            return false
+          })
+      }
+
       if (chatIsNew && actualChatId && body.resourceAttachments?.length) {
         // Canonicalizes here, not just inside `persistChatResources`: several
         // browser tabs collapse onto the one Browser panel before they are
@@ -1377,6 +1482,19 @@ export async function handleUnifiedChatPost(req: NextRequest) {
     if (chatStreamLockAcquired && actualChatId && userMessageId) {
       await releasePendingChatStream(actualChatId, userMessageId)
     }
+    // Nothing was recorded against this send, so let a retry of it through
+    // rather than deduplicating against a chat that was never opened. A failed
+    // release just leaves the claim to expire on its own short in-progress TTL.
+    if (sendClaim?.claimToken && !sendClaimRecorded) {
+      await chatSendIdempotency
+        .release(sendClaim.normalizedKey, sendClaim.storageMethod, sendClaim.claimToken)
+        .catch((releaseError) => {
+          logger.warn(`[${requestId}] Could not release the claim for a failed send`, {
+            userMessageId,
+            error: getErrorMessage(releaseError, 'Unknown error'),
+          })
+        })
+    }
     otelRoot?.finish('error', error)
 
     if (isZodError(error)) {

apps/sim/lib/copilot/chat/post.test.ts

@@ -733,3 +733,29 @@ export const billingIdempotency = new IdempotencyService({
   ttlSeconds: 60 * 60, // 1 hour
   forceStorage: 'database',
 })
+
+/**
+ * Dedupes a chat send by its client-generated `userMessageId`, so re-sending
+ * one is safe.
+ *
+ * The client cannot tell whether a request it aborted reached the server: the
+ * chat route never reads `request.signal`, so an accepted request creates the
+ * chat, persists the user message, and runs the (billed) turn even after the
+ * browser drops the socket. Without this, a client that recovers an aborted
+ * send has to choose between losing the message and duplicating the run.
+ *
+ * Storage is forced to Postgres for the same reason as {@link billingIdempotency}:
+ * a missed dedup is a second LLM turn billed to the workspace, so the key must
+ * not be evictable under Redis memory pressure. The added 1-5ms is invisible
+ * next to the LLM call this request is about to make.
+ *
+ * `inProgressTtlSeconds` is short so a crashed pod cannot block a genuine retry
+ * for the full hour, while completed sends stay deduplicated for it.
+ */
+export const chatSendIdempotency = new IdempotencyService({
+  namespace: 'chat-send',
+  ttlSeconds: 60 * 60, // 1 hour
+  inProgressTtlSeconds: 60,
+  retryFailures: true,
+  forceStorage: 'database',
+})