fix: keep finished turns finished and cached chats current in the flow chat pool

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

chat-sdk/README.md

@@ -227,7 +227,8 @@ set) means the turn could not run or be followed at all, such as a refused reque
 Methods: `sendMessage(text, { inputs? })`, `stop()`, `newConversation()`,
 `selectConversation(id)`, `loadConversations({ page?, perPage?, kind? })`,
 `deleteConversation(id)`, `renameConversation(id, title)`, `loadOlderMessages()`,
-`destroy()`. `kind` lists the flow editor's test chats (`'test'`), the deployed flow's
+`resumeTurn(turn)`, `refreshMessages()` (reads what another tab added to the open
+conversation), `destroy()`. `kind` lists the flow editor's test chats (`'test'`), the deployed flow's
 own (`'deployed'`, the server's default) or both (`'all'`); each `Conversation` carries
 `isTest`. A rename keeps the conversation's place in the list. Switching conversations
 stops following the current answer; the flow keeps running and, with server history,

chat-sdk/src/ai-sdk.ts

@@ -152,9 +152,19 @@ function chunkStream(
             for (const e of event.events) parts.apply(e)
             continue
           }
+          // A result polled after the stream failed: the round that was streaming stopped
+          // wherever the connection did. Sent chunks cannot be taken back, so the answer's
+          // rest follows them, or the whole answer when it does not continue them.
+          const cut = parts.openText ?? ''
           parts.closeOpen()
           failure = await failureText(api, entry.jobId, event.result, signal)
-          if (failure === undefined && !parts.streamedText) {
+          if (failure === undefined && event.streamLost) {
+            const answer = extractChatAnswer(event.result)
+            if (answer !== undefined && answer !== cut) {
+              parts.text(answer.startsWith(cut) ? answer.slice(cut.length) : answer)
+              parts.closeOpen()
+            }
+          } else if (failure === undefined && !parts.streamedText) {
             // No agent streamed: the flow's result is the answer.
             const answer = extractChatAnswer(event.result)
             if (answer !== undefined) {
@@ -199,6 +209,8 @@ async function failureText(
  */
 class PartWriter {
   streamedText = false
+  /** The text of the round still streaming, until a tool call or the end closes it. */
+  openText: string | undefined
   #textId: string | undefined
   #reasoningId: string | undefined
   #started = new Set<string>()
@@ -210,8 +222,10 @@ class PartWriter {
     this.streamedText = true
     if (!this.#textId) {
       this.#textId = randomId()
+      this.openText = ''
       this.emit({ type: 'text-start', id: this.#textId })
     }
+    this.openText += delta
     this.emit({ type: 'text-delta', id: this.#textId, delta })
   }
 
@@ -231,6 +245,7 @@ class PartWriter {
     if (this.#textId) {
       this.emit({ type: 'text-end', id: this.#textId })
       this.#textId = undefined
+      this.openText = undefined
     }
   }
 

chat-sdk/src/chat.ts

@@ -425,6 +425,19 @@ class ChatImpl implements Chat {
     }
   }
 
+  refreshMessages = async (): Promise<void> => {
+    const conversationId = this.#state.conversationId
+    if (
+      !conversationId ||
+      this.#turn ||
+      this.#state.history !== 'server' ||
+      this.#state.loadingMessages
+    ) {
+      return
+    }
+    await this.#syncFromServer(conversationId)
+  }
+
   destroy = (): void => {
     this.#leaveConversation()
   }
@@ -434,7 +447,10 @@ class ChatImpl implements Chat {
   async #follow(turn: Turn, onStreamStart: () => void): Promise<unknown> {
     let started = false
     for await (const event of followJob(this.#api, turn.jobId!, { signal: turn.controller.signal })) {
-      if (event.type === 'completed') return event.result
+      if (event.type === 'completed') {
+        if (event.streamLost) this.#dropCutRound(turn)
+        return event.result
+      }
       if (!started) {
         started = true
         // Persisted rows for the streaming step would duplicate what is streaming.
@@ -445,6 +461,19 @@ class ChatImpl implements Chat {
     throw new Error('windmill-chat: the job stream ended before the flow completed')
   }
 
+  /**
+   * The stream failed while a round's text was arriving, so that text stops wherever the
+   * connection did. It goes: the persisted rows or, without them, the flow result give
+   * the whole answer instead. Rounds a tool call closed were complete and stay.
+   */
+  #dropCutRound(turn: Turn): void {
+    if (!this.#turnActive(turn)) return
+    const cut = turn.assistantId
+    turn.assistantId = undefined
+    turn.streamedText = false
+    if (cut) this.#set({ messages: this.#state.messages.filter((m) => m.id !== cut) })
+  }
+
   #applyEvents(turn: Turn, events: AgentStreamEvent[]): void {
     if (events.length === 0 || !this.#turnActive(turn)) return
     let messages = [...this.#state.messages]

chat-sdk/src/follow.ts

@@ -11,7 +11,8 @@ const RESULT_POLL_MS = 2000
 export type FollowEvent =
   /** Agent events decoded from the job's result stream; empty when a chunk ended mid-line. */
   | { type: 'stream'; events: AgentStreamEvent[] }
-  | { type: 'completed'; result: unknown }
+  /** `streamLost`: the result was polled after the stream failed, so what streamed may stop short. */
+  | { type: 'completed'; result: unknown; streamLost?: boolean }
 
 /**
  * Follows a job to completion across the server's stream timeouts: every
@@ -91,7 +92,7 @@ export async function* followJob(
     try {
       const { completed, result } = await api.getCompletedResult(jobId, options.signal)
       if (completed) {
-        yield { type: 'completed', result }
+        yield { type: 'completed', result, streamLost: true }
         return
       }
     } catch (e) {

chat-sdk/src/react.ts

@@ -14,6 +14,7 @@ export type UseWindmillChat = ChatState &
     | 'deleteConversation'
     | 'renameConversation'
     | 'loadOlderMessages'
+    | 'refreshMessages'
   > & { chat: Chat }
 
 /**
@@ -69,7 +70,8 @@ export function useWindmillChat(options: ChatOptions): UseWindmillChat {
       loadConversations: chat.loadConversations,
       deleteConversation: chat.deleteConversation,
       renameConversation: chat.renameConversation,
-      loadOlderMessages: chat.loadOlderMessages
+      loadOlderMessages: chat.loadOlderMessages,
+      refreshMessages: chat.refreshMessages
     }),
     [state, chat]
   )

chat-sdk/src/types.ts

@@ -165,6 +165,11 @@ export interface Chat {
   /** Sets a conversation's title. The list keeps its order: only a turn moves a conversation. */
   renameConversation(conversationId: string, title: string): Promise<void>
   loadOlderMessages(): Promise<void>
+  /**
+   * Reads what the current conversation gained since its newest message held here, such
+   * as a turn another tab ran. Server history only; does nothing while a turn is followed.
+   */
+  refreshMessages(): Promise<void>
   /** Stops background work (stream, polling) and writes local history out. The chat stays usable. */
   destroy(): void
 }

chat-sdk/test/ai-sdk-chat.test.ts

@@ -59,4 +59,26 @@ describe('AI SDK Chat over the Windmill transport', () => {
     expect(chat.error?.message).toBe('boom')
     expect(memoryIds.length === 1 || calls.filter((c) => c.method === 'POST')[1].url.searchParams.get('memory_id') === memoryIds[0]).toBe(true)
   })
+
+  test('a text part cut by a lost stream is completed from the polled result', async () => {
+    let streams = 0
+    const { fetch } = fetchMock(
+      (c) => (c.method === 'POST' && c.url.pathname === `/api/w/ws/jobs/run/f/${FLOW}` ? text('job-1') : undefined),
+      (c) =>
+        c.url.pathname.endsWith('/getupdate_sse/job-1')
+          ? ++streams === 1
+            ? sse([{ type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'The ans' }), stream_offset: 1 }])
+            : text('bad gateway', 502)
+          : undefined,
+      (c) =>
+        c.url.pathname.endsWith('/get_result_maybe/job-1')
+          ? json({ completed: true, success: true, result: { output: 'The answer is 42', messages: [] } })
+          : undefined
+    )
+    const transport = createWindmillChatTransport({ baseUrl: 'http://wm.test', workspace: 'ws', flowPath: FLOW, token: 'tok', fetch })
+    const chat = new Chat({ id: 'cut-chat', transport })
+    await chat.sendMessage({ text: 'what is it?' })
+    const texts = chat.messages[1].parts.filter((p) => p.type === 'text').map((p) => (p as { text: string }).text)
+    expect(texts.join('')).toBe('The answer is 42')
+  }, 15000)
 })

chat-sdk/test/chat.test.ts

@@ -783,6 +783,29 @@ describe('createChat with server history', () => {
     expect(calls.some((c) => c.url.pathname.includes('/jobs_u/get/'))).toBe(false)
   })
 
+  test('an answer cut by a lost stream gives way to the polled result', async () => {
+    let streams = 0
+    const { fetch } = fetchMock(
+      run,
+      (c) =>
+        c.url.pathname === streamPath
+          ? ++streams === 1
+            ? sse([{ type: 'update', new_result_stream: ndjson({ type: 'token_delta', content: 'Hel' }), stream_offset: 1 }])
+            : text('bad gateway', 502)
+          : undefined,
+      (c) =>
+        c.url.pathname.endsWith('/get_result_maybe/job-1')
+          ? json({ completed: true, success: true, result: { windmill_chat_answer: 'Hello, full answer' } })
+          : undefined
+    )
+    const chat = createChat(options({ history: 'none' }, fetch))
+    await chat.sendMessage('hi')
+    expect(chat.getState().messages.map((m) => [m.role, m.content, m.pending])).toEqual([
+      ['user', 'hi', false],
+      ['assistant', 'Hello, full answer', false]
+    ])
+  }, 15000)
+
   test('a stream that keeps failing hands the turn to polling the job', async () => {
     const { fetch } = fetchMock(
       run,

frontend/src/lib/components/flows/conversations/FlowChat.svelte

@@ -113,10 +113,7 @@
 			resumeTurn: (host, turn) => host.resumeTurn(turn),
 			isRunFinished: async (jobId) => (await api.getCompletedResult(jobId)).completed
 		})
-		const unsubscribeList = createdList.subscribe((s) => {
-			listState = s
-			createdPool.setListed(s.conversations)
-		})
+		const unsubscribeList = createdList.subscribe((s) => (listState = s))
 		const unsubscribePool = createdPool.subscribe((s) => (poolState = s))
 		listChat = createdList
 		pool = createdPool

frontend/src/lib/components/flows/conversations/FlowConversationsSidebar.svelte

@@ -85,7 +85,11 @@
 			// selected kind brings its own, whichever of the two lands last.
 			l.setLoader(async (page, perPage) => {
 				const requested = kind
+				// Only a listing read now says which turns run: the rows the chat holds keep
+				// what the last one said, and a rename or delete publishes those again.
+				const since = pool.listingStarted()
 				const rows = await c.loadConversations({ page, perPage, kind: requested })
+				pool.setListed(rows, since)
 				return requested === kind ? rows : items
 			})
 			l.setDeleteItemFn(async (id: string) => {

frontend/src/lib/components/flows/conversations/flowChatPool.test.ts

@@ -30,11 +30,15 @@ function fakeChat() {
 		resumeTurn: vi.fn(async () => {}),
 		stop: vi.fn(async () => {}),
 		newConversation: vi.fn(),
-		selectConversation: vi.fn(async (id: string) => set({ conversationId: id })),
+		// Like the SDK: the id and the loading flag are set before the first page is read.
+		selectConversation: vi.fn(async (id: string) =>
+			set({ conversationId: id, loadingMessages: true })
+		),
 		loadConversations: vi.fn(async () => []),
 		deleteConversation: vi.fn(async () => {}),
 		renameConversation: vi.fn(async () => {}),
 		loadOlderMessages: vi.fn(async () => {}),
+		refreshMessages: vi.fn(async () => {}),
 		destroy: vi.fn()
 	} satisfies Chat
 	return { chat, set }
@@ -75,7 +79,7 @@ describe('FlowChatPool', () => {
 	it('runs a turn in each of two conversations and counts an answer that lands out of view', () => {
 		const { pool: p, chatOf } = pool()
 		p.select('a')
-		chatOf('a').set({ status: 'streaming' })
+		chatOf('a').set({ loadingMessages: false, status: 'streaming' })
 		p.select('b')
 		chatOf('b').set({ status: 'submitted' })
 		expect(p.getState().activity).toEqual({ a: 'running', b: 'running' })
@@ -95,10 +99,10 @@ describe('FlowChatPool', () => {
 		let finished = false
 		const { pool: p, chatOf, hosts } = pool({ isRunFinished: async () => finished })
 		const turn = { jobId: 'job-1', userSeq: 7 }
-		p.setListed([
-			conversation('a', { runningTurn: turn }),
-			conversation('b', { runningTurn: turn })
-		])
+		p.setListed(
+			[conversation('a', { runningTurn: turn }), conversation('b', { runningTurn: turn })],
+			p.listingStarted()
+		)
 		expect(p.getState().activity).toEqual({ a: 'running', b: 'running' })
 
 		p.select('a')
@@ -109,6 +113,30 @@ describe('FlowChatPool', () => {
 		p.destroy()
 	})
 
+	it('does not take a turn back from a listing requested before it ended', () => {
+		const { pool: p, chatOf, hosts } = pool()
+		p.select('a')
+		chatOf('a').set({ status: 'streaming' })
+		const since = p.listingStarted()
+		chatOf('a').set({ status: 'idle', messages: [answer('m1')] })
+		p.setListed([conversation('a', { runningTurn: { jobId: 'job-1', userSeq: 3 } })], since)
+		expect(p.getState().activity).toEqual({})
+		expect(hosts.resumeTurn).not.toHaveBeenCalled()
+		p.destroy()
+	})
+
+	it('counts no answer of a first page that lands after the reader moved on, and rereads a held chat on return', () => {
+		const { pool: p, chatOf } = pool()
+		p.select('a')
+		p.select('b')
+		chatOf('a').set({ loadingMessages: false, messages: [answer('old-1'), answer('old-2')] })
+		expect(p.getState().unread).toEqual({})
+
+		p.select('a')
+		expect(chatOf('a').chat.refreshMessages).toHaveBeenCalledTimes(1)
+		p.destroy()
+	})
+
 	it('releases settled chats past the budget, never one still running', () => {
 		const { pool: p, chatOf, hosts } = pool({ keepSettled: 1 })
 		p.select('busy')

frontend/src/lib/components/flows/conversations/flowChatPool.ts

@@ -65,6 +65,11 @@ interface Entry<H> extends PooledChat<H> {
 	lastShownAt: number
 	/** Assistant messages already counted, so a message counts once as it settles. */
 	counted: Set<string>
+	/** Its first page has been read: answers already there were never unread. */
+	loaded: boolean
+	busy: boolean
+	/** The pool's clock when its last turn ended. */
+	settledAt: number
 }
 
 /**
@@ -135,22 +140,30 @@ export class FlowChatPool<H> {
 		entry.lastShownAt = ++this.#clock
 		this.#unread.delete(conversationId)
 		const turn = this.#running.get(conversationId)
-		if (turn && !isBusy(entry.chat.getState().status)) {
+		if (turn && !entry.busy) {
 			this.#running.delete(conversationId)
 			this.#options.resumeTurn(entry.host, turn)
+		} else if (entry.loaded && !entry.busy) {
+			// Held while another conversation was shown: another tab may have written since.
+			void entry.chat.refreshMessages()
 		}
 		this.#evict()
 		this.#publish()
 	}
 
+	/** Marks a listing request; pass the mark to `setListed` with what it returns. */
+	listingStarted = (): number => ++this.#clock
+
 	/**
-	 * What the conversation list last said. A conversation it reports running that no chat
-	 * here is following gets its run polled, so its row stops saying so when the run ends.
+	 * What a listing requested at `since` said. A conversation it reports running that no
+	 * chat here is following gets its run polled, so its row stops saying so when the run
+	 * ends. A turn that ended here after the request is not running, whatever it said.
 	 */
-	setListed = (conversations: readonly Conversation[]): void => {
+	setListed = (conversations: readonly Conversation[], since: number): void => {
 		for (const conversation of conversations) {
 			const followed = this.#entries.get(conversation.id)
-			if (conversation.runningTurn && !(followed && isBusy(followed.chat.getState().status))) {
+			const stale = followed && (followed.busy || followed.settledAt > since)
+			if (conversation.runningTurn && !stale) {
 				this.#running.set(conversation.id, conversation.runningTurn)
 			} else {
 				this.#running.delete(conversation.id)
@@ -195,7 +208,10 @@ export class FlowChatPool<H> {
 			host: this.#options.createHost(chat),
 			unsubscribe: () => {},
 			lastShownAt: ++this.#clock,
-			counted: new Set()
+			counted: new Set(),
+			loaded: conversationId === undefined,
+			busy: false,
+			settledAt: 0
 		}
 		entry.unsubscribe = chat.subscribe((state) => this.#onChatState(entry, state))
 		return entry
@@ -211,14 +227,20 @@ export class FlowChatPool<H> {
 		}
 		const id = state.conversationId
 		if (id === undefined) return
-		if (isBusy(state.status)) this.#running.delete(id)
+		const busy = isBusy(state.status)
+		if (busy) this.#running.delete(id)
+		else if (entry.busy) entry.settledAt = ++this.#clock
+		entry.busy = busy
 		const shown = id === this.#selectedId
 		for (const message of state.messages) {
 			if (message.role !== 'assistant' || message.pending || entry.counted.has(message.id)) continue
 			entry.counted.add(message.id)
-			if (!shown && !state.loadingMessages) this.#unread.set(id, (this.#unread.get(id) ?? 0) + 1)
+			if (!shown && entry.loaded && !state.loadingMessages) {
+				this.#unread.set(id, (this.#unread.get(id) ?? 0) + 1)
+			}
 		}
-		if (!isBusy(state.status)) this.#evict()
+		if (!state.loadingMessages) entry.loaded = true
+		if (!busy) this.#evict()
 		this.#publish()
 	}
 

frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts

@@ -47,6 +47,7 @@ function fakeChat(initial: ChatState = idleState()) {
 		deleteConversation: vi.fn(async () => {}),
 		renameConversation: vi.fn(async () => {}),
 		loadOlderMessages: vi.fn(async () => {}),
+		refreshMessages: vi.fn(async () => {}),
 		destroy: vi.fn()
 	} satisfies Chat
 	const set = (patch: Partial<ChatState>) => {