Skip to content

Commit 49b06c7

Browse files
committed
fix(chat): carry every message of a new chat past its surface's unmount
A second message sent during a new chat's admission went back to pending in the dead mount: only the withdrawn send was handed off, and a provisional key is not journaled. A message waiting on a retry delay was lost the same way. Everything under the provisional key now goes on, in order. A stored handoff holds one message, so handing each off in turn would overwrite the first: the first goes through the handoff under its id, and the rest wait in the journal for the surface that resumes it. One shown as not delivered stays that way.
1 parent aa51b01 commit 49b06c7

4 files changed

Lines changed: 227 additions & 6 deletions

File tree

‎apps/sim/app/workspace/[workspaceId]/home/hooks/outgoing-state.ts‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -189,6 +189,29 @@ export function outgoingRetryDelay(attempts: number): number {
189189
})
190190
}
191191

192+
/**
193+
* Where the rest of a new chat's messages wait once its surface is gone: the handoff
194+
* carries one message, and whichever surface resumes that one takes these behind it.
195+
*/
196+
export function handedOffJournalKey(resumedId: string): string {
197+
return `handoff:${resumedId}`
198+
}
199+
200+
/**
201+
* What a new chat's message goes on to the next surface as. Nothing took it, since
202+
* its chat never existed, so it starts over; one shown as not delivered stays that
203+
* way until the user asks.
204+
*/
205+
export function toHandedOff(message: OutgoingMessage): JournaledOutgoing {
206+
return {
207+
id: message.id,
208+
payload: message.payload,
209+
status: message.status === 'failed' ? 'failed' : 'pending',
210+
...(message.unconfirmed ? { unconfirmed: true } : {}),
211+
writtenAt: message.writtenAt,
212+
}
213+
}
214+
192215
type PersistedRow = MothershipChatHistory['messages'][number]
193216

194217
/**

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

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2178,6 +2178,151 @@ describe('useChat remount send recovery', () => {
21782178
expect(MothershipHandoffStorage.consume('ws-1')?.message).toBe('must survive my own teardown')
21792179
})
21802180

2181+
describe('a new chat’s messages when its surface unmounts', () => {
2182+
const idOf = (view: { getResult: () => ReturnType<typeof useChat> }, content: string) =>
2183+
view.getResult().messages.find((message) => message.content === content)?.id
2184+
2185+
/** From here on a send is admitted and opens its chat; its turn ends at once. */
2186+
function admitSends() {
2187+
mockRequestJson.mockImplementation(() =>
2188+
Promise.resolve({
2189+
chat: {
2190+
id: DEDUPED_CHAT_ID,
2191+
mode: 'agent',
2192+
title: 'Handed off',
2193+
messages: [],
2194+
activeStreamId: null,
2195+
resources: [],
2196+
} satisfies MothershipChatHistory,
2197+
})
2198+
)
2199+
vi.stubGlobal('fetch', (input: RequestInfo | URL, init?: RequestInit) => {
2200+
if (String(input) !== '/api/mothership/chat' || init?.method !== 'POST') {
2201+
return fetchStub(input, init)
2202+
}
2203+
state.postBodies.push(JSON.parse(String(init.body)))
2204+
return Promise.resolve(
2205+
new Response(emptySseResponse().body, {
2206+
headers: {
2207+
'Content-Type': 'text/event-stream',
2208+
'x-mothership-chat-id': DEDUPED_CHAT_ID,
2209+
},
2210+
})
2211+
)
2212+
})
2213+
}
2214+
2215+
async function sendTwoDuringAdmission() {
2216+
const view = renderUseChat()
2217+
await act(async () => {
2218+
void view.getResult().sendMessage('First')
2219+
})
2220+
await waitFor(() => state.postBodies.length === 1)
2221+
await act(async () => {
2222+
void view.getResult().sendMessage('Second')
2223+
await settle()
2224+
})
2225+
return { view, secondId: idOf(view, 'Second') }
2226+
}
2227+
2228+
/** Sent while the first waited on its admission, the second has no chat to wait in. */
2229+
it('hands off a message sent during the admission, behind the first, from storage', async () => {
2230+
const { view, secondId } = await sendTwoDuringAdmission()
2231+
expect(secondId).toBeDefined()
2232+
view.unmount()
2233+
await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null)
2234+
2235+
const handoff = MothershipHandoffStorage.consume('ws-1')
2236+
expect(handoff?.message).toBe('First')
2237+
admitSends()
2238+
const next = renderUseChat()
2239+
await act(async () => {
2240+
await next.getResult().sendMessage(handoff?.message ?? '', undefined, undefined, {
2241+
resumeUserMessageId: handoff?.resumeUserMessageId,
2242+
})
2243+
})
2244+
await waitFor(() => state.postBodies.length === 3)
2245+
2246+
expect(state.postBodies.slice(1).map((body) => [body.message, body.userMessageId])).toEqual([
2247+
['First', handoff?.resumeUserMessageId],
2248+
['Second', secondId],
2249+
])
2250+
})
2251+
2252+
it('hands off a message sent during the admission to a live replacement', async () => {
2253+
const replacement = renderHomeLikeSurface()
2254+
const { view, secondId } = await sendTwoDuringAdmission()
2255+
admitSends()
2256+
view.unmount()
2257+
await waitFor(() => state.postBodies.length === 3)
2258+
2259+
expect(replacement.claimedByOwnListener()).toBe(1)
2260+
expect(state.postBodies[1].message).toBe('First')
2261+
expect(state.postBodies[2]).toMatchObject({ message: 'Second', userMessageId: secondId })
2262+
expect(window.localStorage.getItem('sim_mothership_handoff')).toBeNull()
2263+
})
2264+
2265+
/** Its POST came back with no answer, so no send is out to come back withdrawn. */
2266+
it('hands off what waits on a retry delay when no send is out', async () => {
2267+
vi.spyOn(window.navigator, 'onLine', 'get').mockReturnValue(true)
2268+
vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
2269+
if (String(input) === '/api/mothership/chat' && init?.method === 'POST') {
2270+
state.postBodies.push(JSON.parse(String(init.body)))
2271+
throw new TypeError('Failed to fetch')
2272+
}
2273+
return fetchStub(input, init)
2274+
})
2275+
const view = renderUseChat()
2276+
await act(async () => {
2277+
void view.getResult().sendMessage('First')
2278+
await settle()
2279+
})
2280+
await act(async () => {
2281+
void view.getResult().sendMessage('Second')
2282+
await settle()
2283+
})
2284+
const firstId = idOf(view, 'First')
2285+
const secondId = idOf(view, 'Second')
2286+
const postsBefore = state.postBodies.length
2287+
2288+
view.unmount()
2289+
const handoff = MothershipHandoffStorage.consume('ws-1')
2290+
expect(handoff?.resumeUserMessageId).toBe(firstId)
2291+
2292+
admitSends()
2293+
const next = renderUseChat()
2294+
await act(async () => {
2295+
await next.getResult().sendMessage(handoff?.message ?? '', undefined, undefined, {
2296+
resumeUserMessageId: handoff?.resumeUserMessageId,
2297+
})
2298+
})
2299+
await waitFor(() => state.postBodies.length >= postsBefore + 2)
2300+
2301+
expect(
2302+
state.postBodies.slice(postsBefore).map((body) => [body.message, body.userMessageId])
2303+
).toEqual([
2304+
['First', firstId],
2305+
['Second', secondId],
2306+
])
2307+
})
2308+
2309+
/** What a new send of its own carries is no resume, so it takes nothing behind it. */
2310+
it('leaves handed-off messages for the surface that resumes the first', async () => {
2311+
const { view } = await sendTwoDuringAdmission()
2312+
view.unmount()
2313+
await waitFor(() => window.localStorage.getItem('sim_mothership_handoff') !== null)
2314+
2315+
admitSends()
2316+
const next = renderUseChat()
2317+
await act(async () => {
2318+
await next.getResult().sendMessage('Unrelated')
2319+
})
2320+
await settle()
2321+
2322+
expect(state.postBodies.slice(1).map((body) => body.message)).toEqual(['Unrelated'])
2323+
})
2324+
})
2325+
21812326
describe('retrying a withdrawn send', () => {
21822327
/**
21832328
* The whole point of carrying the id: the server sees one logical send, so

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

Lines changed: 55 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,7 @@ import {
113113
canSteer,
114114
deliveryOf,
115115
fromJournaled,
116+
handedOffJournalKey,
116117
hasOutgoingFollowUp,
117118
holdAfterErroredTurn,
118119
isDeliveredAsTurn,
@@ -125,6 +126,7 @@ import {
125126
orphanSteeringRows,
126127
outgoingRetryDelay,
127128
type SteeringOutcome,
129+
toHandedOff,
128130
toJournaled,
129131
toMessageContexts,
130132
} from '@/app/workspace/[workspaceId]/home/hooks/outgoing-state'
@@ -4233,6 +4235,31 @@ export function useChat(
42334235
[workspaceId, organizationId]
42344236
)
42354237

4238+
/**
4239+
* A provisional key dies with this mount, so what it holds goes on to the next
4240+
* surface, in order: the first through the handoff, under its id, and the rest in
4241+
* the journal for the surface that resumes it, since a stored handoff holds one.
4242+
*/
4243+
const handOffProvisionalOutgoing = useCallback(
4244+
(chatKey: string) => {
4245+
const undelivered = outgoingRef.current.filter(
4246+
(message) =>
4247+
message.chatKey === chatKey &&
4248+
message.status !== 'steered' &&
4249+
message.status !== 'applied'
4250+
)
4251+
writeOutgoing((current) => current.filter((message) => message.chatKey !== chatKey))
4252+
const resumed = undelivered.find((message) => message.status !== 'failed')
4253+
if (!resumed) return
4254+
const behind = undelivered.filter((message) => message !== resumed).map(toHandedOff)
4255+
if (behind.length > 0) {
4256+
useMothershipOutgoingStore.getState().record(handedOffJournalKey(resumed.id), behind)
4257+
}
4258+
handOffWithdrawnSend({ ...resumed.payload, userMessageId: resumed.id })
4259+
},
4260+
[handOffWithdrawnSend, writeOutgoing]
4261+
)
4262+
42364263
/**
42374264
* The turn a message sent now would steer: the running one, or the one whose
42384265
* admission is still out in this chat (a message waits for that answer rather
@@ -4556,10 +4583,10 @@ export function useChat(
45564583
}
45574584
if (result.reason === 'withdrawn') {
45584585
/* An unmount withdrew it before the server answered. A chat's journal keeps
4559-
it for the next mount of that chat; a new chat's goes to the next surface. */
4586+
it for the next mount of that chat; a new chat's goes to the next surface,
4587+
with whatever was waiting on its admission. */
45604588
if (message.chatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) {
4561-
settled()
4562-
handOffWithdrawnSend({ ...message.payload, userMessageId: result.userMessageId })
4589+
handOffProvisionalOutgoing(message.chatKey)
45634590
} else {
45644591
updateOutgoing(message.id, { status: 'pending', unconfirmed: true })
45654592
}
@@ -4610,7 +4637,7 @@ export function useChat(
46104637
}
46114638
}, [
46124639
deferOutgoing,
4613-
handOffWithdrawnSend,
4640+
handOffProvisionalOutgoing,
46144641
liveStreamId,
46154642
outgoingTurnSends,
46164643
queryClient,
@@ -4788,6 +4815,12 @@ export function useChat(
47884815
})
47894816
const id = options?.resumeUserMessageId ?? generateId()
47904817
const chatKey = chatKeyRef.current
4818+
const journal = useMothershipOutgoingStore.getState()
4819+
const handedOff = options?.resumeUserMessageId
4820+
? (journal.chats[handedOffJournalKey(id)] ?? [])
4821+
: []
4822+
if (handedOff.length > 0) journal.record(handedOffJournalKey(id), [])
4823+
const now = Date.now()
47914824
/* A new message is the user asking again, so anything held back in this chat
47924825
(an unreachable Sim, a busy chat) tries again with it, ahead of it. */
47934826
writeOutgoing((current) => [
@@ -4805,8 +4838,9 @@ export function useChat(
48054838
status: 'pending',
48064839
attempts: 0,
48074840
reads: 0,
4808-
writtenAt: Date.now(),
4841+
writtenAt: now,
48094842
},
4843+
...handedOff.map((entry) => fromJournaled(chatKey, entry, now)),
48104844
])
48114845
void deliverOutgoing()
48124846
/* A message that goes straight out as its own turn resolves with that turn,
@@ -4860,6 +4894,22 @@ export function useChat(
48604894
deliverOutgoing,
48614895
writeOutgoing,
48624896
])
4897+
/* Nothing is out to come back withdrawn (a retry waiting on its delay), so nothing
4898+
else would carry what a new chat holds past this unmount. */
4899+
useEffect(
4900+
() => () => {
4901+
if (surfaceMountedRef.current) return
4902+
const chatKey = chatKeyRef.current
4903+
if (!chatKey.startsWith(PENDING_CHAT_KEY_PREFIX)) return
4904+
const sendOut = outgoingRef.current.some(
4905+
(message) =>
4906+
message.chatKey === chatKey &&
4907+
(message.status === 'sending' || message.status === 'steering')
4908+
)
4909+
if (!sendOut) handOffProvisionalOutgoing(chatKey)
4910+
},
4911+
[handOffProvisionalOutgoing]
4912+
)
48634913
useEffect(() => {
48644914
if (typeof window === 'undefined') return
48654915
const retryOnline = () => {

‎apps/sim/stores/mothership-outgoing/store.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,10 @@ export interface JournaledOutgoing {
2323
}
2424

2525
interface MothershipOutgoingState {
26-
/** Per chat id, in the order the messages were written. */
26+
/**
27+
* Per chat id, in the order the messages were written, and per handed-off message
28+
* for what the surface that resumes it takes behind it (`handedOffJournalKey`).
29+
*/
2730
chats: Record<string, JournaledOutgoing[]>
2831
/** Replaces what a chat has to deliver; an empty list drops the chat. */
2932
record: (chatId: string, messages: JournaledOutgoing[]) => void

0 commit comments

Comments
 (0)