Skip to content

Commit c677e91

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 0668369 commit c677e91

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
@@ -234,19 +234,8 @@ interface WithdrawnSendResult {
234234
busy?: boolean
235235
/** Not sent at all (its Stop handoff failed); kept queued for the user to send. */
236236
held?: boolean
237-
/**
238-
* The server refused this id outright (busy, or a predecessor still shutting
239-
* down). Within the 1-hour claim TTL it answers a retry of an admitted id as a
240-
* duplicate instead, so the server is known not to have it, and its queue
241-
* entry can be edited.
242-
*/
243-
notAdmitted?: boolean
244-
/**
245-
* This attempt never reached the server (its Stop did not settle). That says
246-
* nothing about an earlier attempt the message resumes, whose uncertainty it
247-
* keeps.
248-
*/
249-
neverSent?: boolean
237+
/** Whether the server may hold `userMessageId`; see `admissionUnknown` in `startSendMessage`. */
238+
admissionUnknown: boolean
250239
}
251240

252241
/**
@@ -270,6 +259,8 @@ interface StartSendMessageOptions {
270259
* opening a second chat and billing a second turn.
271260
*/
272261
resumeUserMessageId?: string
262+
/** The queued entry's `admissionUnknown`, for the id it reuses. */
263+
admissionUnknown?: boolean
273264
requestMode?: ChatRequestMode
274265
assistantSearch?: WorkspaceSearchFilters
275266
assistantSearchLevel?: AssistantSearchLevel
@@ -3575,8 +3566,15 @@ export function useChat(
35753566

35763567
/* A retry of a withdrawn send reuses its id so the server deduplicates
35773568
the two attempts; anything else mints a fresh one. */
3578-
const userMessageId =
3579-
queuedSendHandoff?.userMessageId ?? options?.resumeUserMessageId ?? generateId()
3569+
const reusedId = queuedSendHandoff?.userMessageId ?? options?.resumeUserMessageId
3570+
const userMessageId = reusedId ?? generateId()
3571+
/* Whether the server may already hold `userMessageId`: the one fact that keeps a
3572+
queued message from being edited into a second turn. A reused id may have been
3573+
sent before, unless its entry knows the server refused it; a fresh id is unsent
3574+
until its POST goes out. Only the server refusing the id clears it again
3575+
(within the 1-hour claim TTL a retry of an admitted id is answered as a
3576+
duplicate, never refused). It rides the stored handoff and every withdrawal. */
3577+
let admissionUnknown = reusedId !== undefined && options?.admissionUnknown !== false
35803578
const assistantId = getLiveAssistantMessageId(userMessageId)
35813579

35823580
const storedAttachments: PersistedFileAttachment[] | undefined =
@@ -3626,6 +3624,8 @@ export function useChat(
36263624
organizationId,
36273625
supersededStreamId: queuedSendHandoff.supersededStreamId,
36283626
...(queuedSendHandoff.stopRequired ? { stopRequired: true } : {}),
3627+
/** Without a pending Stop, its POST goes out next. */
3628+
admissionUnknown: admissionUnknown || !queuedSendHandoff.stopRequired,
36293629
userMessageId,
36303630
message,
36313631
...(fileAttachments ? { fileAttachments } : {}),
@@ -3900,7 +3900,7 @@ export function useChat(
39003900
setError(getErrorMessage(err, 'Failed to stop the previous response'))
39013901
/* Nothing was sent. Hand the message back so it stays in its chat's queue
39023902
even if the user has switched chats since the Stop began. */
3903-
return { userMessageId, held: true, neverSent: true }
3903+
return { userMessageId, held: true, admissionUnknown }
39043904
}
39053905
}
39063906

@@ -3924,6 +3924,7 @@ export function useChat(
39243924
? {}
39253925
: await getDesktopChatCapabilities(desktopScopeIdRef.current)
39263926

3927+
admissionUnknown = true
39273928
const response = await fetch(apiPathRef.current, {
39283929
method: 'POST',
39293930
headers: { 'Content-Type': 'application/json' },
@@ -4017,7 +4018,7 @@ export function useChat(
40174018
}
40184019
if (viewOnSend)
40194020
setError('Previous response is still shutting down; queued message was restored.')
4020-
return { userMessageId, held: true, notAdmitted: true }
4021+
return { userMessageId, held: true, admissionUnknown: false }
40214022
}
40224023
/** Withdraws this refused send so the queue retries it, under the same id, later. */
40234024
const releaseRefusedSend = () => {
@@ -4056,7 +4057,7 @@ export function useChat(
40564057
/* Only the chat lock refuses without naming this id. Admission's
40574058
"superseded" conflict, where another attempt took this id's claim and may
40584059
admit it, is answered as a duplicate naming this id instead. */
4059-
return { userMessageId, busy: true, notAdmitted: true }
4060+
return { userMessageId, busy: true, admissionUnknown: false }
40604061
}
40614062
/* "Already sent" with no stream for it means the earlier attempt is still
40624063
in flight on the server (or died before starting a turn), not that a turn
@@ -4078,7 +4079,7 @@ export function useChat(
40784079
)
40794080
if (!dedupedStreamExists) {
40804081
releaseRefusedSend()
4081-
return { userMessageId, busy: true }
4082+
return { userMessageId, busy: true, admissionUnknown }
40824083
}
40834084
/** The user may have moved on (another chat, another send) during the check. */
40844085
if (streamGenRef.current !== gen) return consumedByTranscript
@@ -4189,7 +4190,7 @@ export function useChat(
41894190
server deduplicates it against that turn instead of billing
41904191
another one. */
41914192
rollbackOptimisticSend()
4192-
return { userMessageId }
4193+
return { userMessageId, admissionUnknown }
41934194
}
41944195
return consumedByTranscript
41954196
}
@@ -4224,6 +4225,7 @@ export function useChat(
42244225
)
42254226
return {
42264227
userMessageId,
4228+
admissionUnknown,
42274229
unreachable: true,
42284230
...(onlineEventsRef.current !== onlineEventsAtSend ? { networkReturned: true } : {}),
42294231
}
@@ -4430,9 +4432,7 @@ export function useChat(
44304432
: {}),
44314433
...(result.held ? { retryRequired: true } : {}),
44324434
...(result.busy ? busyRetry(1) : {}),
4433-
/* Only a refusal of this id settles it. A direct send never waits on a Stop
4434-
(one pending queues it instead), so `neverSent` cannot occur here. */
4435-
admissionUnknown: !result.notAdmitted,
4435+
admissionUnknown: result.admissionUnknown,
44364436
...((result.unreachable || result.busy) && activeChatKey.startsWith(PENDING_CHAT_KEY_PREFIX)
44374437
? { heldSurface: heldSendSurface }
44384438
: {}),
@@ -4648,6 +4648,9 @@ export function useChat(
46484648
userMessageId: handoff.userMessageId,
46494649
...(handoff.stopRequired ? { stopRequired: true } : {}),
46504650
},
4651+
...(handoff.admissionUnknown !== undefined
4652+
? { admissionUnknown: handoff.admissionUnknown }
4653+
: {}),
46514654
})
46524655
clearQueuedSendHandoffState(handoff.id)
46534656
clearQueuedSendHandoffClaim(handoff.id)
@@ -5062,17 +5065,7 @@ export function useChat(
50625065
? { heldSurface: heldSendSurface }
50635066
: {}),
50645067
...(withdrawnUserMessageId ? { resumeUserMessageId: withdrawnUserMessageId } : {}),
5065-
/* A refusal of this id settles it; an attempt that never left keeps the
5066-
earlier uncertainty; any other withdrawal may have reached the server. */
5067-
...(withdrawn
5068-
? {
5069-
admissionUnknown: withdrawn.notAdmitted
5070-
? false
5071-
: withdrawn.neverSent
5072-
? dispatched.admissionUnknown === true
5073-
: true,
5074-
}
5075-
: {}),
5068+
...(withdrawn ? { admissionUnknown: withdrawn.admissionUnknown } : {}),
50765069
})
50775070
}
50785071

@@ -5104,6 +5097,9 @@ export function useChat(
51045097
...(liveMsg.resumeUserMessageId
51055098
? { resumeUserMessageId: liveMsg.resumeUserMessageId }
51065099
: {}),
5100+
...(liveMsg.admissionUnknown !== undefined
5101+
? { admissionUnknown: liveMsg.admissionUnknown }
5102+
: {}),
51075103
...(liveMsg.requestMode ? { requestMode: liveMsg.requestMode } : {}),
51085104
...(liveMsg.assistantSearch ? { assistantSearch: liveMsg.assistantSearch } : {}),
51095105
...(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)