Skip to content

Commit 07b32dd

Browse files
committed
fix(mothership): decide once whether a send's id may already be on the server
Whether a queued message may already be a turn on the server was inferred in several places (the store guard, a stopRequired exemption, the handoff restore, notAdmitted/neverSent), and the stopRequired exemption was wrong: a Send-now reuses a re-queued message's earlier id, which may have been sent. startSendMessage now decides it where it chooses the id (a reused id carries its entry's flag, a fresh one is unsent until its POST, a refusal clears it) and carries that one fact on the withdrawal result, the stored handoff and the queue entry. The store guard reads only the flag, filling it in only for writers with no say.
1 parent 159d115 commit 07b32dd

6 files changed

Lines changed: 188 additions & 50 deletions

File tree

‎apps/sim/app/workspace/[workspaceId]/home/hooks/send-handoff.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,8 @@ export interface QueuedSendHandoffState extends QueuedSendHandoffSeed {
3939
assistantSearchLevel?: AssistantSearchLevel
4040
requestedAt: number
4141
resolveAttempts?: number
42+
/** Whether the server may already hold `userMessageId` (see `startSendMessage`). */
43+
admissionUnknown?: boolean
4244
}
4345

4446
interface QueuedSendHandoffClaim {
@@ -191,6 +193,9 @@ export function readQueuedSendHandoffState(): QueuedSendHandoffState | null {
191193
organizationId: parsed.organizationId,
192194
supersededStreamId,
193195
...(parsed.stopRequired === true ? { stopRequired: true } : {}),
196+
...(typeof parsed.admissionUnknown === 'boolean'
197+
? { admissionUnknown: parsed.admissionUnknown }
198+
: {}),
194199
userMessageId: parsed.userMessageId,
195200
message: parsed.message,
196201
...(Array.isArray(parsed.fileAttachments)

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

Lines changed: 126 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2336,11 +2336,11 @@ describe('useChat remount send recovery', () => {
23362336
* must not be editable into a second message.
23372337
*/
23382338
it.each([
2339-
{ stopRequired: false, editable: false },
2340-
{ stopRequired: true, editable: true },
2339+
{ stopRequired: false, admissionUnknown: true, editable: false },
2340+
{ stopRequired: true, admissionUnknown: false, editable: true },
23412341
])(
2342-
'guards a Send-now restored from its stored handoff (Stop still required: $stopRequired)',
2343-
async ({ stopRequired, editable }) => {
2342+
'guards a Send-now restored from its stored handoff (possibly sent: $admissionUnknown)',
2343+
async ({ stopRequired, admissionUnknown, editable }) => {
23442344
const history: MothershipChatHistory = {
23452345
id: 'chat-a',
23462346
mode: 'agent',
@@ -2364,6 +2364,7 @@ describe('useChat remount send recovery', () => {
23642364
userMessageId: 'send-now-request',
23652365
message: 'inspect the second invoice instead',
23662366
...(stopRequired ? { stopRequired: true } : {}),
2367+
admissionUnknown,
23672368
requestedAt: Date.now(),
23682369
})
23692370
const { getResult } = renderUseChatInChat('chat-a')
@@ -2393,6 +2394,127 @@ describe('useChat remount send recovery', () => {
23932394
}
23942395
)
23952396

2397+
/**
2398+
* The id a Send-now stores is not always fresh: a re-queued message reuses its
2399+
* earlier attempt's id, which may already be on the server. Reloaded while the
2400+
* Stop is still pending, the restored entry must stay uneditable.
2401+
*/
2402+
it('keeps a resumed Send-now uneditable when the page reloads before its Stop settles', async () => {
2403+
const first = renderUseChatInChat('chat-a')
2404+
await act(async () => {
2405+
void first.getResult().sendMessage('Original request')
2406+
})
2407+
await waitFor(() => state.postBodies.length === 1 && first.getResult().isSending)
2408+
vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
2409+
/** The Stop never settles before the reload. */
2410+
if (String(input).includes('/api/copilot/chat/abort')) return new Promise<Response>(() => {})
2411+
return fetchStub(input, init)
2412+
})
2413+
useMothershipQueueStore.getState().enqueue('chat-a', {
2414+
id: 'resumed',
2415+
content: 'sent earlier with no answer',
2416+
resumeUserMessageId: 'unanswered-attempt',
2417+
admissionUnknown: true,
2418+
})
2419+
await act(async () => {
2420+
void first.getResult().sendNow('resumed')
2421+
})
2422+
await waitFor(() => readQueuedSendHandoffState()?.userMessageId === 'unanswered-attempt')
2423+
first.unmount()
2424+
2425+
/** After the reload: the previous turn is over and the stored handoff comes back. */
2426+
const history: MothershipChatHistory = {
2427+
id: 'chat-a',
2428+
mode: 'agent',
2429+
title: 'Reloaded',
2430+
messages: [],
2431+
activeStreamId: null,
2432+
resources: [],
2433+
}
2434+
let loadHistory: (() => void) | undefined
2435+
mockRequestJson.mockImplementation(
2436+
() =>
2437+
new Promise((resolve) => {
2438+
loadHistory = () => resolve({ chat: history })
2439+
})
2440+
)
2441+
const reloaded = renderUseChatInChat('chat-a')
2442+
const editAtRestore: Array<ReturnType<ReturnType<typeof useChat>['editQueuedMessage']>> = []
2443+
let tried = false
2444+
const unsubscribe = useMothershipQueueStore.subscribe((queueState) => {
2445+
if (tried) return
2446+
const restored = queueState.queues['chat-a']?.find(
2447+
(message) => message.content === 'sent earlier with no answer'
2448+
)
2449+
if (!restored) return
2450+
tried = true
2451+
editAtRestore.push(reloaded.getResult().editQueuedMessage(restored.id))
2452+
})
2453+
try {
2454+
await waitFor(() => loadHistory !== undefined)
2455+
await act(async () => {
2456+
loadHistory?.()
2457+
await sleep(50)
2458+
})
2459+
await waitFor(() => tried)
2460+
} finally {
2461+
unsubscribe()
2462+
}
2463+
2464+
expect(editAtRestore[0]).toBeUndefined()
2465+
})
2466+
2467+
/**
2468+
* A send superseded at admission is answered like a duplicate, naming its own
2469+
* id with no stream yet: another attempt holding that id may still admit it.
2470+
* The message waits uneditable, then goes out again under the same id.
2471+
*/
2472+
it('keeps a send answered as a duplicate with no stream uneditable, then retries it', async () => {
2473+
const history: MothershipChatHistory = {
2474+
id: 'chat-superseded-send',
2475+
mode: 'agent',
2476+
title: 'Superseded',
2477+
messages: [],
2478+
activeStreamId: null,
2479+
resources: [],
2480+
}
2481+
mockRequestJson.mockImplementation(() => Promise.resolve({ chat: history }))
2482+
vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
2483+
const url = String(input)
2484+
if (url === '/api/mothership/chat' && init?.method === 'POST') {
2485+
state.postBodies.push(JSON.parse(String(init.body)))
2486+
if (state.postBodies.length === 1) {
2487+
return Response.json(
2488+
{
2489+
error: 'This message was already sent.',
2490+
activeStreamId: state.postBodies[0].userMessageId,
2491+
},
2492+
{ status: 409 }
2493+
)
2494+
}
2495+
return emptySseResponse()
2496+
}
2497+
if (url.includes('/api/mothership/chat/stream') && state.postBodies.length === 1) {
2498+
return Response.json({ error: 'Stream not found' }, { status: 404 })
2499+
}
2500+
return fetchStub(input, init)
2501+
})
2502+
const { getResult } = renderUseChatInChat(history.id, history)
2503+
await act(async () => {
2504+
await getResult().sendMessage('Superseded at admission')
2505+
})
2506+
const waiting = useMothershipQueueStore.getState().queues[history.id]?.[0]
2507+
let edited: ReturnType<ReturnType<typeof useChat>['editQueuedMessage']>
2508+
await act(async () => {
2509+
edited = getResult().editQueuedMessage(waiting?.id ?? '')
2510+
})
2511+
2512+
expect(waiting?.admissionUnknown).toBe(true)
2513+
expect(edited).toBeUndefined()
2514+
await waitFor(() => state.postBodies.length === 2, 5_000)
2515+
expect(state.postBodies[1].userMessageId).toBe(state.postBodies[0].userMessageId)
2516+
})
2517+
23962518
/**
23972519
* A held message the server then refuses as busy is known not to be a turn
23982520
* there: the server answers a retry of an admitted id as a duplicate, never

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

Lines changed: 30 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -237,19 +237,8 @@ interface WithdrawnSendResult {
237237
busy?: boolean
238238
/** Not sent at all (its Stop handoff failed); kept queued for the user to send. */
239239
held?: boolean
240-
/**
241-
* The server refused this id outright (busy, or a predecessor still shutting
242-
* down). Within the 1-hour claim TTL it answers a retry of an admitted id as a
243-
* duplicate instead, so the server is known not to have it, and its queue
244-
* entry can be edited.
245-
*/
246-
notAdmitted?: boolean
247-
/**
248-
* This attempt never reached the server (its Stop did not settle). That says
249-
* nothing about an earlier attempt the message resumes, whose uncertainty it
250-
* keeps.
251-
*/
252-
neverSent?: boolean
240+
/** Whether the server may hold `userMessageId`; see `admissionUnknown` in `startSendMessage`. */
241+
admissionUnknown: boolean
253242
}
254243

255244
/**
@@ -273,6 +262,8 @@ interface StartSendMessageOptions {
273262
* opening a second chat and billing a second turn.
274263
*/
275264
resumeUserMessageId?: string
265+
/** The queued entry's `admissionUnknown`, for the id it reuses. */
266+
admissionUnknown?: boolean
276267
requestMode?: ChatRequestMode
277268
assistantSearch?: WorkspaceSearchFilters
278269
assistantSearchLevel?: AssistantSearchLevel
@@ -3586,8 +3577,15 @@ export function useChat(
35863577

35873578
/* A retry of a withdrawn send reuses its id so the server deduplicates
35883579
the two attempts; anything else mints a fresh one. */
3589-
const userMessageId =
3590-
queuedSendHandoff?.userMessageId ?? options?.resumeUserMessageId ?? generateId()
3580+
const reusedId = queuedSendHandoff?.userMessageId ?? options?.resumeUserMessageId
3581+
const userMessageId = reusedId ?? generateId()
3582+
/* Whether the server may already hold `userMessageId`: the one fact that keeps a
3583+
queued message from being edited into a second turn. A reused id may have been
3584+
sent before, unless its entry knows the server refused it; a fresh id is unsent
3585+
until its POST goes out. Only the server refusing the id clears it again
3586+
(within the 1-hour claim TTL a retry of an admitted id is answered as a
3587+
duplicate, never refused). It rides the stored handoff and every withdrawal. */
3588+
let admissionUnknown = reusedId !== undefined && options?.admissionUnknown !== false
35913589
const assistantId = getLiveAssistantMessageId(userMessageId)
35923590

35933591
const storedAttachments: PersistedFileAttachment[] | undefined =
@@ -3637,6 +3635,8 @@ export function useChat(
36373635
organizationId,
36383636
supersededStreamId: queuedSendHandoff.supersededStreamId,
36393637
...(queuedSendHandoff.stopRequired ? { stopRequired: true } : {}),
3638+
/** Without a pending Stop, its POST goes out next. */
3639+
admissionUnknown: admissionUnknown || !queuedSendHandoff.stopRequired,
36403640
userMessageId,
36413641
message,
36423642
...(fileAttachments ? { fileAttachments } : {}),
@@ -3911,7 +3911,7 @@ export function useChat(
39113911
setError(getErrorMessage(err, 'Failed to stop the previous response'))
39123912
/* Nothing was sent. Hand the message back so it stays in its chat's queue
39133913
even if the user has switched chats since the Stop began. */
3914-
return { userMessageId, held: true, neverSent: true }
3914+
return { userMessageId, held: true, admissionUnknown }
39153915
}
39163916
}
39173917

@@ -3935,6 +3935,7 @@ export function useChat(
39353935
? {}
39363936
: await getDesktopChatCapabilities(desktopScopeIdRef.current)
39373937

3938+
admissionUnknown = true
39383939
const response = await fetch(apiPathRef.current, {
39393940
method: 'POST',
39403941
headers: { 'Content-Type': 'application/json' },
@@ -4031,7 +4032,7 @@ export function useChat(
40314032
}
40324033
if (viewOnSend)
40334034
setError('Previous response is still shutting down; queued message was restored.')
4034-
return { userMessageId, held: true, notAdmitted: true }
4035+
return { userMessageId, held: true, admissionUnknown: false }
40354036
}
40364037
/** Withdraws this refused send so the queue retries it, under the same id, later. */
40374038
const releaseRefusedSend = () => {
@@ -4070,7 +4071,7 @@ export function useChat(
40704071
/* Only the chat lock refuses without naming this id. Admission's
40714072
"superseded" conflict, where another attempt took this id's claim and may
40724073
admit it, is answered as a duplicate naming this id instead. */
4073-
return { userMessageId, busy: true, notAdmitted: true }
4074+
return { userMessageId, busy: true, admissionUnknown: false }
40744075
}
40754076
/* "Already sent" with no stream for it means the earlier attempt is still
40764077
in flight on the server (or died before starting a turn), not that a turn
@@ -4092,7 +4093,7 @@ export function useChat(
40924093
)
40934094
if (!dedupedStreamExists) {
40944095
releaseRefusedSend()
4095-
return { userMessageId, busy: true }
4096+
return { userMessageId, busy: true, admissionUnknown }
40964097
}
40974098
/** The user may have moved on (another chat, another send) during the check. */
40984099
if (streamGenRef.current !== gen) return consumedByTranscript
@@ -4203,7 +4204,7 @@ export function useChat(
42034204
server deduplicates it against that turn instead of billing
42044205
another one. */
42054206
rollbackOptimisticSend()
4206-
return { userMessageId }
4207+
return { userMessageId, admissionUnknown }
42074208
}
42084209
return consumedByTranscript
42094210
}
@@ -4245,6 +4246,7 @@ export function useChat(
42454246
)
42464247
return {
42474248
userMessageId,
4249+
admissionUnknown,
42484250
unreachable: true,
42494251
...(retryLater ? {} : { heldUntilOnline: true }),
42504252
}
@@ -4449,9 +4451,7 @@ export function useChat(
44494451
...(result.heldUntilOnline ? { retryRequired: true, heldUntilOnline: true } : {}),
44504452
...(result.held ? { retryRequired: true } : {}),
44514453
...((result.unreachable && !result.heldUntilOnline) || result.busy ? sendRetry(1) : {}),
4452-
/* Only a refusal of this id settles it. A direct send never waits on a Stop
4453-
(one pending queues it instead), so `neverSent` cannot occur here. */
4454-
admissionUnknown: !result.notAdmitted,
4454+
admissionUnknown: result.admissionUnknown,
44554455
...((result.unreachable || result.busy) && activeChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)
44564456
? { heldSurface: heldSendSurface }
44574457
: {}),
@@ -4667,6 +4667,9 @@ export function useChat(
46674667
userMessageId: handoff.userMessageId,
46684668
...(handoff.stopRequired ? { stopRequired: true } : {}),
46694669
},
4670+
...(handoff.admissionUnknown !== undefined
4671+
? { admissionUnknown: handoff.admissionUnknown }
4672+
: {}),
46704673
})
46714674
clearQueuedSendHandoffState(handoff.id)
46724675
clearQueuedSendHandoffClaim(handoff.id)
@@ -5083,17 +5086,7 @@ export function useChat(
50835086
? { heldSurface: heldSendSurface }
50845087
: {}),
50855088
...(withdrawnUserMessageId ? { resumeUserMessageId: withdrawnUserMessageId } : {}),
5086-
/* A refusal of this id settles it; an attempt that never left keeps the
5087-
earlier uncertainty; any other withdrawal may have reached the server. */
5088-
...(withdrawn
5089-
? {
5090-
admissionUnknown: withdrawn.notAdmitted
5091-
? false
5092-
: withdrawn.neverSent
5093-
? dispatched.admissionUnknown === true
5094-
: true,
5095-
}
5096-
: {}),
5089+
...(withdrawn ? { admissionUnknown: withdrawn.admissionUnknown } : {}),
50975090
})
50985091
}
50995092

@@ -5125,6 +5118,9 @@ export function useChat(
51255118
...(liveMsg.resumeUserMessageId
51265119
? { resumeUserMessageId: liveMsg.resumeUserMessageId }
51275120
: {}),
5121+
...(liveMsg.admissionUnknown !== undefined
5122+
? { admissionUnknown: liveMsg.admissionUnknown }
5123+
: {}),
51285124
...(liveMsg.requestMode ? { requestMode: liveMsg.requestMode } : {}),
51295125
...(liveMsg.assistantSearch ? { assistantSearch: liveMsg.assistantSearch } : {}),
51305126
...(liveMsg.assistantSearchLevel !== undefined

‎apps/sim/lib/mothership/chat/application/admit-turn.test.ts‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import {
2323
} from '@sim/testing/mocks/permission-groups-resolve.mock'
2424
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest'
2525
import { admitChatTurn, turnDesktopDevice } from '@/lib/mothership/chat/application/admit-turn'
26+
import { ChatSendSupersededError } from '@/lib/mothership/chat/application/send-superseded'
2627

2728
const hoisted = vi.hoisted(() => ({
2829
lease: vi.fn(),
@@ -123,6 +124,22 @@ describe('organization turn admission through current private-chat authorization
123124
expect.anything()
124125
)
125126
})
127+
/**
128+
* Another attempt with this id re-took the claim after this one's in-progress
129+
* TTL ran out. That attempt may admit the turn, so this one must not, and the
130+
* route answers it as a duplicate.
131+
*/
132+
it('refuses to admit a send whose claim another attempt took', async () => {
133+
queueTableRows(copilotChats, [chat])
134+
queueTableRows(member, [{ role: 'member' }])
135+
dbChainMockFns.returning
136+
.mockResolvedValueOnce([{ model: null }])
137+
.mockResolvedValueOnce([{ id: 'run-1', organizationId: 'org-1', workspaceId: null }])
138+
.mockResolvedValueOnce([])
139+
await expect(admitChatTurn.execute({ principal, input: input() })).rejects.toBeInstanceOf(
140+
ChatSendSupersededError
141+
)
142+
})
126143
it("keeps an organization chat's turn with its chat view, never on a desktop", async () => {
127144
hoisted.resolveDesktop.mockResolvedValue('device-1')
128145
const offered = '33333333-3333-4333-8333-333333333333'

‎apps/sim/stores/mothership-queue/store.test.ts‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ describe('useMothershipQueueStore', () => {
3030
})
3131

3232
describe('replaceAt', () => {
33-
it('treats a Send-now whose Stop settled as possibly sent under its handoff id', () => {
33+
it('reads a reused handoff id as possibly sent unless the entry says otherwise', () => {
3434
useMothershipQueueStore.getState().enqueue('chat-A', {
3535
id: 'sent',
3636
content: 'original',
@@ -51,6 +51,8 @@ describe('useMothershipQueueStore', () => {
5151
userMessageId: 'not-sent-yet',
5252
stopRequired: true,
5353
},
54+
/** A fresh id still waiting on its Stop, as the hook records it. */
55+
admissionUnknown: false,
5456
})
5557
useMothershipQueueStore.getState().replaceAt('chat-A', 'sent', { content: 'edited' })
5658
useMothershipQueueStore.getState().replaceAt('chat-A', 'waiting', { content: 'edited' })

0 commit comments

Comments
 (0)