Skip to content

Commit ba35219

Browse files
j15zj15z
committed
fix: isolate copilot resource persistence by chat
1 parent 905fc86 commit ba35219

3 files changed

Lines changed: 230 additions & 111 deletions

File tree

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

Lines changed: 81 additions & 90 deletions
Original file line numberDiff line numberDiff line change
@@ -1404,9 +1404,8 @@ export function useChat(
14041404
})
14051405
}
14061406
const resourcePersistenceQueue = resourcePersistenceQueueRef.current
1407-
const pendingPersistResourceKeysRef = useRef(resourcePersistenceQueue.pendingKeys)
1408-
const inFlightResourceAddsRef = useRef(resourcePersistenceQueue.inFlight)
1409-
const reorderNeededAfterFlushRef = useRef(false)
1407+
const pendingResourceReordersRef = useRef(new Map<string, MothershipResource[]>())
1408+
const pendingResourceReorderFlushesRef = useRef(new Map<string, Promise<void>>())
14101409

14111410
// Derive the effective active resource ID for rendering without writing a
14121411
// passive fallback back into the user's URL selection.
@@ -1663,8 +1662,6 @@ export function useChat(
16631662
// Pending view pins belong to the chat whose stream issued them.
16641663
useTableViewPinStore.getState().reset()
16651664
undisplayableResourcesRef.current = []
1666-
resourcePersistenceQueue.clear()
1667-
reorderNeededAfterFlushRef.current = false
16681665
resetEphemeralPreviewState()
16691666
// Editing binds to this hook's composer — release it before rotating chatKey.
16701667
useMothershipQueueStore.getState().setEditing(chatKeyRef.current, null)
@@ -1691,30 +1688,59 @@ export function useChat(
16911688
workspaceId,
16921689
])
16931690

1694-
const flushPendingResources = useCallback(
1695-
async (chatId: string) => {
1696-
if (pendingPersistResourceKeysRef.current.size === 0) return
1697-
await resourcePersistenceQueue.flush(chatId)
1698-
if (!reorderNeededAfterFlushRef.current) return
1699-
reorderNeededAfterFlushRef.current = false
1700-
const localOrder = [
1701-
...resourcesRef.current.filter(
1702-
(r) =>
1703-
r.id !== 'streaming-file' &&
1704-
!pendingPersistResourceKeysRef.current.has(`${r.type}:${r.id}`)
1705-
),
1706-
...undisplayableResourcesRef.current,
1707-
]
1708-
if (localOrder.length === 0) return
1709-
requestJson(reorderMothershipChatResourcesContract, {
1710-
body: { chatId, resources: localOrder },
1711-
}).catch((err) => {
1712-
logger.warn('Failed to sync resource order after flush', err)
1691+
const flushPendingResourceReorder = useCallback(
1692+
(chatId: string): Promise<void> => {
1693+
const activeFlush = pendingResourceReorderFlushesRef.current.get(chatId)
1694+
if (activeFlush) return activeFlush
1695+
1696+
const flush = async () => {
1697+
while (true) {
1698+
const pendingOrder = pendingResourceReordersRef.current.get(chatId)
1699+
if (!pendingOrder) return
1700+
if (resourcePersistenceQueue.getPendingResourceKeys(chatId).size > 0) return
1701+
1702+
const inFlightWrites = resourcePersistenceQueue.getInFlightWrites(chatId)
1703+
if (inFlightWrites.length > 0) {
1704+
await Promise.allSettled(inFlightWrites)
1705+
continue
1706+
}
1707+
1708+
pendingResourceReordersRef.current.delete(chatId)
1709+
if (pendingOrder.length === 0) return
1710+
try {
1711+
await requestJson(reorderMothershipChatResourcesContract, {
1712+
body: { chatId, resources: pendingOrder },
1713+
})
1714+
} catch (error) {
1715+
if (!pendingResourceReordersRef.current.has(chatId)) {
1716+
pendingResourceReordersRef.current.set(chatId, pendingOrder)
1717+
}
1718+
logger.warn('Failed to persist resource reorder; will retry on next hydration', error)
1719+
return
1720+
}
1721+
}
1722+
}
1723+
const tracked = flush().finally(() => {
1724+
if (pendingResourceReorderFlushesRef.current.get(chatId) === tracked) {
1725+
pendingResourceReorderFlushesRef.current.delete(chatId)
1726+
}
17131727
})
1728+
pendingResourceReorderFlushesRef.current.set(chatId, tracked)
1729+
return tracked
17141730
},
17151731
[resourcePersistenceQueue]
17161732
)
17171733

1734+
const flushPendingResources = useCallback(
1735+
async (chatId: string, sourceScopeId: string = chatId) => {
1736+
if (resourcePersistenceQueue.getPendingResourceKeys(sourceScopeId).size > 0) {
1737+
await resourcePersistenceQueue.flush(chatId, sourceScopeId)
1738+
}
1739+
await flushPendingResourceReorder(chatId)
1740+
},
1741+
[flushPendingResourceReorder, resourcePersistenceQueue]
1742+
)
1743+
17181744
const adoptResolvedChatId = useCallback(
17191745
(chatId: string, options?: { replaceHomeHistory?: boolean; invalidateList?: boolean }) => {
17201746
const selectedChatId = selectedChatIdRef.current
@@ -1790,7 +1816,7 @@ export function useChat(
17901816
if (options?.invalidateList) {
17911817
queryClient.invalidateQueries({ queryKey: mothershipChatKeys.list(workspaceId) })
17921818
}
1793-
flushPendingResources(chatId)
1819+
flushPendingResources(chatId, pendingChatKey)
17941820
},
17951821
[flushPendingResources, queryClient, workspaceId]
17961822
)
@@ -1834,7 +1860,8 @@ export function useChat(
18341860
}
18351861

18361862
const persistChatId = chatIdRef.current ?? selectedChatIdRef.current
1837-
resourcePersistenceQueue.enqueue(resourceUpdate, persistChatId, existing)
1863+
const persistenceScopeId = persistChatId ?? pendingChatKeyRef.current
1864+
resourcePersistenceQueue.enqueue(resourceUpdate, persistChatId, existing, persistenceScopeId)
18381865
return existing === undefined
18391866
},
18401867
[resourcePersistenceQueue]
@@ -1848,15 +1875,24 @@ export function useChat(
18481875
// Ephemeral panels were never persisted; nothing to delete server-side.
18491876
if (isEphemeralResource({ type: resourceType, id: resourceId, title: '' })) return
18501877

1878+
const existing = resourcesRef.current.find(
1879+
(resource) => resource.type === resourceType && resource.id === resourceId
1880+
)
1881+
const persistChatId = chatIdRef.current ?? selectedChatIdRef.current
1882+
const persistenceScopeId = persistChatId ?? pendingChatKeyRef.current
18511883
const {
18521884
inFlight: inFlightAdd,
18531885
scheduleDelete,
18541886
wasPending,
18551887
wasPersisted,
1856-
} = resourcePersistenceQueue.remove(resourceType, resourceId)
1888+
} = resourcePersistenceQueue.remove(
1889+
resourceType,
1890+
resourceId,
1891+
persistenceScopeId,
1892+
Boolean(existing && persistChatId)
1893+
)
18571894
if (wasPending && !inFlightAdd && !wasPersisted) return
18581895

1859-
const persistChatId = chatIdRef.current ?? selectedChatIdRef.current
18601896
if (!persistChatId) return
18611897
scheduleDelete(persistChatId, () =>
18621898
requestJson(removeMothershipChatResourceContract, {
@@ -1898,53 +1934,20 @@ export function useChat(
18981934
[workspaceId, removeResource]
18991935
)
19001936

1901-
const reorderResources = useCallback((newOrder: MothershipResource[]) => {
1902-
setResources(newOrder)
1903-
const persistChatId = chatIdRef.current ?? selectedChatIdRef.current
1904-
if (!persistChatId) return
1905-
const pendingKeys = pendingPersistResourceKeysRef.current
1906-
const inFlightAdds = inFlightResourceAddsRef.current
1907-
const hasUnsyncedAdds = newOrder.some((r) => {
1908-
const key = `${r.type}:${r.id}`
1909-
return pendingKeys.has(key) || inFlightAdds.has(key)
1910-
})
1911-
if (hasUnsyncedAdds) {
1912-
reorderNeededAfterFlushRef.current = true
1913-
if (pendingKeys.size === 0 && inFlightAdds.size > 0) {
1914-
Promise.allSettled(Array.from(inFlightAdds.values())).then(() => {
1915-
if (!reorderNeededAfterFlushRef.current) return
1916-
reorderNeededAfterFlushRef.current = false
1917-
const chatId = chatIdRef.current ?? selectedChatIdRef.current
1918-
if (!chatId) return
1919-
const order = [
1920-
...resourcesRef.current.filter(
1921-
(r) =>
1922-
!isEphemeralResource(r) &&
1923-
!pendingPersistResourceKeysRef.current.has(`${r.type}:${r.id}`)
1924-
),
1925-
...undisplayableResourcesRef.current,
1926-
]
1927-
if (order.length === 0) return
1928-
requestJson(reorderMothershipChatResourcesContract, {
1929-
body: { chatId, resources: order },
1930-
}).catch((err) => {
1931-
logger.warn('Failed to sync resource order after in-flight ADDs', err)
1932-
})
1933-
})
1934-
}
1935-
return
1936-
}
1937-
const persistableResources = [
1938-
...newOrder.filter((r) => !isEphemeralResource(r)),
1939-
...undisplayableResourcesRef.current,
1940-
]
1941-
if (persistableResources.length === 0) return
1942-
requestJson(reorderMothershipChatResourcesContract, {
1943-
body: { chatId: persistChatId, resources: persistableResources },
1944-
}).catch((err) => {
1945-
logger.warn('Failed to persist resource reorder', err)
1946-
})
1947-
}, [])
1937+
const reorderResources = useCallback(
1938+
(newOrder: MothershipResource[]) => {
1939+
setResources(newOrder)
1940+
const persistChatId = chatIdRef.current ?? selectedChatIdRef.current
1941+
if (!persistChatId) return
1942+
const persistableResources = [
1943+
...newOrder.filter((resource) => !isEphemeralResource(resource)),
1944+
...undisplayableResourcesRef.current,
1945+
]
1946+
pendingResourceReordersRef.current.set(persistChatId, persistableResources)
1947+
void flushPendingResourceReorder(persistChatId)
1948+
},
1949+
[flushPendingResourceReorder]
1950+
)
19481951

19491952
const ensureWorkflowToolResource = useCallback(
19501953
(toolArgs: Record<string, unknown>): string | undefined => {
@@ -2228,7 +2231,8 @@ export function useChat(
22282231
const streamOwnerId = chatIdRef.current
22292232
const pendingTurn = activeTurnRef.current
22302233
const pendingStreamId = streamIdRef.current ?? pendingTurn?.userMessageId
2231-
const pendingResources = resourcePersistenceQueue.getPendingUpdates()
2234+
const pendingResourceScopeId =
2235+
streamOwnerId ?? pendingTurn?.pendingChatKey ?? pendingChatKeyRef.current
22322236
const navigatedToDifferentChat =
22332237
sendingRef.current &&
22342238
initialChatId !== streamOwnerId &&
@@ -2273,18 +2277,7 @@ export function useChat(
22732277
if (pendingChatKey) {
22742278
useMothershipQueueStore.getState().migrate(pendingChatKey, resolvedChatId)
22752279
}
2276-
await Promise.allSettled(
2277-
pendingResources.map((update) => {
2278-
const { clearViewId, ...resource } = update
2279-
return requestJson(addMothershipChatResourceContract, {
2280-
body: {
2281-
chatId: resolvedChatId,
2282-
resource,
2283-
...(clearViewId === true ? { clearViewId: true as const } : {}),
2284-
},
2285-
})
2286-
})
2287-
)
2280+
await resourcePersistenceQueue.flush(resolvedChatId, pendingResourceScopeId)
22882281
queryClient.invalidateQueries({
22892282
queryKey: mothershipChatKeys.detail(resolvedChatId),
22902283
})
@@ -2331,8 +2324,6 @@ export function useChat(
23312324
setResources([])
23322325
setActiveResourceId(null)
23332326
useTableViewPinStore.getState().reset()
2334-
resourcePersistenceQueue.clear()
2335-
reorderNeededAfterFlushRef.current = false
23362327
resetEphemeralPreviewState()
23372328
// Rotate the bucket key; the previous chat's queue stays in the store.
23382329
// Release editing on the chat we're leaving (composer-scoped).

‎apps/sim/lib/copilot/resources/client-persistence-queue.test.ts‎

Lines changed: 72 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ describe('ResourcePersistenceQueue', () => {
6464
first.reject(new Error('offline'))
6565
await Promise.allSettled(Array.from(queue.inFlight.values()))
6666

67-
expect(queue.getPendingUpdates()).toEqual([{ ...TABLE_RESOURCE, clearViewId: true }])
67+
expect(queue.getPendingUpdates('chat-1')).toEqual([{ ...TABLE_RESOURCE, clearViewId: true }])
6868
await queue.flush('chat-1')
6969

7070
expect(persist.mock.calls[1]).toEqual(['chat-1', { ...TABLE_RESOURCE, clearViewId: true }])
@@ -82,7 +82,7 @@ describe('ResourcePersistenceQueue', () => {
8282
const queue = new ResourcePersistenceQueue({ persist, onError })
8383

8484
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-a' }, 'chat-1')
85-
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id)
85+
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id, 'chat-1')
8686
removal.scheduleDelete('chat-1', remove)
8787
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-b' }, 'chat-1')
8888
await Promise.resolve()
@@ -92,6 +92,7 @@ describe('ResourcePersistenceQueue', () => {
9292
stale.resolve({ success: true })
9393
await vi.waitFor(() => expect(persist).toHaveBeenCalledTimes(2))
9494

95+
expect(remove).not.toHaveBeenCalled()
9596
expect(queue.inFlight.size).toBe(1)
9697
fresh.resolve({ success: true })
9798
await Promise.allSettled(Array.from(queue.inFlight.values()))
@@ -109,7 +110,7 @@ describe('ResourcePersistenceQueue', () => {
109110
failed.reject(new Error('offline'))
110111
await Promise.allSettled(Array.from(queue.inFlight.values()))
111112

112-
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id)
113+
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id, 'chat-1')
113114
expect(removal.wasPending).toBe(true)
114115
expect(removal.wasPersisted).toBe(true)
115116
const remove = vi.fn().mockResolvedValue({ success: true })
@@ -128,7 +129,7 @@ describe('ResourcePersistenceQueue', () => {
128129
failed.reject(new Error('offline'))
129130
await Promise.allSettled(Array.from(queue.inFlight.values()))
130131

131-
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id)
132+
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id, 'chat-1')
132133
expect(removal.wasPending).toBe(true)
133134
expect(removal.wasPersisted).toBe(false)
134135
})
@@ -143,7 +144,7 @@ describe('ResourcePersistenceQueue', () => {
143144

144145
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-a' }, 'chat-1', TABLE_RESOURCE)
145146
await Promise.allSettled(Array.from(queue.inFlight.values()))
146-
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id)
147+
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id, 'chat-1')
147148
removal.scheduleDelete('chat-1', remove)
148149
await vi.waitFor(() => expect(remove).toHaveBeenCalledOnce())
149150

@@ -165,16 +166,80 @@ describe('ResourcePersistenceQueue', () => {
165166
.mockResolvedValueOnce({ success: true })
166167
const queue = new ResourcePersistenceQueue({ persist, onError })
167168

168-
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id)
169+
const removal = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id, 'chat-1')
169170
removal.scheduleDelete('chat-1', remove)
170171
first.reject(new Error('offline'))
171172
await Promise.allSettled(Array.from(queue.inFlight.values()))
172173

173-
expect(queue.pendingKeys.has(`${TABLE_RESOURCE.type}:${TABLE_RESOURCE.id}`)).toBe(true)
174+
expect(
175+
queue.getPendingResourceKeys('chat-1').has(`${TABLE_RESOURCE.type}:${TABLE_RESOURCE.id}`)
176+
).toBe(true)
174177
expect(onError).toHaveBeenCalledOnce()
175178
await queue.flush('chat-1')
176179

177180
expect(remove).toHaveBeenCalledTimes(2)
178181
expect(queue.pendingKeys.size).toBe(0)
179182
})
183+
184+
it('keeps failed writes isolated by chat until that chat is flushed again', async () => {
185+
const persist = vi
186+
.fn<(chatId: string, update: MothershipResourceUpdate) => Promise<unknown>>()
187+
.mockRejectedValueOnce(new Error('offline'))
188+
.mockResolvedValue({ success: true })
189+
const queue = new ResourcePersistenceQueue({ persist, onError })
190+
191+
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-a' }, 'chat-1')
192+
await vi.waitFor(() => expect(onError).toHaveBeenCalledOnce())
193+
194+
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-b' }, 'chat-2')
195+
await vi.waitFor(() => expect(persist).toHaveBeenCalledTimes(2))
196+
197+
expect(persist.mock.calls[1]).toEqual(['chat-2', { ...TABLE_RESOURCE, viewId: 'view-b' }])
198+
expect(queue.getPendingResourceKeys('chat-1')).toEqual(new Set(['table:table-1']))
199+
expect(queue.getPendingResourceKeys('chat-2')).toEqual(new Set())
200+
201+
await queue.flush('chat-1')
202+
203+
expect(persist.mock.calls[2]).toEqual(['chat-1', { ...TABLE_RESOURCE, viewId: 'view-a' }])
204+
expect(queue.getPendingResourceKeys('chat-1')).toEqual(new Set())
205+
})
206+
207+
it('adopts provisional writes when a new chat receives its durable id', async () => {
208+
const persist = vi
209+
.fn<(chatId: string, update: MothershipResourceUpdate) => Promise<unknown>>()
210+
.mockResolvedValue({ success: true })
211+
const queue = new ResourcePersistenceQueue({ persist, onError })
212+
213+
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-a' }, undefined, undefined, 'pending-chat-1')
214+
215+
expect(queue.getPendingResourceKeys('pending-chat-1')).toEqual(new Set(['table:table-1']))
216+
await queue.flush('chat-1', 'pending-chat-1')
217+
218+
expect(persist).toHaveBeenCalledWith('chat-1', { ...TABLE_RESOURCE, viewId: 'view-a' })
219+
expect(queue.getPendingResourceKeys('pending-chat-1')).toEqual(new Set())
220+
expect(queue.getPendingResourceKeys('chat-1')).toEqual(new Set())
221+
})
222+
223+
it('remembers a stored resource until a failed deletion eventually succeeds', async () => {
224+
const persist = vi
225+
.fn<(chatId: string, update: MothershipResourceUpdate) => Promise<unknown>>()
226+
.mockResolvedValueOnce({ success: true })
227+
.mockRejectedValueOnce(new Error('re-add failed'))
228+
const remove = vi.fn().mockRejectedValueOnce(new Error('delete failed'))
229+
const queue = new ResourcePersistenceQueue({ persist, onError })
230+
231+
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-a' }, 'chat-1', TABLE_RESOURCE)
232+
await Promise.allSettled(queue.getInFlightWrites('chat-1'))
233+
234+
const firstRemoval = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id, 'chat-1')
235+
firstRemoval.scheduleDelete('chat-1', remove)
236+
await Promise.allSettled(queue.getInFlightWrites('chat-1'))
237+
238+
queue.enqueue({ ...TABLE_RESOURCE, viewId: 'view-b' }, 'chat-1')
239+
await Promise.allSettled(queue.getInFlightWrites('chat-1'))
240+
241+
const secondRemoval = queue.remove(TABLE_RESOURCE.type, TABLE_RESOURCE.id, 'chat-1')
242+
expect(secondRemoval.wasPending).toBe(true)
243+
expect(secondRemoval.wasPersisted).toBe(true)
244+
})
180245
})

0 commit comments

Comments
 (0)