From ac7a7322f8e6768e5cf01ee13e7cf112527eb1d9 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 11 Aug 2026 00:51:56 -0700 Subject: [PATCH 1/4] fix(chat): deduplicate chat sends server-side instead of probing for them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../app/workspace/[workspaceId]/home/home.tsx | 6 +- .../home/hooks/use-chat.mount-send.test.tsx | 388 +++++++----------- .../[workspaceId]/home/hooks/use-chat.ts | 355 +++++++--------- .../w/[workflowId]/components/panel/panel.tsx | 2 +- apps/sim/lib/copilot/chat/post.test.ts | 123 ++++++ apps/sim/lib/copilot/chat/post.ts | 118 ++++++ apps/sim/lib/core/idempotency/service.ts | 26 ++ apps/sim/lib/core/utils/browser-storage.ts | 16 +- apps/sim/lib/mothership/events.ts | 14 +- apps/sim/stores/mothership-queue/store.ts | 8 +- apps/sim/stores/mothership-queue/types.ts | 13 +- 11 files changed, 577 insertions(+), 492 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/home/home.tsx b/apps/sim/app/workspace/[workspaceId]/home/home.tsx index 093f75c6556..03008c597ae 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/home.tsx +++ b/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 } diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx index dbc50800722..185127cc2f5 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx @@ -3,18 +3,20 @@ * * Regression tests for the remount send loss: a send started on a fresh chat * surface was silently dropped when the hook's unmount cleanup ran mid-flight - * and aborted the POST. Two things run that cleanup while an auto-send from a - * cross-route handoff is still in flight — StrictMode's dev double-mount, and a - * real client-side navigation away — and because `MothershipHandoffStorage` - * consumes atomically, the second mount finds nothing left to retry. + * and aborted the POST. Two things run that cleanup while an auto-send is still + * in flight — the chat route's `key={chatId}` remount when the user switches + * chats, and StrictMode's dev double-mount — and because + * `MothershipHandoffStorage` consumes atomically, the replacement mount finds + * nothing left to retry. * * (A Suspense hide/reveal does NOT cause this: React 19 disappears layout * effects only, so this passive cleanup never runs for it.) * - * The fix routes idle sends through the durable queue so every send has a - * recoverable entry, and recovers one the cleanup withdrew — probing the - * orphaned stream first so a request the server had already accepted is - * adopted rather than sent twice. + * Recovery hands the message to the next surface carrying the original + * `userMessageId`. Reusing that id is what makes the retry safe: the server + * deduplicates it against the first attempt rather than opening a second chat + * and billing a second turn, so the client never has to guess whether the + * request it aborted was accepted. */ import { act, type ReactNode, StrictMode, useEffect } from 'react' import { sleep } from '@sim/utils/helpers' @@ -42,29 +44,20 @@ import { MothershipHandoffStorage } from '@/lib/core/utils/browser-storage' import { useChat } from '@/app/workspace/[workspaceId]/home/hooks/use-chat' import { useMothershipQueueStore } from '@/stores/mothership-queue/store' +const DEDUPED_CHAT_ID = 'chat-the-first-attempt-opened' + interface NetworkState { - /** How the chat POST behaves for the next call. */ - postBehavior: 'hang' | 'accept' - postCalls: number /** - * How the orphaned-stream probe answers: - * - `found` — the server accepted the withdrawn request and owns a chat - * - `gone` — 404, it has no such stream (never accepted) - * - `pending` — registered but no owner yet, so the probe keeps polling; - * this is the only mode that leaves a probe in flight to interrupt + * How the chat POST behaves: + * - `hang` — accepted but never answered, the window the cleanup abort lands in + * - `accept` — a normal streaming response + * - `deduped` — the 409 the server returns for an already-claimed send */ - probeBehavior: 'found' | 'gone' | 'pending' - orphanedStreamChatId: string - streamProbes: number + postBehavior: 'hang' | 'accept' | 'deduped' + postBodies: Array<{ message: string; userMessageId?: string }> } -const state: NetworkState = { - postBehavior: 'hang', - postCalls: 0, - probeBehavior: 'gone', - orphanedStreamChatId: 'chat-server-already-made', - streamProbes: 0, -} +const state: NetworkState = { postBehavior: 'hang', postBodies: [] } /** An SSE response whose stream ends immediately without a terminal event. */ function emptySseResponse(): Response { @@ -79,29 +72,28 @@ function emptySseResponse(): Response { async function fetchStub(input: RequestInfo | URL, init?: RequestInit): Promise { const url = String(input instanceof Request ? input.url : input) - // The orphaned-stream probe: does the server hold a stream for the send the - // cleanup abort withdrew? + /* Stream replay, used by the reconnect a deduplicated send falls into. + `complete` is the terminal status the hook recognises — anything else and + reconnect polls forever. */ if (url.includes('/api/mothership/chat/stream')) { - state.streamProbes++ - // 404 is what the server returns for a stream it never registered — i.e. - // the request really was withdrawn before it was accepted. - if (state.probeBehavior === 'gone') { - return new Response(JSON.stringify({ error: 'stream gone' }), { status: 404 }) - } - return new Response( - JSON.stringify({ - success: true, - events: [], - status: 'streaming', - // `pending` omits the owner, so the probe keeps polling. - ...(state.probeBehavior === 'found' ? { chatId: state.orphanedStreamChatId } : {}), - }), - { status: 200, headers: { 'Content-Type': 'application/json' } } - ) + return new Response(JSON.stringify({ success: true, events: [], status: 'complete' }), { + status: 200, + headers: { 'Content-Type': 'application/json' }, + }) } if (url.includes('/api/mothership/chat') && init?.method === 'POST') { - state.postCalls++ + state.postBodies.push(JSON.parse(String(init.body))) + if (state.postBehavior === 'deduped') { + return new Response( + JSON.stringify({ + error: 'This message was already sent.', + activeStreamId: state.postBodies.at(-1)?.userMessageId, + chatId: DEDUPED_CHAT_ID, + }), + { status: 409, headers: { 'Content-Type': 'application/json' } } + ) + } if (state.postBehavior === 'accept') return emptySseResponse() return new Promise((_, reject) => { const signal = init?.signal @@ -154,49 +146,11 @@ function renderUseChat(): { } /** - * Mounts the hook under StrictMode with a handoff already in storage, mirroring - * `home.tsx`'s consume-and-auto-send effect. This is the production-shaped - * failure: the dev double-mount runs the passive cleanup between the two - * mounts, aborting the in-flight POST, and `consume` has already cleared the - * entry so the second mount has nothing to replay. - */ -function renderStrictModeHandoffConsumer(): { unmount: () => void } { - ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true - queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }) - const container = document.createElement('div') - const root = createRoot(container) - mountedRoots.push(root) - - function Probe() { - const { sendMessage } = useChat('ws-1', undefined) - useEffect(() => { - const handoff = MothershipHandoffStorage.consume('ws-1') - if (!handoff?.message) return - sendMessage(handoff.message, handoff.fileAttachments, handoff.contexts, { - ...(handoff.recoverStreamId ? { recoverStreamId: handoff.recoverStreamId } : {}), - }) - }, [sendMessage]) - return null - } - - act(() => { - root.render( - - {() as ReactNode} - - ) - }) - - return { unmount: () => act(() => root.unmount()) } -} - -/** - * Mounts a surface shaped like `home.tsx`: it drives `useChat` AND registers - * the `mothership-send-message` listener that claims the event with - * `preventDefault`. Unmounting this exercises the ordering question — whether - * the departing surface's own still-attached listener can claim the recovery - * event its own teardown emitted, which would suppress the storage fallback - * and strand the message. + * Mounts a surface shaped like `home.tsx`: it drives `useChat` AND registers the + * `mothership-send-message` listener that claims the event with + * `preventDefault`. Unmounting it exercises whether the departing surface's own + * still-attached listener can claim the recovery event its teardown emitted, + * which would suppress the storage fallback and strand the message. */ function renderHomeLikeSurface(): { getResult: () => ReturnType @@ -215,16 +169,18 @@ function renderHomeLikeSurface(): { const chat = useChat('ws-1', undefined) result = chat const { sendMessage } = chat - // Mirrors home.tsx:339 — declared AFTER useChat, so on unmount React runs + // Mirrors home.tsx — declared AFTER useChat, so on unmount React runs // useChat's cleanup (which aborts) before this removeEventListener. useEffect(() => { const handler = (e: Event) => { - const detail = (e as CustomEvent<{ message?: string; recoverStreamId?: string }>).detail + const detail = (e as CustomEvent<{ message?: string; resumeUserMessageId?: string }>).detail if (!detail?.message) return claims++ e.preventDefault() sendMessage(detail.message, undefined, undefined, { - ...(detail.recoverStreamId ? { recoverStreamId: detail.recoverStreamId } : {}), + ...(detail.resumeUserMessageId + ? { resumeUserMessageId: detail.resumeUserMessageId } + : {}), }) } window.addEventListener('mothership-send-message', handler) @@ -249,6 +205,43 @@ function renderHomeLikeSurface(): { } } +/** + * Mounts the hook under StrictMode with a handoff already in storage, mirroring + * `home.tsx`'s consume-and-auto-send effect. This is the production-shaped + * failure: the dev double-mount runs the passive cleanup between the two + * mounts, aborting the in-flight POST, and `consume` has already cleared the + * entry so the second mount has nothing to replay. + */ +function renderStrictModeHandoffConsumer(): void { + ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true + queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }) + const container = document.createElement('div') + const root = createRoot(container) + mountedRoots.push(root) + + function Probe() { + const { sendMessage } = useChat('ws-1', undefined) + useEffect(() => { + const handoff = MothershipHandoffStorage.consume('ws-1') + if (!handoff?.message) return + sendMessage(handoff.message, handoff.fileAttachments, handoff.contexts, { + ...(handoff.resumeUserMessageId + ? { resumeUserMessageId: handoff.resumeUserMessageId } + : {}), + }) + }, [sendMessage]) + return null + } + + act(() => { + root.render( + + {() as ReactNode} + + ) + }) +} + /** Every queued message across all chat keys, flattened. */ function allQueuedMessages() { return Object.values(useMothershipQueueStore.getState().queues).flat() @@ -268,9 +261,7 @@ describe('useChat remount send recovery', () => { beforeEach(() => { vi.stubGlobal('fetch', fetchStub) state.postBehavior = 'hang' - state.postCalls = 0 - state.probeBehavior = 'gone' - state.streamProbes = 0 + state.postBodies = [] mockRequestJson.mockResolvedValue({ chats: [] }) useMothershipQueueStore.setState({ queues: {}, editing: {} }) window.sessionStorage.clear() @@ -286,7 +277,23 @@ describe('useChat remount send recovery', () => { vi.clearAllMocks() }) - it('delivers an aborted chatless send directly to a live replacement surface', async () => { + it('keeps a cross-route handoff recoverable across a StrictMode double-mount', async () => { + MothershipHandoffStorage.store({ message: 'investigate this failed run' }, 'ws-1') + + renderStrictModeHandoffConsumer() + await waitFor(() => state.postBodies.length >= 1) + + // Something must still hold the message: the live event was claimed and it + // is in flight again, or it is back in storage for the next mount. + await waitFor( + () => + window.localStorage.getItem('sim_mothership_handoff') !== null || + allQueuedMessages().length > 0 || + state.postBodies.length > 1 + ) + }) + + it('delivers a withdrawn chatless send to a live replacement surface', async () => { const attachment = { id: 'file-1', key: 'uploads/file-1', @@ -294,10 +301,13 @@ describe('useChat remount send recovery', () => { media_type: 'text/plain', size: 12, } - const received: Array<{ message: string; fileAttachments?: unknown[] }> = [] + const received: Array<{ + message: string + fileAttachments?: unknown[] + resumeUserMessageId?: string + }> = [] const claim = (event: Event) => { - const detail = (event as CustomEvent<{ message: string; fileAttachments?: unknown[] }>).detail - received.push(detail) + received.push((event as CustomEvent<(typeof received)[number]>).detail) event.preventDefault() } window.addEventListener('mothership-send-message', claim) @@ -307,20 +317,22 @@ describe('useChat remount send recovery', () => { await act(async () => { void getResult().sendMessage('hello from the palette', [attachment]) }) - await waitFor(() => state.postCalls === 1) + await waitFor(() => state.postBodies.length === 1) unmount() await waitFor(() => received.length === 1) expect(received[0].message).toBe('hello from the palette') expect(received[0].fileAttachments).toEqual([attachment]) + // Carried so the replacement retries as the same send, not a new one. + expect(received[0].resumeUserMessageId).toBe(state.postBodies[0].userMessageId) expect(window.localStorage.getItem('sim_mothership_handoff')).toBeNull() } finally { window.removeEventListener('mothership-send-message', claim) } }) - it('re-persists an aborted chatless send as a handoff for the next mount', async () => { + it('re-persists a withdrawn chatless send as a handoff for the next mount', async () => { const attachment = { id: 'file-2', key: 'uploads/file-2', @@ -333,16 +345,11 @@ describe('useChat remount send recovery', () => { await act(async () => { void getResult().sendMessage('hello from the palette', [attachment]) }) - await waitFor(() => state.postCalls === 1) + await waitFor(() => state.postBodies.length === 1) - // The dispatch claimed the queue head when the optimistic send applied. + // An idle send goes straight out — it never occupies the queue. expect(allQueuedMessages()).toHaveLength(0) - // The cleanup abort (the same code path a StrictMode remount or a real - // navigation away runs) fires while the POST is still awaiting the server. - // A chatless surface regenerates its queue key per mount, so recovery - // re-persists the send as a one-shot handoff for the next mount's consumer - // instead of restoring the dead instance's queue. unmount() await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null) @@ -350,16 +357,17 @@ describe('useChat remount send recovery', () => { const handoff = MothershipHandoffStorage.consume('ws-1') expect(handoff?.message).toBe('hello from the palette') expect(handoff?.fileAttachments).toEqual([attachment]) + expect(handoff?.resumeUserMessageId).toBe(state.postBodies[0].userMessageId) }) - it('does not re-queue a send the server already received', async () => { + it('does not recover a send the server already answered', async () => { state.postBehavior = 'accept' const { getResult, unmount } = renderUseChat() await act(async () => { void getResult().sendMessage('already accepted') }) - await waitFor(() => state.postCalls === 1) + await waitFor(() => state.postBodies.length === 1) unmount() await act(async () => { @@ -373,15 +381,15 @@ describe('useChat remount send recovery', () => { /** * A departing surface's own listener must not claim the recovery event its * teardown emitted: claiming returns `true`, which suppresses the storage - * fallback, and the enqueue would land under the disposed pending key — the - * message would be stranded exactly where this fix is supposed to save it. + * fallback, and the message would be stranded exactly where this fix is meant + * to save it. */ it('does not let a departing surface claim its own recovery event', async () => { const surface = renderHomeLikeSurface() await act(async () => { void surface.getResult().sendMessage('must survive my own teardown') }) - await waitFor(() => state.postCalls === 1) + await waitFor(() => state.postBodies.length === 1) surface.unmount() await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null) @@ -390,166 +398,50 @@ describe('useChat remount send recovery', () => { expect(MothershipHandoffStorage.consume('ws-1')?.message).toBe('must survive my own teardown') }) - /** - * The end-to-end failure, driven by the thing that actually runs the cleanup - * mid-flight rather than by a hand-rolled unmount. On the unfixed hook the - * handoff is consumed, the POST is aborted, and nothing survives to retry. - */ - it('keeps a cross-route handoff recoverable across a StrictMode double-mount', async () => { - MothershipHandoffStorage.store({ message: 'investigate this failed run' }, 'ws-1') - - renderStrictModeHandoffConsumer() - await waitFor(() => state.postCalls >= 1) - - // Something must still be holding the message: either the live event was - // claimed and it is queued/in flight again, or it is back in storage. - await waitFor(() => { - const stored = window.localStorage.getItem('sim_mothership_handoff') - return stored !== null || allQueuedMessages().length > 0 || state.postCalls > 1 - }) - }) - - /** - * The abort tears down the client socket but the route handler never reads - * `request.signal` — a request the server had already accepted still creates - * the chat, persists the user message, and runs (and bills) the turn. So the - * recovered send has to ask whether that happened before sending again. - */ - describe('recovered send probes the orphaned stream before re-sending', () => { - it('adopts the chat the server already created instead of sending twice', async () => { - // The server accepted the withdrawn request and registered its stream. - state.probeBehavior = 'found' - - const { getResult, unmount } = renderUseChat() - await act(async () => { - void getResult().sendMessage('only once please') - }) - await waitFor(() => state.postCalls === 1) - unmount() - await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null) - - // The next mount consumes the handoff, exactly as home.tsx does. - const handoff = MothershipHandoffStorage.consume('ws-1') - expect(handoff?.recoverStreamId).toBeTruthy() - - const replacement = renderUseChat() - await act(async () => { - void replacement.getResult().sendMessage(handoff?.message as string, undefined, undefined, { - recoverStreamId: handoff?.recoverStreamId as string, - }) - }) - await waitFor(() => state.streamProbes > 0) - await waitFor(() => replacement.getResult().resolvedChatId === 'chat-server-already-made') - - expect(state.postCalls).toBe(1) - expect(allQueuedMessages()).toHaveLength(0) - }) - + describe('retrying a withdrawn send', () => { /** - * Adoption alone does not surface the running turn: hydration reconnects - * only when `chatHistory.activeStreamId` is set, and that query is cached - * for 30s. Without an explicit detail invalidation the adopted chat renders - * with the live response invisible. + * The whole point of carrying the id: the server sees one logical send, so + * it deduplicates instead of opening a second chat and billing again. */ - it('invalidates the adopted chat detail so hydration can reconnect', async () => { - state.probeBehavior = 'found' - + it('reuses the original message id so the server can deduplicate', async () => { const { getResult, unmount } = renderUseChat() await act(async () => { - void getResult().sendMessage('surface the running turn') + void getResult().sendMessage('only bill me once') }) - await waitFor(() => state.postCalls === 1) + await waitFor(() => state.postBodies.length === 1) unmount() await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null) const handoff = MothershipHandoffStorage.consume('ws-1') const replacement = renderUseChat() - const invalidateSpy = vi.spyOn(queryClient, 'invalidateQueries') - await act(async () => { void replacement.getResult().sendMessage(handoff?.message as string, undefined, undefined, { - recoverStreamId: handoff?.recoverStreamId as string, + resumeUserMessageId: handoff?.resumeUserMessageId as string, }) }) - await waitFor(() => - invalidateSpy.mock.calls.some(([arg]) => { - const key = (arg as { queryKey?: unknown[] } | undefined)?.queryKey - return Array.isArray(key) && key.includes('chat-server-already-made') - }) - ) + await waitFor(() => state.postBodies.length === 2) - expect(state.postCalls).toBe(1) + expect(state.postBodies[1].userMessageId).toBe(state.postBodies[0].userMessageId) }) /** - * A probe cut short by unmount answers "unknown", not "safe to send". - * Falling through to `startSendMessage` there would open a POST whose - * abort controller the teardown already dropped — an uncancellable request - * that duplicates the send. The entry must stay queued instead. + * When the first attempt did reach the server, the retry comes back 409 + * naming the chat it opened. The client adopts that chat rather than + * starting another turn. */ - it('does not send when the probe is cut short by unmount', async () => { - state.probeBehavior = 'gone' - const { getResult, unmount } = renderUseChat() - await act(async () => { - void getResult().sendMessage('do not zombie me') - }) - await waitFor(() => state.postCalls === 1) - unmount() - await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null) + it('adopts the chat a deduplicated retry names', async () => { + state.postBehavior = 'deduped' + const { getResult } = renderUseChat() - const handoff = MothershipHandoffStorage.consume('ws-1') - - /* `pending` keeps the probe polling instead of answering on the first - attempt, which is what leaves one in flight to interrupt. A `gone` - probe answers immediately and the re-send would already have happened - before the unmount — that version of this test cannot fail. */ - state.probeBehavior = 'pending' - const replacement = renderUseChat() await act(async () => { - void replacement.getResult().sendMessage(handoff?.message as string, undefined, undefined, { - recoverStreamId: handoff?.recoverStreamId as string, - }) - }) - await waitFor(() => state.streamProbes > 0) - const postsBeforeUnmount = state.postCalls - replacement.unmount() - - // Well past the probe's poll budget: nothing may send after teardown. - await act(async () => { - await sleep(3000) - }) - expect(state.postCalls).toBe(postsBeforeUnmount) - - /* Not sending is only half of it — the message must still be - recoverable. A chatless surface's `pending::` key is regenerated per - mount, so leaving the entry there would strand it just as surely as - re-sending would have duplicated it. */ - expect(MothershipHandoffStorage.consume('ws-1')?.message).toBe('do not zombie me') - expect(allQueuedMessages()).toHaveLength(0) - }) - - it('re-sends when the server has no stream for it', async () => { - // 404 from the probe: the request really was withdrawn before acceptance. - state.probeBehavior = 'gone' - - const { getResult, unmount } = renderUseChat() - await act(async () => { - void getResult().sendMessage('please actually send me') - }) - await waitFor(() => state.postCalls === 1) - unmount() - await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null) - - const handoff = MothershipHandoffStorage.consume('ws-1') - const replacement = renderUseChat() - await act(async () => { - void replacement.getResult().sendMessage(handoff?.message as string, undefined, undefined, { - recoverStreamId: handoff?.recoverStreamId as string, + void getResult().sendMessage('this one already landed', undefined, undefined, { + resumeUserMessageId: 'the-first-attempt', }) }) - await waitFor(() => state.postCalls === 2) + await waitFor(() => getResult().resolvedChatId === DEDUPED_CHAT_ID) - expect(state.streamProbes).toBeGreaterThan(0) + expect(state.postBodies).toHaveLength(1) + expect(state.postBodies[0].userMessageId).toBe('the-first-attempt') }) }) }) diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts index 1055f20b703..531bf9f5080 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -138,31 +138,43 @@ import type { export interface SendMessageOptions { /** - * Stream id of a send a cleanup abort withdrew, when this call is the - * recovery of it. The dispatcher probes the id before sending and adopts the - * chat the server already created instead, when there is one. + * Message id of a prior attempt this send retries, set when recovering a send + * an unmount cleanup withdrew. Reusing it lets the server deduplicate the two + * attempts instead of opening a second chat. */ - recoverStreamId?: string + resumeUserMessageId?: string } /** * `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 a cleanup abort withdrew the send while its request was on the wire — - * `streamId` is what recovery probes before re-sending. + * when an unmount cleanup withdrew it — `userMessageId` is what a retry reuses + * so the server deduplicates the two attempts. */ -type StartSendMessageResult = boolean | { kind: 'recoverable_cleanup_abort'; streamId: string } +type StartSendMessageResult = boolean | { kind: 'withdrawn_by_cleanup'; userMessageId: string } + +interface StartSendMessageOptions { + /** Awaited before dispatch. Defaults to the hook's in-flight stop, if any. */ + pendingStop?: Promise | null + /** Runs once the optimistic user/assistant pair is in the transcript. */ + onOptimisticSendApplied?: () => void + /** Seed for a queued send that superseded a stopped stream. */ + queuedSendHandoff?: QueuedSendHandoffSeed + /** + * Message id of a prior attempt this send retries. Reusing it is what makes + * the retry safe: the server deduplicates against that attempt rather than + * opening a second chat and billing a second turn. + */ + resumeUserMessageId?: string +} -/** - * Outcome of asking the server whether it accepted the request a cleanup abort - * withdrew. `superseded` is deliberately distinct from `not_found`: the former - * means the answer is unknown because this dispatch is no longer current, and - * must not be treated as licence to re-send. - */ -type RecoveredSendProbe = - | { status: 'adopted'; chatId: string } - | { status: 'not_found' } - | { status: 'superseded' } +/** A send an unmount cleanup withdrew, as handed to the next chat surface. */ +interface WithdrawnSend { + content: string + fileAttachments?: FileAttachmentForApi[] + contexts?: ChatContext[] + userMessageId: string +} export interface UseChatReturn { messages: ChatMessage[] @@ -205,16 +217,6 @@ const RECONNECT_MAX_DELAY_MS = 30_000 const STREAM_BATCH_FETCH_TIMEOUT_MS = 10_000 const STREAM_CHAT_ID_RESOLVE_TIMEOUT_MS = 10_000 const CHAT_HISTORY_RECOVERY_TIMEOUT_MS = 10_000 -/** - * How long a recovered send waits to find out whether the request its cleanup - * abort withdrew had in fact been accepted by the server. The server registers - * the stream early in the request (before it responds), so a short poll is - * enough; the ceiling is deliberately low because every millisecond here delays - * a send the user is watching for. Timing out re-sends, which is the safe - * direction: a lost message is worse than a rare duplicate. - */ -const RECOVERED_SEND_PROBE_TIMEOUT_MS = 2500 -const RECOVERED_SEND_PROBE_INTERVAL_MS = 250 const STOP_REQUEST_TIMEOUT_MS = 15_000 const QUEUED_SEND_HANDOFF_STORAGE_KEY = `${STREAM_STORAGE_KEY}:queued-send-handoff` const QUEUED_SEND_HANDOFF_CLAIM_STORAGE_KEY = `${STREAM_STORAGE_KEY}:queued-send-handoff-claim` @@ -3387,7 +3389,7 @@ export function useChat( message: string, fileAttachments?: FileAttachmentForApi[], contexts?: ChatContext[], - recoverStreamId?: string + resumeUserMessageId?: string ): QueuedMothershipMessage => { const id = generateId() const handoffChatId = selectedChatIdRef.current ?? chatIdRef.current @@ -3407,7 +3409,7 @@ export function useChat( content: message, fileAttachments, contexts, - ...(recoverStreamId ? { recoverStreamId } : {}), + ...(resumeUserMessageId ? { resumeUserMessageId } : {}), ...(supersededStreamId || handoffChatId ? { queuedSendHandoff: { @@ -3494,12 +3496,11 @@ export function useChat( message: string, fileAttachments?: FileAttachmentForApi[], contexts?: ChatContext[], - pendingStopOverride?: Promise | null, - onOptimisticSendApplied?: () => void, - queuedSendHandoff?: QueuedSendHandoffSeed + options?: StartSendMessageOptions ): Promise => { if (!message.trim() || !workspaceId) return false - const pendingStop = pendingStopOverride ?? pendingStopPromiseRef.current + const { onOptimisticSendApplied, queuedSendHandoff } = options ?? {} + const pendingStop = options?.pendingStop ?? pendingStopPromiseRef.current const pendingStopStreamId = pendingStop ? queuedSendHandoff?.supersededStreamId || locallyTerminalStreamIdRef.current || @@ -3514,7 +3515,10 @@ export function useChat( setError(null) setTransportStreaming() - const userMessageId = queuedSendHandoff?.userMessageId ?? generateId() + /* A retry of a withdrawn send reuses its id so the server deduplicates + the two attempts; anything else mints a fresh one. */ + const userMessageId = + queuedSendHandoff?.userMessageId ?? options?.resumeUserMessageId ?? generateId() const assistantId = getLiveAssistantMessageId(userMessageId) const storedAttachments: PersistedFileAttachment[] | undefined = @@ -3809,6 +3813,19 @@ export function useChat( setError('Previous response is still shutting down; queued message was restored.') return false } + /* 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 + chat before the reconnect below replays it. */ + const conflictChatId = + typeof errorData.chatId === 'string' ? errorData.chatId : undefined + if (conflictChatId && !streamTargetChatId) { + adoptResolvedChatId(conflictChatId, { + replaceHomeHistory: true, + invalidateList: true, + }) + streamTargetChatId = conflictChatId + } streamIdRef.current = conflictStreamId const succeeded = await retryReconnect({ streamId: conflictStreamId, @@ -3877,19 +3894,18 @@ export function useChat( (err instanceof Error && err.name === 'AbortError') || sendAbortSignal?.aborted === true if (sendWasAborted) { if (sendAbortSignal?.reason === 'unmount:client_cleanup' && !sendReachedServer) { - /* A remount (StrictMode's dev double-mount, or a real navigation - away) ran the unmount cleanup before this send's response headers - arrived. Nothing was rendered from it, so withdraw the optimistic - pair and report the distinct outcome so the dispatcher recovers - the message. - - `sendReachedServer` only rules out a send whose RESPONSE landed — - the request itself may well have been accepted, and the route - never reads `request.signal`, so it runs to completion either - way. Recovery therefore carries `userMessageId` as the stream id - to probe before re-sending. */ + /* A remount ran the unmount cleanup before this send's response + arrived — a chat-route `key` change, or StrictMode's dev + double-mount. Nothing was rendered from it, so withdraw the + optimistic pair and report the message id, which a retry reuses. + + The request itself may well have been accepted: the route never + reads `request.signal`, so it runs to completion regardless of + the abort. Reusing the id is what makes the retry safe — the + server deduplicates it against that turn instead of billing + another one. */ rollbackOptimisticSend() - return { kind: 'recoverable_cleanup_abort', streamId: userMessageId } + return { kind: 'withdrawn_by_cleanup', userMessageId } } return consumedByTranscript } @@ -3942,6 +3958,32 @@ export function useChat( setTransportStreaming, ] ) + /** + * Hands a send the unmount cleanup withdrew to whatever chat surface comes + * next: the live replacement's listener when one is mounted, else a one-shot + * stored handoff for the next mount. Both lanes carry `userMessageId`, so + * whoever picks it up retries as the same send rather than a new one. + */ + const handOffWithdrawnSend = useCallback( + (send: WithdrawnSend) => { + if ( + sendMothershipMessage(send.content, send.contexts, send.fileAttachments, send.userMessageId) + ) { + return + } + MothershipHandoffStorage.store( + { + message: send.content, + ...(send.contexts?.length ? { contexts: send.contexts } : {}), + ...(send.fileAttachments?.length ? { fileAttachments: send.fileAttachments } : {}), + resumeUserMessageId: send.userMessageId, + }, + workspaceId + ) + }, + [workspaceId] + ) + const sendMessage = useCallback( async ( message: string, @@ -3975,33 +4017,30 @@ export function useChat( queueStore.setEditing(activeChatKey, null) } - const queued = createQueuedMessage( - message, - fileAttachments, - contexts, - options?.recoverStreamId - ) - - if (sendingRef.current) { - queueStore.enqueue(activeChatKey, queued) + // An in-flight send drains the queue from `finalize`; a pending stop kicks + // the dispatcher itself, since nothing else will once the stop settles. + if (sendingRef.current || pendingStopPromiseRef.current) { + queueStore.enqueue( + activeChatKey, + createQueuedMessage(message, fileAttachments, contexts, options?.resumeUserMessageId) + ) + if (pendingStopPromiseRef.current) { + void enqueueQueueDispatchRef.current({ type: 'send_head' }) + } return } - if (pendingStopPromiseRef.current) { - queueStore.enqueue(activeChatKey, queued) - void enqueueQueueDispatchRef.current({ type: 'send_head' }) - return + const result = await startSendMessage(message, fileAttachments, contexts, options) + if (typeof result === 'object') { + handOffWithdrawnSend({ + content: message, + fileAttachments, + contexts, + userMessageId: result.userMessageId, + }) } - - /* Even an idle-path send goes through the durable queue: a direct - startSendMessage has no backing entry, so a cleanup abort mid-flight - silently drops it. The dispatch loop claims the head in the same tick - for a chatless send; a chat-bound one yields once on `cancelQueries` - first, so it can briefly show as queued. */ - queueStore.enqueue(activeChatKey, queued) - void enqueueQueueDispatchRef.current({ type: 'send_head' }) }, - [workspaceId, createQueuedMessage] + [workspaceId, createQueuedMessage, startSendMessage, handOffWithdrawnSend] ) useEffect(() => { if (typeof window === 'undefined') return @@ -4170,19 +4209,15 @@ export function useChat( const claimOwnerId = writeQueuedSendHandoffClaim(handoff.id) recoveringQueuedSendHandoffRef.current = { id: handoff.id, ownerId: claimOwnerId } - void startSendMessage( - handoff.message, - handoff.fileAttachments, - handoff.contexts, - null, - undefined, - { + void startSendMessage(handoff.message, handoff.fileAttachments, handoff.contexts, { + pendingStop: null, + queuedSendHandoff: { id: handoff.id, chatId: handoff.chatId, supersededStreamId: handoff.supersededStreamId, userMessageId: handoff.userMessageId, - } - ).finally(() => { + }, + }).finally(() => { if ( recoveringQueuedSendHandoffRef.current?.id === handoff.id && recoveringQueuedSendHandoffRef.current.ownerId === claimOwnerId @@ -4509,49 +4544,6 @@ export function useChat( ] ) - /** - * Answers "did the server accept the request that a cleanup abort withdrew?" - * by polling for the stream it would have registered. - * - * The abort tears down the client's socket but the route handler never reads - * `request.signal`, so an accepted request still creates the chat, persists - * the user message, and runs the turn. Re-sending in that case bills a second - * run and leaves the user with two chats, so recovery adopts the existing - * chat instead whenever this resolves one. - * - * @returns The chat the orphaned stream belongs to, or `undefined` when the - * server has no such stream (never accepted, or already gone) — in which case - * the caller re-sends. - */ - const resolveRecoveredSendChatId = useCallback( - async (streamId: string, epoch: number): Promise => { - const deadline = Date.now() + RECOVERED_SEND_PROBE_TIMEOUT_MS - while (true) { - const resolve = resolveDetachedChatForStreamRef.current - if (!resolve) return { status: 'not_found' } - /* "Superseded" is NOT the same answer as "the server has no such - stream", and the caller must not conflate them: adopting rewrites - the URL, and re-sending after an unmount would open a POST whose - abort controller the teardown has already dropped — a zombie request - nothing can cancel, duplicating the very send this recovery exists - to protect. Reported distinctly so the caller leaves the entry - queued for a later mount instead. */ - if (epoch !== queueDispatchEpochRef.current) return { status: 'superseded' } - const resolution = await resolve(streamId) - if (epoch !== queueDispatchEpochRef.current) return { status: 'superseded' } - if (resolution.chatId) return { status: 'adopted', chatId: resolution.chatId } - // A terminal status means the stream existed and finished without a - // durable owner; polling cannot improve on that. - if (resolution.terminal) return { status: 'not_found' } - if (Date.now() + RECOVERED_SEND_PROBE_INTERVAL_MS >= deadline) { - return { status: 'not_found' } - } - await sleep(RECOVERED_SEND_PROBE_INTERVAL_MS) - } - }, - [] - ) - const dispatchQueuedMessage = useCallback( async ( msg: QueuedMothershipMessage, @@ -4586,32 +4578,11 @@ export function useChat( useMothershipQueueStore.getState().remove(dispatchChatKey, msg.id) } - /** - * Hands a chatless send to whatever surface comes next, because its - * `pending::` queue key is regenerated per mount and anything left under - * this one is unreachable. Prefers the live replacement surface's - * listener and falls back to a one-shot stored handoff for a real - * navigation away. Both lanes carry attachments and the stream id, so the - * next surface probes before it sends. - */ - const handOffChatlessRecovery = (recoverStreamId?: string) => { - if ( - !sendMothershipMessage(msg.content, msg.contexts, msg.fileAttachments, recoverStreamId) - ) { - MothershipHandoffStorage.store( - { - message: msg.content, - ...(msg.contexts?.length ? { contexts: msg.contexts } : {}), - ...(msg.fileAttachments?.length ? { fileAttachments: msg.fileAttachments } : {}), - ...(recoverStreamId ? { recoverStreamId } : {}), - }, - workspaceId - ) - } - } - - const restoreQueuedMessage = (handoff?: QueuedSendHandoffSeed, recoverStreamId?: string) => { - const recoverableCleanupAbort = recoverStreamId !== undefined + const restoreQueuedMessage = ( + handoff?: QueuedSendHandoffSeed, + withdrawnUserMessageId?: string + ) => { + const withdrawnByCleanup = withdrawnUserMessageId !== undefined if (!handoff) { clearQueuedSendHandoffState(msg.id) } @@ -4619,7 +4590,7 @@ export function useChat( if (!removedFromQueue) { return } - if (options.epoch !== queueDispatchEpochRef.current && !recoverableCleanupAbort) { + if (options.epoch !== queueDispatchEpochRef.current && !withdrawnByCleanup) { return } // If the user explicitly removed this message during dispatch, honor @@ -4627,22 +4598,22 @@ export function useChat( if (userRemovedDuringDispatchRef.current.delete(msg.id)) { return } - /* A pending (chatless) surface regenerates its chat key per mount, and - the cleanup that aborted this send belongs to a full remount — a - queue restore would orphan the message under the dead instance's - key. Deliver to the replacement surface instead: its send listener - is live by the time this microtask executes. When nothing claims the - event (a real navigation away), a one-shot handoff covers the next - mount; both lanes carry attachments and the stream id to probe. - Chat-bound sends keep the queue restore — their key is the stable - chat id — and carry the stream id on the restored entry. */ - if (recoverableCleanupAbort && dispatchChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) { - handOffChatlessRecovery(recoverStreamId) + /* A chatless surface regenerates its queue key every mount, so a + 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)) { + handOffWithdrawnSend({ + content: msg.content, + fileAttachments: msg.fileAttachments, + contexts: msg.contexts, + userMessageId: withdrawnUserMessageId, + }) return } useMothershipQueueStore.getState().insertAt(dispatchChatKey, originalIndex, { ...msg, - ...(recoverStreamId ? { recoverStreamId } : {}), + ...(withdrawnUserMessageId ? { resumeUserMessageId: withdrawnUserMessageId } : {}), }) } @@ -4662,64 +4633,24 @@ export function useChat( const liveMsg = queueAtSend[currentIndex] activeQueuedSendHandoff = options.queuedSendHandoff ?? liveMsg.queuedSendHandoff - /* This entry is the recovery of a send a cleanup abort withdrew while - its request was already on the wire. The server never sees that - abort, so if it had accepted the request it created the chat and - persisted the message regardless — re-sending would duplicate both - the chat and the billed run. Probe the orphaned stream first and - adopt its chat instead when it exists. */ - if (liveMsg.recoverStreamId) { - const probe = await resolveRecoveredSendChatId(liveMsg.recoverStreamId, options.epoch) - /* Unknown, not "safe to send". A chat-bound key is the stable chat - id, so leaving the entry queued IS the retry — the next mount's - drain probes again. A chatless `pending::` key is regenerated per - mount, so the same move would strand the message under a dead key; - hand it to the recovery lanes instead, still carrying the stream - id. Skipped when the entry is no longer under this key (adoption - migrated it to a live chat), where it is already recoverable. */ - if (probe.status === 'superseded') { - if (dispatchChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) { - const queueStore = useMothershipQueueStore.getState() - const stillUnderDeadKey = (queueStore.queues[dispatchChatKey] ?? []).some( - (queued) => queued.id === msg.id - ) - if (stillUnderDeadKey) { - queueStore.remove(dispatchChatKey, msg.id) - handOffChatlessRecovery(liveMsg.recoverStreamId) - } - } - return - } - if (probe.status === 'adopted') { - removeQueuedMessage() - adoptResolvedChatId(probe.chatId, { - replaceHomeHistory: true, - invalidateList: true, - }) - /* Adoption alone does not surface the running turn. Hydration only - reconnects when `chatHistory.activeStreamId` is set, and that - query is cached for `MOTHERSHIP_CHAT_HISTORY_STALE_TIME` — on a - chat-bound recover the client usually holds a copy predating this - stream, so without an explicit detail invalidation the adopted - chat renders with the live response invisible. */ - invalidateChatQueries({ includeDetail: true, targetChatId: probe.chatId }) - return - } - } - const sendResult = await startSendMessage( liveMsg.content, liveMsg.fileAttachments, liveMsg.contexts, - options.pendingStop, - removeQueuedMessage, - activeQueuedSendHandoff + { + pendingStop: options.pendingStop, + onOptimisticSendApplied: removeQueuedMessage, + queuedSendHandoff: activeQueuedSendHandoff, + ...(liveMsg.resumeUserMessageId + ? { resumeUserMessageId: liveMsg.resumeUserMessageId } + : {}), + } ) if (sendResult !== true) { restoreQueuedMessage( activeQueuedSendHandoff, - typeof sendResult === 'object' ? sendResult.streamId : undefined + typeof sendResult === 'object' ? sendResult.userMessageId : undefined ) } } catch { @@ -4730,13 +4661,7 @@ export function useChat( userRemovedDuringDispatchRef.current.delete(msg.id) } }, - [ - startSendMessage, - workspaceId, - resolveRecoveredSendChatId, - adoptResolvedChatId, - invalidateChatQueries, - ] + [startSendMessage, handOffWithdrawnSend] ) const runQueueDispatchLoop = useCallback(async () => { diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/panel.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/panel.tsx index feba0bec19e..c8a924870fe 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/panel.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/panel.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) diff --git a/apps/sim/lib/copilot/chat/post.test.ts b/apps/sim/lib/copilot/chat/post.test.ts index 864cc5100f3..422b04fac34 100644 --- a/apps/sim/lib/copilot/chat/post.test.ts +++ b/apps/sim/lib/copilot/chat/post.test.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() + }) + }) }) diff --git a/apps/sim/lib/copilot/chat/post.ts b/apps/sim/lib/copilot/chat/post.ts index 4147aa87894..5be519852ee 100644 --- a/apps/sim/lib/copilot/chat/post.ts +++ b/apps/sim/lib/copilot/chat/post.ts @@ -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 { + 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)) { diff --git a/apps/sim/lib/core/idempotency/service.ts b/apps/sim/lib/core/idempotency/service.ts index d176df33d1b..35da0527309 100644 --- a/apps/sim/lib/core/idempotency/service.ts +++ b/apps/sim/lib/core/idempotency/service.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', +}) diff --git a/apps/sim/lib/core/utils/browser-storage.ts b/apps/sim/lib/core/utils/browser-storage.ts index 61776fa1618..d6f4c131bd2 100644 --- a/apps/sim/lib/core/utils/browser-storage.ts +++ b/apps/sim/lib/core/utils/browser-storage.ts @@ -311,12 +311,12 @@ export interface MothershipHandoff { /** Already-uploaded attachment references riding along with the message. */ fileAttachments?: FileAttachmentForApi[] /** - * Set only when a cleanup abort withdrew an in-flight send and this handoff - * is the recovery of it: the aborted send's stream id. The consuming chat - * probes it before sending, so a request the server had already accepted is - * adopted rather than sent a second time. + * Set only when this handoff is the recovery of a send an unmount cleanup + * withdrew: that attempt's message id. The consuming chat reuses it so the + * server deduplicates against the first attempt instead of opening a second + * chat and billing a second turn. */ - recoverStreamId?: string + resumeUserMessageId?: string } interface StoredHandoff extends MothershipHandoff { @@ -364,7 +364,7 @@ export class MothershipHandoffStorage { ? contexts : [...MothershipHandoffStorage.pendingContexts(workspaceId), ...contexts], ...(handoff.fileAttachments?.length ? { fileAttachments: handoff.fileAttachments } : {}), - ...(handoff.recoverStreamId ? { recoverStreamId: handoff.recoverStreamId } : {}), + ...(handoff.resumeUserMessageId ? { resumeUserMessageId: handoff.resumeUserMessageId } : {}), workspaceId, timestamp: Date.now(), }) @@ -427,8 +427,8 @@ export class MothershipHandoffStorage { ...(Array.isArray(data.fileAttachments) && data.fileAttachments.length > 0 ? { fileAttachments: data.fileAttachments } : {}), - ...(typeof data.recoverStreamId === 'string' && data.recoverStreamId - ? { recoverStreamId: data.recoverStreamId } + ...(typeof data.resumeUserMessageId === 'string' && data.resumeUserMessageId + ? { resumeUserMessageId: data.resumeUserMessageId } : {}), } } diff --git a/apps/sim/lib/mothership/events.ts b/apps/sim/lib/mothership/events.ts index 9259051e6ed..4ce4d844da8 100644 --- a/apps/sim/lib/mothership/events.ts +++ b/apps/sim/lib/mothership/events.ts @@ -28,12 +28,12 @@ export interface MothershipSendMessageDetail { /** Already-uploaded attachments riding along with the message. */ fileAttachments?: FileAttachmentForApi[] /** - * Set only when a cleanup abort withdrew an in-flight send and this is the - * recovery of it: the aborted send's stream id. The receiving chat probes it - * before sending, so a request the server had already accepted is adopted - * rather than sent a second time. + * Set only when this message is the recovery of a send an unmount cleanup + * withdrew: that attempt's message id. The receiving chat reuses it so the + * server deduplicates against the first attempt instead of opening a second + * chat and billing a second turn. */ - recoverStreamId?: string + resumeUserMessageId?: string } /** @@ -49,7 +49,7 @@ export function sendMothershipMessage( message: string, contexts?: ChatContext[], fileAttachments?: FileAttachmentForApi[], - recoverStreamId?: string + resumeUserMessageId?: string ): boolean { const trimmed = message.trim() if (!trimmed) { @@ -60,7 +60,7 @@ export function sendMothershipMessage( message: trimmed, contexts, fileAttachments, - ...(recoverStreamId ? { recoverStreamId } : {}), + ...(resumeUserMessageId ? { resumeUserMessageId } : {}), }) logger.info('Dispatched mothership message event', { messageLength: trimmed.length, consumed }) return consumed diff --git a/apps/sim/stores/mothership-queue/store.ts b/apps/sim/stores/mothership-queue/store.ts index a70eea369f0..4ecdbb89c61 100644 --- a/apps/sim/stores/mothership-queue/store.ts +++ b/apps/sim/stores/mothership-queue/store.ts @@ -99,12 +99,12 @@ export const useMothershipQueueStore = create()( const next = [...current] // Strip `queuedSendHandoff` — references the stream active at // original enqueue time; the dispatcher mints a fresh one at send. - // Strip `recoverStreamId` too: it dedupes against a server-side copy - // of the PRE-edit text, which the edited message is no longer a - // duplicate of, so probing it would wrongly suppress this send. + // Strip `resumeUserMessageId` too: it deduplicates against a + // server-side copy of the PRE-edit text, which the edited message is + // not a duplicate of, so reusing it would suppress this send. const { queuedSendHandoff: _stale, - recoverStreamId: _staleRecover, + resumeUserMessageId: _staleResume, ...rest } = next[index] next[index] = { diff --git a/apps/sim/stores/mothership-queue/types.ts b/apps/sim/stores/mothership-queue/types.ts index dde4808a422..b7beb4ede09 100644 --- a/apps/sim/stores/mothership-queue/types.ts +++ b/apps/sim/stores/mothership-queue/types.ts @@ -11,14 +11,13 @@ export interface QueuedSendHandoffSeed { export type QueuedMothershipMessage = QueuedMessage & { queuedSendHandoff?: QueuedSendHandoffSeed /** - * Stream id (the aborted send's `userMessageId`) of a send a cleanup abort - * withdrew while its request was already on the wire. The dispatcher probes - * it before re-sending: the server does not observe `request.signal`, so a - * request it had already accepted still creates the chat and persists the - * message, and re-sending blind would duplicate both. Persisted, so a - * nav-back restore probes too. + * Message id of a prior attempt at this send that an unmount cleanup + * withdrew. Reused when the entry is dispatched so the server deduplicates + * against that attempt — it never sees the client's abort, so a request it + * had already accepted still opened the chat and billed the turn. Persisted, + * so a retry after a reload deduplicates too. */ - recoverStreamId?: string + resumeUserMessageId?: string } // Mutable fields an in-place edit overwrites; id and index are preserved by `replaceAt`. From 98099a6087584a9aa8b22024b3c0a883d461d39c Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 11 Aug 2026 01:03:55 -0700 Subject: [PATCH 2/4] fix(chat): keep a withdrawn send in its own chat, and release stranded claims MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit follow-ups, two of them real defects in the previous commit. A withdrawn send routed unconditionally through the cross-surface lanes. Those deliver to whatever chat is mounted next, so sending in one chat and switching to another re-sent the message into the second one. The dispatcher already drew the distinction; the idle path now draws it too — a chat-bound key is the stable chat id, so re-queueing under it both retries durably and keeps the message where the user put it. Only a chatless key, which dies with its mount, goes to the lanes. The claim release sat in `catch`, so the two paths that return a response without throwing — a rejected branch, and a missing chat — stranded an in-progress claim for its full 60s TTL, and a retry inside that window got a spurious "already sent" instead of the real error. Moved to `finally`. Also: `userMessageId` is now length-bounded, since it becomes part of a Postgres key and an oversized one would throw inside the claim; `requestId` was still empty at claim time, so both dedup logs printed a blank prefix; the provider segment said `mothership` on a handler that also serves the workflow copilot, and now says what the key identifies; `retryFailures` was dead config, only read by `executeWithIdempotency`, which this caller never invokes; the doc pointed at `billingIdempotency`, which has no consumers, and now points at the live Stripe analogue. Trimmed: `sendClaimRecorded` folded into clearing `sendClaim`, the unread `kind` discriminant dropped from a one-arm union, the single-use `claimedChatId` inlined, and the prose on all three of those cut back to what the code does not already say. --- .../home/hooks/use-chat.mount-send.test.tsx | 65 +++++++++++++ .../[workspaceId]/home/hooks/use-chat.ts | 46 ++++++--- apps/sim/lib/copilot/chat/post.test.ts | 8 +- apps/sim/lib/copilot/chat/post.ts | 95 ++++++++----------- apps/sim/lib/core/idempotency/service.ts | 9 +- 5 files changed, 147 insertions(+), 76 deletions(-) diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx index 185127cc2f5..48e7b83e235 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.mount-send.test.tsx @@ -145,6 +145,43 @@ function renderUseChat(): { } } +/** + * As `renderUseChat`, but bound to an existing chat rather than chatless. The + * pathname has to match: the hook resets a chat-bound surface back to a fresh + * pending key when it finds itself on the home route. + */ +function renderUseChatInChat(chatId: string): { + getResult: () => ReturnType + unmount: () => void +} { + navigationMocks.usePathname.mockReturnValue(`/workspace/ws-1/chat/${chatId}`) + ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true + queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }) + const container = document.createElement('div') + const root = createRoot(container) + mountedRoots.push(root) + let result: ReturnType | undefined + + function Probe() { + result = useChat('ws-1', chatId) + return null + } + + act(() => { + root.render( + {() as ReactNode} + ) + }) + + return { + getResult: () => { + if (result === undefined) throw new Error('Hook result is not ready') + return result + }, + unmount: () => act(() => root.unmount()), + } +} + /** * Mounts a surface shaped like `home.tsx`: it drives `useChat` AND registers the * `mothership-send-message` listener that claims the event with @@ -260,6 +297,7 @@ async function waitFor(predicate: () => boolean, budgetMs = 2000): Promise describe('useChat remount send recovery', () => { beforeEach(() => { vi.stubGlobal('fetch', fetchStub) + navigationMocks.usePathname.mockReturnValue('/workspace/ws-1/home') state.postBehavior = 'hang' state.postBodies = [] mockRequestJson.mockResolvedValue({ chats: [] }) @@ -444,4 +482,31 @@ describe('useChat remount send recovery', () => { expect(state.postBodies[0].userMessageId).toBe('the-first-attempt') }) }) + + /** + * 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 + * send through them would drop the message into a different conversation — + * exactly what happens if the user switches chats mid-send. Its key is the + * stable chat id, so re-queueing under that key is the durable retry. + */ + it('re-queues a withdrawn chat-bound send instead of following the user', async () => { + const { getResult, unmount } = renderUseChatInChat('chat-a') + + await act(async () => { + void getResult().sendMessage('belongs to chat-a') + }) + await waitFor(() => state.postBodies.length === 1) + + unmount() + await waitFor(() => allQueuedMessages().length === 1) + + const queues = useMothershipQueueStore.getState().queues + expect(Object.keys(queues)).toEqual(['chat-a']) + expect(queues['chat-a'][0].content).toBe('belongs to chat-a') + // Reused on the retry so the server deduplicates it. + expect(queues['chat-a'][0].resumeUserMessageId).toBe(state.postBodies[0].userMessageId) + // Must NOT have gone to the cross-surface handoff. + expect(MothershipHandoffStorage.consume('ws-1')).toBeNull() + }) }) diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts index 531bf9f5080..fe4d1d582a0 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts @@ -151,7 +151,7 @@ export interface SendMessageOptions { * when an unmount cleanup withdrew it — `userMessageId` is what a retry reuses * so the server deduplicates the two attempts. */ -type StartSendMessageResult = boolean | { kind: 'withdrawn_by_cleanup'; userMessageId: string } +type StartSendMessageResult = boolean | { userMessageId: string } interface StartSendMessageOptions { /** Awaited before dispatch. Defaults to the hook's in-flight stop, if any. */ @@ -3905,7 +3905,7 @@ export function useChat( server deduplicates it against that turn instead of billing another one. */ rollbackOptimisticSend() - return { kind: 'withdrawn_by_cleanup', userMessageId } + return { userMessageId } } return consumedByTranscript } @@ -4031,14 +4031,29 @@ export function useChat( } const result = await startSendMessage(message, fileAttachments, contexts, options) - if (typeof result === 'object') { - handOffWithdrawnSend({ - content: message, - fileAttachments, - contexts, - userMessageId: result.userMessageId, - }) + 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. */ + const withdrawn = { + content: message, + fileAttachments, + contexts, + userMessageId: result.userMessageId, } + if (activeChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) { + handOffWithdrawnSend(withdrawn) + return + } + useMothershipQueueStore + .getState() + .enqueue( + activeChatKey, + createQueuedMessage(message, fileAttachments, contexts, result.userMessageId) + ) }, [workspaceId, createQueuedMessage, startSendMessage, handOffWithdrawnSend] ) @@ -4578,6 +4593,10 @@ export function useChat( useMothershipQueueStore.getState().remove(dispatchChatKey, msg.id) } + /* What actually went out. `msg` is the snapshot from when the dispatch was + scheduled; the send below uses the re-read live entry, so recovery + tracks that rather than assuming the two still match. */ + let dispatched = msg const restoreQueuedMessage = ( handoff?: QueuedSendHandoffSeed, withdrawnUserMessageId?: string @@ -4604,15 +4623,15 @@ export function useChat( the queue itself is the durable retry. */ if (withdrawnByCleanup && dispatchChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) { handOffWithdrawnSend({ - content: msg.content, - fileAttachments: msg.fileAttachments, - contexts: msg.contexts, + content: dispatched.content, + fileAttachments: dispatched.fileAttachments, + contexts: dispatched.contexts, userMessageId: withdrawnUserMessageId, }) return } useMothershipQueueStore.getState().insertAt(dispatchChatKey, originalIndex, { - ...msg, + ...dispatched, ...(withdrawnUserMessageId ? { resumeUserMessageId: withdrawnUserMessageId } : {}), }) } @@ -4631,6 +4650,7 @@ export function useChat( // Re-read live: the user may have applied an in-place edit (`replaceAt`) // between dispatch scheduling and this send. const liveMsg = queueAtSend[currentIndex] + dispatched = liveMsg activeQueuedSendHandoff = options.queuedSendHandoff ?? liveMsg.queuedSendHandoff const sendResult = await startSendMessage( diff --git a/apps/sim/lib/copilot/chat/post.test.ts b/apps/sim/lib/copilot/chat/post.test.ts index 422b04fac34..b69a06ed1fd 100644 --- a/apps/sim/lib/copilot/chat/post.test.ts +++ b/apps/sim/lib/copilot/chat/post.test.ts @@ -148,7 +148,7 @@ describe('handleUnifiedChatPost', () => { resetDbChainMock() atomicallyClaimChatSend.mockResolvedValue({ claimed: true, - normalizedKey: 'chat-send:mothership:msg-1:userId=user-1', + normalizedKey: 'chat-send:user-message:msg-1:userId=user-1', storageMethod: 'database', claimToken: 'claim-1', }) @@ -758,7 +758,7 @@ describe('handleUnifiedChatPost', () => { 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', + normalizedKey: 'chat-send:user-message:msg-1:userId=user-1', storageMethod: 'database', existingResult: { success: true, status: 'completed', result: { chatId: 'chat-first' } }, }) @@ -798,7 +798,7 @@ describe('handleUnifiedChatPost', () => { }) ) - expect(atomicallyClaimChatSend).toHaveBeenCalledWith('mothership', 'msg-1', { + expect(atomicallyClaimChatSend).toHaveBeenCalledWith('user-message', 'msg-1', { userId: 'user-1', }) }) @@ -817,7 +817,7 @@ describe('handleUnifiedChatPost', () => { ) expect(storeChatSendResult).toHaveBeenCalledWith( - 'chat-send:mothership:msg-1:userId=user-1', + 'chat-send:user-message:msg-1:userId=user-1', expect.objectContaining({ result: { chatId: 'chat-1' } }), 'database', 'claim-1' diff --git a/apps/sim/lib/copilot/chat/post.ts b/apps/sim/lib/copilot/chat/post.ts index 5be519852ee..bc830256ae4 100644 --- a/apps/sim/lib/copilot/chat/post.ts +++ b/apps/sim/lib/copilot/chat/post.ts @@ -260,7 +260,10 @@ const ChatContextSchema = z const ChatMessageSchema = z.object({ message: z.string().min(1, 'Message is required'), - userMessageId: z.string().optional(), + /* Bounded because it becomes part of a Postgres key in `chatSendIdempotency`; + a client-supplied id longer than the btree entry limit would throw there. + A generated id is 36 chars. */ + userMessageId: z.string().max(128).optional(), chatId: z.string().optional(), workflowId: z.string().optional(), workspaceId: z.string().optional(), @@ -945,26 +948,22 @@ async function resolveBranch(params: { } } -/** Namespace-local provider segment for {@link chatSendIdempotency} keys. */ -const CHAT_SEND_IDEMPOTENCY_PROVIDER = 'mothership' +/** Names what the key identifies: `chat-send:user-message::userId=`. */ +const CHAT_SEND_IDEMPOTENCY_PROVIDER = 'user-message' /** * 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. + * Fails open: a missed deduplication costs a duplicate chat and turn, but + * refusing the send loses the user's message. Returns `undefined` when the + * store is unreachable, 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. + * The key is scoped to the caller — `userMessageId` is client-supplied, so an + * unscoped one would let a user probe another's sends for their chat id. */ async function claimChatSend( userMessageId: string, - userId: string, - requestId: string + userId: string ): Promise { try { return await chatSendIdempotency.atomicallyClaim( @@ -973,7 +972,7 @@ async function claimChatSend( { userId } ) } catch (error) { - logger.warn(`[${requestId}] Could not claim chat send; proceeding without deduplication`, { + logger.warn('Could not claim chat send; proceeding without deduplication', { userMessageId, error: getErrorMessage(error, 'Unknown error'), }) @@ -981,12 +980,6 @@ async function claimChatSend( } } -/** 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. * @@ -997,13 +990,10 @@ function claimedChatId(claim: AtomicClaimResult): string | undefined { * 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 }) +function duplicateChatSendResponse(claim: AtomicClaimResult, userMessageId: string): NextResponse { + const claimed = claim.existingResult?.result?.chatId + const chatId = typeof claimed === 'string' && claimed ? claimed : undefined + logger.info('Deduplicated a repeated chat send', { userMessageId, chatId }) return NextResponse.json( { error: 'This message was already sent.', @@ -1018,13 +1008,8 @@ 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. */ + /** Cleared once the chat is recorded against it, which makes it permanent. */ 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 @@ -1059,9 +1044,9 @@ export async function handleUnifiedChatPost(req: NextRequest) { const normalizedContexts = normalizeContexts(body.contexts) ?? [] userMessageId = body.userMessageId || generateId() - sendClaim = await claimChatSend(userMessageId, authenticatedUserId, requestId) + sendClaim = await claimChatSend(userMessageId, authenticatedUserId) if (sendClaim?.claimed === false) { - return duplicateChatSendResponse(sendClaim, userMessageId, requestId) + return duplicateChatSendResponse(sendClaim, userMessageId) } otelRoot = startCopilotOtelRoot({ @@ -1156,13 +1141,12 @@ 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. */ + /* Record the chat as soon as it is known — the earliest a retry can be + answered with somewhere to go. The claim becomes permanent here: the + user message lands moments later, so a retry must resolve to this chat + rather than open another. */ if (sendClaim?.claimToken && actualChatId) { - sendClaimRecorded = await chatSendIdempotency + const recorded = await chatSendIdempotency .storeResult( sendClaim.normalizedKey, { success: true, status: 'completed', result: { chatId: actualChatId } }, @@ -1176,6 +1160,7 @@ export async function handleUnifiedChatPost(req: NextRequest) { }) return false }) + if (recorded) sendClaim = undefined } if (chatIsNew && actualChatId && body.resourceAttachments?.length) { @@ -1482,19 +1467,6 @@ 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)) { @@ -1524,5 +1496,20 @@ export async function handleUnifiedChatPost(req: NextRequest) { }, { status: 500 } ) + } finally { + /* A claim still held here never recorded a chat — the send threw, or + returned early on a rejected branch or a missing chat. Release it so a + retry is treated as new rather than deduplicated against a chat that was + never opened. Must be `finally`: those early returns skip `catch`. */ + if (sendClaim?.claimToken) { + await chatSendIdempotency + .release(sendClaim.normalizedKey, sendClaim.storageMethod, sendClaim.claimToken) + .catch((releaseError) => { + logger.warn('Could not release the claim for an unfinished send', { + userMessageId, + error: getErrorMessage(releaseError, 'Unknown error'), + }) + }) + } } } diff --git a/apps/sim/lib/core/idempotency/service.ts b/apps/sim/lib/core/idempotency/service.ts index 35da0527309..43fa6cf9feb 100644 --- a/apps/sim/lib/core/idempotency/service.ts +++ b/apps/sim/lib/core/idempotency/service.ts @@ -744,10 +744,10 @@ export const billingIdempotency = new IdempotencyService({ * 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. + * Storage is forced to Postgres for the same reason Stripe webhook claims are + * (see `stripeWebhookIdempotency`): a missed deduplication 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 that follows. * * `inProgressTtlSeconds` is short so a crashed pod cannot block a genuine retry * for the full hour, while completed sends stay deduplicated for it. @@ -756,6 +756,5 @@ export const chatSendIdempotency = new IdempotencyService({ namespace: 'chat-send', ttlSeconds: 60 * 60, // 1 hour inProgressTtlSeconds: 60, - retryFailures: true, forceStorage: 'database', }) From d7a16abb50052f0041048203b0f675db53a2e9d1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 11 Aug 2026 01:07:06 -0700 Subject: [PATCH 3/4] fix(chat): make a send's claim permanent only once its turn starts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The claim became permanent as soon as the chat resolved, but three exits still return without starting a turn — a rejected branch, a missing chat, and a pending-stream collision. The last one matters: the queued-send-handoff path deliberately retries under the original `userMessageId` after a collision, and against a permanent claim that retry deduplicated to a chat whose turn never ran, reattaching to a stream that does not exist. A send that had merely collided became unsendable for the claim's full hour. The claim is now dropped immediately before the stream response is returned, so `finally` releases it on every other exit. Recording the chat still happens as early as possible — a concurrent duplicate needs somewhere to go — it just no longer implies the turn happened. --- apps/sim/lib/copilot/chat/post.test.ts | 47 ++++++++++++++++++++++++++ apps/sim/lib/copilot/chat/post.ts | 23 ++++++++----- 2 files changed, 62 insertions(+), 8 deletions(-) diff --git a/apps/sim/lib/copilot/chat/post.test.ts b/apps/sim/lib/copilot/chat/post.test.ts index b69a06ed1fd..b9586520232 100644 --- a/apps/sim/lib/copilot/chat/post.test.ts +++ b/apps/sim/lib/copilot/chat/post.test.ts @@ -847,5 +847,52 @@ describe('handleUnifiedChatPost', () => { expect(createSSEStream).toHaveBeenCalled() expect(storeChatSendResult).not.toHaveBeenCalled() }) + + /** + * The queued-send-handoff path deliberately retries under the original + * `userMessageId` after a stream collision. If the collided attempt left a + * permanent claim, that retry would deduplicate against a chat whose turn + * never started and reattach to a stream that does not exist. + */ + it('releases the claim when a stream collision stops the turn from starting', async () => { + acquirePendingChatStream.mockResolvedValue(false) + getPendingChatStreamId.mockResolvedValue('other-stream') + + 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) + expect(releaseChatSendClaim).toHaveBeenCalledWith( + 'chat-send:user-message:msg-1:userId=user-1', + 'database', + 'claim-1' + ) + }) + + it('keeps the claim once a turn is actually streaming', async () => { + 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(releaseChatSendClaim).not.toHaveBeenCalled() + }) }) }) diff --git a/apps/sim/lib/copilot/chat/post.ts b/apps/sim/lib/copilot/chat/post.ts index bc830256ae4..fe32a505465 100644 --- a/apps/sim/lib/copilot/chat/post.ts +++ b/apps/sim/lib/copilot/chat/post.ts @@ -1142,9 +1142,10 @@ export async function handleUnifiedChatPost(req: NextRequest) { } /* Record the chat as soon as it is known — the earliest a retry can be - answered with somewhere to go. The claim becomes permanent here: the - user message lands moments later, so a retry must resolve to this chat - rather than open another. */ + answered with somewhere to go. The claim is not permanent yet: several + exits below still return without starting a turn, and a retry of those + must be allowed to start one. Recording only fails open, so drop the + claim when it does. */ if (sendClaim?.claimToken && actualChatId) { const recorded = await chatSendIdempotency .storeResult( @@ -1160,7 +1161,7 @@ export async function handleUnifiedChatPost(req: NextRequest) { }) return false }) - if (recorded) sendClaim = undefined + if (!recorded) sendClaim = undefined } if (chatIsNew && actualChatId && body.resourceAttachments?.length) { @@ -1456,6 +1457,11 @@ export async function handleUnifiedChatPost(req: NextRequest) { const rootTraceparent = `00-${rootCtx.traceId}-${rootCtx.spanId}-${ (rootCtx.traceFlags & 0x1) === 0x1 ? '01' : '00' }` + /* A turn is running. Only now is the claim permanent, so the `finally` + below leaves it in place and a retry of this send resolves to this + chat instead of opening another. Every earlier exit returns without a + turn, and releases. */ + sendClaim = undefined return new Response(stream, { headers: { ...SSE_RESPONSE_HEADERS, @@ -1497,10 +1503,11 @@ export async function handleUnifiedChatPost(req: NextRequest) { { status: 500 } ) } finally { - /* A claim still held here never recorded a chat — the send threw, or - returned early on a rejected branch or a missing chat. Release it so a - retry is treated as new rather than deduplicated against a chat that was - never opened. Must be `finally`: those early returns skip `catch`. */ + /* A claim still held here never started a turn — the send threw, or + returned early on a rejected branch, a missing chat, or a chat that + already has a stream running. Release it so a retry may start one rather + than deduplicating against a turn that never happened. Must be + `finally`: those early returns skip `catch`. */ if (sendClaim?.claimToken) { await chatSendIdempotency .release(sendClaim.normalizedKey, sendClaim.storageMethod, sendClaim.claimToken) From 9ab60facfec5c37667297d3f1c37ebceba022591 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Tue, 11 Aug 2026 01:10:09 -0700 Subject: [PATCH 4/4] refactor(chat): give the send claim a single point of permanence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recording the chat also dropped the claim when it failed, which left a second way for a claim to stop being tracked and a compound hole behind it: a failed record followed by a throw stranded the claim for its in-progress TTL, and a retry inside that window reattached to a turn that never started. Only one line now decides permanence — the claim is cleared immediately before the stream response — so `finally` releases it on every exit that did not start a turn, including a failed record. The `recorded` flag is gone with it. Covers the 400 early return with a release assertion: that path returns without throwing, so it is the one that proves the release has to live in `finally`. --- apps/sim/lib/copilot/chat/post.test.ts | 2 ++ apps/sim/lib/copilot/chat/post.ts | 12 +++++------- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/apps/sim/lib/copilot/chat/post.test.ts b/apps/sim/lib/copilot/chat/post.test.ts index b9586520232..4f8efd52029 100644 --- a/apps/sim/lib/copilot/chat/post.test.ts +++ b/apps/sim/lib/copilot/chat/post.test.ts @@ -743,6 +743,8 @@ describe('handleUnifiedChatPost', () => { ) expect(response.status).toBe(400) + // Returns without throwing, so only a `finally` can free the claim. + expect(releaseChatSendClaim).toHaveBeenCalled() await expect(response.json()).resolves.toMatchObject({ error: 'workspaceId is required when workflowId is not provided', }) diff --git a/apps/sim/lib/copilot/chat/post.ts b/apps/sim/lib/copilot/chat/post.ts index fe32a505465..0bde82a6c45 100644 --- a/apps/sim/lib/copilot/chat/post.ts +++ b/apps/sim/lib/copilot/chat/post.ts @@ -1142,12 +1142,12 @@ export async function handleUnifiedChatPost(req: NextRequest) { } /* Record the chat as soon as it is known — the earliest a retry can be - answered with somewhere to go. The claim is not permanent yet: several - exits below still return without starting a turn, and a retry of those - must be allowed to start one. Recording only fails open, so drop the - claim when it does. */ + answered with somewhere to go. This does not make the claim permanent: + several exits below still return without starting a turn, and a retry + of those must be free to start one. Failing to record only costs a + retry the chat-id shortcut, so it must not fail the send. */ if (sendClaim?.claimToken && actualChatId) { - const recorded = await chatSendIdempotency + await chatSendIdempotency .storeResult( sendClaim.normalizedKey, { success: true, status: 'completed', result: { chatId: actualChatId } }, @@ -1159,9 +1159,7 @@ export async function handleUnifiedChatPost(req: NextRequest) { userMessageId, error: getErrorMessage(error, 'Unknown error'), }) - return false }) - if (!recorded) sendClaim = undefined } if (chatIsNew && actualChatId && body.resourceAttachments?.length) {