Skip to content

Commit f8644cc

Browse files
waleedlatif1waleedlatif1
authored
fix(chat): deduplicate chat sends server-side instead of probing for them (#6536)
* fix(chat): deduplicate chat sends server-side instead of probing for them A client cannot tell whether a request it aborted reached the server: the chat route never reads `request.signal`, so an accepted one still opens the chat, persists the user message, and bills the turn after the socket drops. #6525 answered that by polling the orphaned stream before retrying — a 2.5s guess that had to distinguish "no such stream" from "we stopped looking", and still left a window open. The codebase already owns the right tool. `IdempotencyService` backs webhook, polling, and billing dedup, and `billingIdempotency` exists for exactly this hazard: "a retry would double-record usage — real money". Chat sends now claim the same way, keyed on the client-generated `userMessageId` and scoped to the caller so nobody can probe another user's sends. A repeat gets 409 naming the chat the first attempt opened — deliberately the shape the pending-stream lock already returns, so the client's existing conflict handler reattaches instead of starting a turn, with only the chat-adoption line added. The claim fails open at every step. Deduplication saves a duplicate chat; the send IS the user's message, so an unreachable bookkeeping store degrades chat rather than taking it down. It is released when a send fails before recording a chat, and deliberately kept once recorded. Retrying now just reuses the id, which deletes the probe outright: the poll and its two constants, the three-state result, the epoch plumbing that kept a superseded poll from re-sending, and the chat-adoption branch it needed. The client hook nets 67 lines smaller. Idle sends go back to calling `startSendMessage` directly. #6525 routed them through the durable queue so recovery had a backing entry, which put every message in the product through the queue store, sessionStorage, and the dispatch loop for the sake of a rare path — and the recovery never needed it, since the message, attachments, contexts, and id are all in scope at the abort. Both callers now share one `handOffWithdrawnSend`. `startSendMessage` takes its optional tail as an options object; it was at six positional parameters and the retry id would have been a seventh. Tests cover both halves: the server dedups, scopes the key per user, records the chat, and still sends when the claim store is down; the client reuses the original id on retry and adopts the chat a deduplicated retry names. Each was confirmed red without its fix. * fix(chat): keep a withdrawn send in its own chat, and release stranded claims 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. * fix(chat): make a send's claim permanent only once its turn starts 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. * refactor(chat): give the send claim a single point of permanence 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`.
1 parent 783e1b5 commit f8644cc

11 files changed

Lines changed: 681 additions & 471 deletions

File tree

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -342,7 +342,7 @@ export function Home({ chatId, userName, userId, tableViewsEnabled }: HomeProps)
342342
if (!detail?.message) return
343343
e.preventDefault()
344344
sendMessage(detail.message, detail.fileAttachments, detail.contexts, {
345-
...(detail.recoverStreamId ? { recoverStreamId: detail.recoverStreamId } : {}),
345+
...(detail.resumeUserMessageId ? { resumeUserMessageId: detail.resumeUserMessageId } : {}),
346346
})
347347
}
348348
window.addEventListener(MOTHERSHIP_SEND_MESSAGE_EVENT, handler)
@@ -373,7 +373,9 @@ export function Home({ chatId, userName, userId, tableViewsEnabled }: HomeProps)
373373
if (!handoff) return
374374
if (handoff.message) {
375375
sendMessage(handoff.message, handoff.fileAttachments, handoff.contexts, {
376-
...(handoff.recoverStreamId ? { recoverStreamId: handoff.recoverStreamId } : {}),
376+
...(handoff.resumeUserMessageId
377+
? { resumeUserMessageId: handoff.resumeUserMessageId }
378+
: {}),
377379
})
378380
return
379381
}

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

Lines changed: 183 additions & 226 deletions
Large diffs are not rendered by default.

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

Lines changed: 160 additions & 215 deletions
Large diffs are not rendered by default.

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -490,7 +490,7 @@ export const Panel = memo(function Panel() {
490490
e.preventDefault()
491491
setActiveTab('copilot')
492492
copilotSendMessage(detail.message, detail.fileAttachments, detail.contexts, {
493-
...(detail.recoverStreamId ? { recoverStreamId: detail.recoverStreamId } : {}),
493+
...(detail.resumeUserMessageId ? { resumeUserMessageId: detail.resumeUserMessageId } : {}),
494494
})
495495
}
496496
window.addEventListener(MOTHERSHIP_SEND_MESSAGE_EVENT, handler)

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

Lines changed: 172 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,9 @@ const {
3636
appendCopilotChatMessages,
3737
persistChatResources,
3838
mockPublishStatusChanged,
39+
atomicallyClaimChatSend,
40+
storeChatSendResult,
41+
releaseChatSendClaim,
3942
} = vi.hoisted(() => ({
4043
generateWorkspaceSnapshot: vi.fn(),
4144
processContextsServer: vi.fn(),
@@ -51,6 +54,9 @@ const {
5154
appendCopilotChatMessages: vi.fn(),
5255
persistChatResources: vi.fn(),
5356
mockPublishStatusChanged: vi.fn(),
57+
atomicallyClaimChatSend: vi.fn(),
58+
storeChatSendResult: vi.fn(),
59+
releaseChatSendClaim: vi.fn(),
5460
}))
5561

5662
const getSession = authMockFns.mockGetSession
@@ -103,6 +109,14 @@ vi.mock('@/lib/copilot/chat/lifecycle', () => ({
103109
resolveOrCreateChat,
104110
}))
105111

112+
vi.mock('@/lib/core/idempotency', () => ({
113+
chatSendIdempotency: {
114+
atomicallyClaim: atomicallyClaimChatSend,
115+
storeResult: storeChatSendResult,
116+
release: releaseChatSendClaim,
117+
},
118+
}))
119+
106120
vi.mock('@/lib/copilot/chat/terminal-state', () => ({
107121
finalizeAssistantTurn,
108122
}))
@@ -132,6 +146,14 @@ describe('handleUnifiedChatPost', () => {
132146
beforeEach(() => {
133147
vi.clearAllMocks()
134148
resetDbChainMock()
149+
atomicallyClaimChatSend.mockResolvedValue({
150+
claimed: true,
151+
normalizedKey: 'chat-send:user-message:msg-1:userId=user-1',
152+
storageMethod: 'database',
153+
claimToken: 'claim-1',
154+
})
155+
storeChatSendResult.mockResolvedValue(true)
156+
releaseChatSendClaim.mockResolvedValue(undefined)
135157
getSession.mockResolvedValue({ user: { id: 'user-1' } })
136158
resolveWorkflowIdForUser.mockResolvedValue({
137159
status: 'resolved',
@@ -721,8 +743,158 @@ describe('handleUnifiedChatPost', () => {
721743
)
722744

723745
expect(response.status).toBe(400)
746+
// Returns without throwing, so only a `finally` can free the claim.
747+
expect(releaseChatSendClaim).toHaveBeenCalled()
724748
await expect(response.json()).resolves.toMatchObject({
725749
error: 'workspaceId is required when workflowId is not provided',
726750
})
727751
})
752+
753+
describe('deduplicating a repeated send', () => {
754+
/**
755+
* The client cannot tell whether a request it aborted reached the server —
756+
* the route never reads `request.signal`, so an accepted one runs to
757+
* completion regardless. Recovering such a send therefore retries it under
758+
* the original `userMessageId`, and this is what makes that safe.
759+
*/
760+
it('answers an already-claimed send with the chat the first attempt opened', async () => {
761+
atomicallyClaimChatSend.mockResolvedValue({
762+
claimed: false,
763+
normalizedKey: 'chat-send:user-message:msg-1:userId=user-1',
764+
storageMethod: 'database',
765+
existingResult: { success: true, status: 'completed', result: { chatId: 'chat-first' } },
766+
})
767+
768+
const response = await handleUnifiedChatPost(
769+
new NextRequest('http://localhost/api/mothership/chat', {
770+
method: 'POST',
771+
body: JSON.stringify({
772+
message: 'Hello',
773+
workspaceId: 'ws-1',
774+
userMessageId: 'msg-1',
775+
createNewChat: true,
776+
}),
777+
})
778+
)
779+
780+
expect(response.status).toBe(409)
781+
await expect(response.json()).resolves.toMatchObject({
782+
activeStreamId: 'msg-1',
783+
chatId: 'chat-first',
784+
})
785+
// The whole point: no second chat, no second billed turn.
786+
expect(resolveOrCreateChat).not.toHaveBeenCalled()
787+
expect(createSSEStream).not.toHaveBeenCalled()
788+
})
789+
790+
it('scopes the claim to the caller so one user cannot probe another', async () => {
791+
await handleUnifiedChatPost(
792+
new NextRequest('http://localhost/api/mothership/chat', {
793+
method: 'POST',
794+
body: JSON.stringify({
795+
message: 'Hello',
796+
workspaceId: 'ws-1',
797+
userMessageId: 'msg-1',
798+
createNewChat: true,
799+
}),
800+
})
801+
)
802+
803+
expect(atomicallyClaimChatSend).toHaveBeenCalledWith('user-message', 'msg-1', {
804+
userId: 'user-1',
805+
})
806+
})
807+
808+
it('records the chat against the send so a retry resolves to it', async () => {
809+
await handleUnifiedChatPost(
810+
new NextRequest('http://localhost/api/mothership/chat', {
811+
method: 'POST',
812+
body: JSON.stringify({
813+
message: 'Hello',
814+
workspaceId: 'ws-1',
815+
userMessageId: 'msg-1',
816+
createNewChat: true,
817+
}),
818+
})
819+
)
820+
821+
expect(storeChatSendResult).toHaveBeenCalledWith(
822+
'chat-send:user-message:msg-1:userId=user-1',
823+
expect.objectContaining({ result: { chatId: 'chat-1' } }),
824+
'database',
825+
'claim-1'
826+
)
827+
})
828+
829+
/**
830+
* Deduplication saves a duplicate chat; the send IS the user's message.
831+
* An unreachable bookkeeping store must degrade chat, never take it down.
832+
*/
833+
it('sends normally when the claim store is unavailable', async () => {
834+
atomicallyClaimChatSend.mockRejectedValue(new Error('idempotency store down'))
835+
836+
const response = await handleUnifiedChatPost(
837+
new NextRequest('http://localhost/api/mothership/chat', {
838+
method: 'POST',
839+
body: JSON.stringify({
840+
message: 'Hello',
841+
workspaceId: 'ws-1',
842+
userMessageId: 'msg-1',
843+
createNewChat: true,
844+
}),
845+
})
846+
)
847+
848+
expect(response.status).toBe(200)
849+
expect(createSSEStream).toHaveBeenCalled()
850+
expect(storeChatSendResult).not.toHaveBeenCalled()
851+
})
852+
853+
/**
854+
* The queued-send-handoff path deliberately retries under the original
855+
* `userMessageId` after a stream collision. If the collided attempt left a
856+
* permanent claim, that retry would deduplicate against a chat whose turn
857+
* never started and reattach to a stream that does not exist.
858+
*/
859+
it('releases the claim when a stream collision stops the turn from starting', async () => {
860+
acquirePendingChatStream.mockResolvedValue(false)
861+
getPendingChatStreamId.mockResolvedValue('other-stream')
862+
863+
const response = await handleUnifiedChatPost(
864+
new NextRequest('http://localhost/api/mothership/chat', {
865+
method: 'POST',
866+
body: JSON.stringify({
867+
message: 'Hello',
868+
workspaceId: 'ws-1',
869+
userMessageId: 'msg-1',
870+
createNewChat: true,
871+
}),
872+
})
873+
)
874+
875+
expect(response.status).toBe(409)
876+
expect(releaseChatSendClaim).toHaveBeenCalledWith(
877+
'chat-send:user-message:msg-1:userId=user-1',
878+
'database',
879+
'claim-1'
880+
)
881+
})
882+
883+
it('keeps the claim once a turn is actually streaming', async () => {
884+
const response = await handleUnifiedChatPost(
885+
new NextRequest('http://localhost/api/mothership/chat', {
886+
method: 'POST',
887+
body: JSON.stringify({
888+
message: 'Hello',
889+
workspaceId: 'ws-1',
890+
userMessageId: 'msg-1',
891+
createNewChat: true,
892+
}),
893+
})
894+
)
895+
896+
expect(response.status).toBe(200)
897+
expect(releaseChatSendClaim).not.toHaveBeenCalled()
898+
})
899+
})
728900
})

0 commit comments

Comments
 (0)