diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index cc15f033d9..948f25ae74 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -157,6 +157,9 @@ export class FlowChatManager { /** Each conversation's rows, live ones included, so a turn keeps writing while the * reader is in another chat. Doubles as the load cache: rows here are never re-fetched. */ + /** Bumped when the chat is re-pointed at another flow. Work started before that must not + * write what it fetched into the chat that replaced it. */ + #generation = 0 #rowsById = $state>({}) /** How far back each conversation has been paged. Held per conversation for the same * reason the rows are: a cached chat keeps its scrollback when the reader returns to it, @@ -488,6 +491,7 @@ export class FlowChatManager { * messages into that transcript and its agent's memory. */ cleanup() { + this.#generation++ for (const conversationId of Object.keys(this.#status)) this.endTurn(conversationId) this.selectedConversationId = undefined this.#rowsById = {} @@ -585,9 +589,11 @@ export class FlowChatManager { */ async selectLatestConversation() { if (this.selectedConversationId || !this.#workspace() || !this.#path) return + const startedIn = this.#generation const [latest] = await this.loadConversations(1, 1) - // Re-checked after the await: a message sent meanwhile has already opened its own. - if (!latest || this.selectedConversationId) return + // Re-checked after the await: a message sent meanwhile has already opened its own, or + // the chat has been re-pointed and this is the previous flow's latest conversation. + if (!latest || this.selectedConversationId || startedIn !== this.#generation) return await this.selectConversation(latest.id) } @@ -816,6 +822,7 @@ export class FlowChatManager { try { const previousScrollHeight = this.messagesContainer?.scrollHeight || 0 + const startedIn = this.#generation const response = await FlowConversationsService.listConversationMessages({ workspace: this.#workspace()!, conversationId: conversationIdToUse, @@ -823,6 +830,10 @@ export class FlowChatManager { perPage: this.#perPage }) + // The chat was re-pointed while this was in flight; these rows belong to a flow it + // no longer shows, and the hold below would be released for the wrong one. + if (startedIn !== this.#generation) return + if (reset) { this.#rowsById[conversationIdToUse] = response this.#pagedTo[conversationIdToUse] = 1 @@ -1376,9 +1387,12 @@ export class FlowChatManager { await new Promise((resolve) => setTimeout(resolve, SETTLE_POLL_MS)) } if (signal.aborted) return + const startedIn = this.#generation try { await this.pollConversationMessages(conversationId, { removeTempMessages: true }) } catch {} + // `endTurn` would put this conversation's status back into a map the re-point emptied. + if (signal.aborted || startedIn !== this.#generation) return this.endTurn(conversationId, { settled: true }) } diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts index ff5be08669..54bddd77cc 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts @@ -135,16 +135,40 @@ describe('a chat re-pointed at another flow', () => { * conversation and send into its transcript and its agent's memory. */ it('forgets the flow it was pointed at when it is re-pointed', async () => { - vi.mocked(FlowConversationsService.listConversationMessages).mockResolvedValue([] as any) + vi.mocked(FlowConversationsService.listConversationMessages).mockResolvedValue( + rows('a', 2) as any + ) const manager = managerWithRows() await manager.selectConversation('a') expect(manager.selectedConversationId).toBe('a') + expect(manager.liveRowIds.size).toBe(2) manager.cleanup() expect(manager.selectedConversationId).toBeUndefined() - expect(manager.messages).toEqual([]) - expect(manager.conversations).toEqual([]) + expect(manager.liveRowIds.size).toBe(0) + }) + + /** + * Forgetting has to hold against work already in flight: a transcript fetched for the + * flow just left would otherwise be written into the one that replaced it. + */ + it('does not let a load started before the re-point write its rows back', async () => { + let release = () => {} + const held = new Promise((resolve) => (release = resolve)) + vi.mocked(FlowConversationsService.listConversationMessages).mockImplementation((async () => { + await held + return rows('a', 2) + }) as any) + const manager = managerWithRows() + const loading = manager.selectConversation('a') + + manager.cleanup() + release() + await loading + + expect(manager.liveRowIds.size).toBe(0) + expect(manager.selectedConversationId).toBeUndefined() }) })