fix(chat): settle a turn on what it produced, not on its last read

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) <noreply@anthropic.com>

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 })
 	}
 
 	/**

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()