From ce9ae589b4d18bfae7620ae54775fad4f3e9dc11 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Wed, 16 Sep 2026 14:02:24 +0200 Subject: [PATCH] fix(chat): settle a turn on what it produced, not on its last read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three things the settle got wrong once it could put an answer on screen of its own. The answer it shows when a turn produced nothing was decided by how many rows the final read brought back. A turn with no stream is read the whole time it runs, though, so its answer is routinely on screen long before the job reports itself over — and the final read then returns nothing and the answer was shown a second time beside the row it already had. Whether a turn produced anything is a question about the whole settle, and is now asked of the rows themselves. Settling arms whatever was queued behind the turn, which a run that failed must not do: the queued turn would go straight into the conversation that just failed, before the reader has seen why. And the sidebar's filter starts a read without waiting for the one before it, so switching quickly leaves several in flight and the list keeps whichever answers last. A read whose filter has been changed since now answers with nothing rather than with the wrong chats. Co-Authored-By: Claude Opus 5 (1M context) --- .../conversations/FlowChatManager.svelte.ts | 27 ++++-- .../conversations/FlowChatManager.test.ts | 92 +++++++++++++++++++ 2 files changed, 111 insertions(+), 8 deletions(-) diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index 32b8e13a17..0a8e780eb1 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -831,14 +831,19 @@ export class FlowChatManager { private async loadConversations(page: number, perPage: number) { if (!this.#workspace() || !this.#path) return [] + // The filter this read is for. Switching it starts another read without waiting for + // this one, and the list keeps whichever answers last — so a read whose filter has + // been changed since answers with nothing rather than with the wrong chats. + const kind = this.conversationKind try { const response = await FlowConversationsService.listFlowConversations({ workspace: this.#workspace()!, flowPath: this.#path, - kind: this.conversationKind, + kind, page: page, perPage: perPage }) + if (kind !== this.conversationKind) return [] return response } catch (error) { console.error('Failed to load conversations:', error) @@ -1491,6 +1496,10 @@ export class FlowChatManager { // arriving without a stream — a turn that never had one, one whose stream is gone, // a conversation picked back up on the polling path — gets it once. turn.startPolling() + // What the conversation showed before this turn produced anything. The catch-up read + // runs throughout the wait below, so whether the turn has an answer is a question + // about the whole settle rather than about its last read. + const rowsBefore = this.#rowsOf(turn.conversationId).length const api = this.#chatApi() const signal = turn.signal let flowResult: unknown @@ -1510,13 +1519,12 @@ export class FlowChatManager { await new Promise((resolve) => setTimeout(resolve, SETTLE_POLL_MS)) } if (!this.#isCurrent(turn)) return - const read = await this.#reconcileTurn(turn) + await this.#reconcileTurn(turn) if (!this.#isCurrent(turn)) return - // Nothing of this turn was read back and nothing was streamed, so the transcript has - // no answer to show and no later read is coming. The run's own result is the same - // answer the row would have carried, and showing it is what keeps a finished turn - // from reading as one that produced nothing. A reload replaces it with the row. - if (read === 0) { + // The turn put nothing on screen and no later read is coming. The run's own result is + // the same answer a row would have carried, and showing it is what keeps a finished + // turn from reading as one that produced nothing. A reload replaces it with the row. + if (this.#rowsOf(turn.conversationId).length === rowsBefore) { // A run that failed says so in an error rather than an answer, and the row has to // carry that: it is what the transcript reads the turn's outcome off, which is what // offers Retry and what holds a queued message back from running into the same @@ -1538,7 +1546,10 @@ export class FlowChatManager { ] } } - this.#endTurnIfCurrent(turn, { settled: true }) + // Settling arms whatever was queued behind this turn. A run that failed must not arm + // it: the next turn would go straight into the conversation that just failed, where + // the reader has not seen why yet. + this.#endTurnIfCurrent(turn, { settled: flowSucceeded }) } /** diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts index 934eb7d898..1a7cf8a2bb 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.test.ts @@ -5,6 +5,7 @@ import { createFlowChatManager } from './FlowChatManager.svelte' vi.mock('$lib/gen', () => ({ FlowConversationsService: { listConversationMessages: vi.fn(), + listFlowConversations: vi.fn(), deleteFlowConversation: vi.fn() }, JobService: { cancelQueuedJob: vi.fn() }, @@ -142,6 +143,29 @@ describe('unread bookkeeping', () => { * start of a turn that wrote a lot of rows — an agent calling several tools a round. Its * answer is among the rows that would be left behind. */ +/** + * The sidebar's filter starts a read and does not wait for it, so switching quickly leaves + * several in flight and the list keeps whichever answers last. An older one answering last + * would leave the sidebar showing chats of a filter the reader has already moved off. + */ +describe('switching the conversation filter', () => { + it('drops a read whose filter has been changed since', async () => { + let answer: ((rows: unknown[]) => void) | undefined + vi.mocked(FlowConversationsService.listFlowConversations) + .mockReset() + .mockImplementationOnce(() => new Promise((resolve) => (answer = resolve)) as any) + const manager = managerWithRows() + manager.conversationKind = 'test' + + const stale = (manager as any).loadConversations(1, 20) + await vi.waitFor(() => expect(answer).toBeTruthy()) + manager.conversationKind = 'deployed' + answer!([{ id: 'a-test-chat', is_test: true }]) + + expect(await stale).toEqual([]) + }) +}) + describe('reading a turn longer than one page', () => { const assistantRows = (from: number, count: number) => Array.from({ length: count }, (_, i) => ({ @@ -801,6 +825,74 @@ describe('a conversation opened while its run is still going', () => { expect(last?.success).toBe(false) }) + /** + * The catch-up read runs for the whole of a turn with no stream, so the answer can be on + * screen well before the job reports itself over. Asking only what the last read brought + * back would call that turn empty and show the run's result beside the row it already has. + */ + it('does not repeat an answer the catch-up read already got', async () => { + vi.mocked(FlowConversationsService.listConversationMessages) + .mockReset() + .mockResolvedValue([ + { + id: 'db-answer', + conversation_id: 'a', + message_type: 'assistant', + content: 'the answer, read back', + created_at: new Date().toISOString(), + created_seq: 5 + } + ] as any) + jobCompleted.value = true + jobCompleted.success = true + jobCompleted.result = { windmill_chat_answer: 'the answer, read back' } + const manager = (live = managerWithRows()) + ;(manager as any).initialize( + vi.fn(async () => 'job-1'), + 'u/admin/flow', + false + ) + manager.operatingWorkspace = () => 'ws' + manager.selectedConversationId = 'a' + manager.inputMessage = 'ask' + + await manager.sendMessage(undefined, undefined, 'a') + await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(false)) + + const answers = manager.messages.filter((m) => m.content === 'the answer, read back') + expect(answers).toHaveLength(1) + }) + + /** + * A queued message goes out when a turn settles. A turn that failed must not settle: the + * next one would run straight into the conversation that just failed, before the reader + * has seen why. + */ + it('does not arm a queued message when the run failed', async () => { + vi.mocked(FlowConversationsService.listConversationMessages) + .mockReset() + .mockResolvedValue([] as any) + jobCompleted.value = true + jobCompleted.success = false + jobCompleted.result = { error: { message: 'the provider refused' } } + const manager = (live = managerWithRows()) + ;(manager as any).initialize( + vi.fn(async () => 'job-1'), + 'u/admin/flow', + false + ) + manager.operatingWorkspace = () => 'ws' + manager.selectedConversationId = 'a' + manager.inputMessage = 'ask' + const settled: string[] = [] + manager.onTurnSettled = (id) => settled.push(id) + + await manager.sendMessage(undefined, undefined, 'a') + await vi.waitFor(() => expect(manager.isConversationBusy('a')).toBe(false)) + + expect(settled).toEqual([]) + }) + it('leaves no card spinning when Stop ends the turn', async () => { vi.mocked(FlowConversationsService.listConversationMessages) .mockReset()