fix: ignore a stale running-turn snapshot, keep a withdrawn chat's draft, poll after clean stream ends

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

backend/windmill-api/openapi.yaml

@@ -11368,6 +11368,11 @@ paths:
           content:
             application/json:
               schema: {}
+        "409":
+          description: >-
+            Chat-enabled flow only: the conversation named by `memory_id` is still answering
+            a message. The body is JSON: `{ "error": string, "running_turn": { "job_id",
+            "user_seq" } }`.
 
   /w/{workspace}/jobs/run_and_stream/f/{path}:
     post:
@@ -11409,6 +11414,11 @@ paths:
             text/event-stream:
               schema:
                 type: string
+        "409":
+          description: >-
+            Chat-enabled flow only: the conversation named by `memory_id` is still answering
+            a message. The body is JSON: `{ "error": string, "running_turn": { "job_id",
+            "user_seq" } }`.
 
     get:
       summary: run flow by path with GET and stream updates via SSE
@@ -11442,6 +11452,11 @@ paths:
             text/event-stream:
               schema:
                 type: string
+        "409":
+          description: >-
+            Chat-enabled flow only: the conversation named by `memory_id` is still answering
+            a message. The body is JSON: `{ "error": string, "running_turn": { "job_id",
+            "user_seq" } }`.
 
   /w/{workspace}/jobs/run_and_stream/fv/{version}:
     post:
@@ -11489,6 +11504,11 @@ paths:
             text/event-stream:
               schema:
                 type: string
+        "409":
+          description: >-
+            Chat-enabled flow only: the conversation named by `memory_id` is still answering
+            a message. The body is JSON: `{ "error": string, "running_turn": { "job_id",
+            "user_seq" } }`.
 
     get:
       summary: run flow by version with GET and stream updates via SSE
@@ -11528,6 +11548,11 @@ paths:
             text/event-stream:
               schema:
                 type: string
+        "409":
+          description: >-
+            Chat-enabled flow only: the conversation named by `memory_id` is still answering
+            a message. The body is JSON: `{ "error": string, "running_turn": { "job_id",
+            "user_seq" } }`.
 
   /w/{workspace}/jobs/run_and_stream/p/{path}:
     post:

chat-sdk/src/chat.ts

@@ -233,6 +233,17 @@ class ChatImpl implements Chat {
       // The conversation's first page may still be on its way; it would land over the turn.
       await this.#selecting
       if (!this.#turnActive(turn)) return
+      // A listing is a snapshot: the turn it named can have ended and another one started
+      // since. Following it would drop the newer turn's rows and leave the chat idle while
+      // that turn runs, so it is left to the next listing to name the turn that runs now.
+      const newerTurn = this.#state.messages.some(
+        (m) => m.role === 'user' && m.seq !== undefined && m.seq > userSeq
+      )
+      if (newerTurn) {
+        if (this.#turn === turn) this.#turn = undefined
+        this.#set({ status: 'idle' })
+        return
+      }
       // The stream replays the turn from its start, so the rows it already wrote go and
       // come back as it replays them. The message that started it stays: it is the turn's
       // anchor, and a long turn can have pushed it off the page this chat opened on.

chat-sdk/src/follow.ts

@@ -85,9 +85,15 @@ export async function* followJob(
       continue
     }
     if (options.signal?.aborted) throw abortError()
-    // The server closes the connection after its timeout; a dropped connection looks
-    // the same minus the event. Either way the offset lets the next one resume.
-    if (!reopen) await sleep(RECONNECT_DELAY_MS, options.signal)
+    // The server closes the connection after its timeout; a connection that ends without
+    // that event, and without the job completing, carried nothing to its end. The offset
+    // lets the next one resume, and it counts like a failed one so a gateway closing every
+    // stream this way still reaches the polling below rather than reconnecting for ever.
+    if (!reopen) {
+      failures++
+      if (failures >= MAX_CONNECTION_FAILURES) break
+      await sleep(RECONNECT_DELAY_MS, options.signal)
+    }
   }
   while (true) {
     await sleep(RESULT_POLL_MS, options.signal)

chat-sdk/test/chat.test.ts

@@ -966,6 +966,51 @@ describe('createChat with server history', () => {
     expect(state.conversations.map((c) => c.id)).toEqual(['conv'])
   })
 
+  test('a running turn the listing named is not followed once a newer one has started', async () => {
+    const { fetch, calls } = fetchMock(
+      (c) =>
+        c.url.pathname.endsWith('/messages')
+          ? json([
+              messageRow(50, 'user', 'first'),
+              messageRow(51, 'assistant', 'first answer'),
+              messageRow(52, 'user', 'second')
+            ])
+          : undefined
+    )
+    const chat = createChat(options({}, fetch))
+    await chat.selectConversation('conv')
+    // The listing named the first turn; it ended and another one started before this select.
+    await chat.resumeTurn({ jobId: 'job-1', userSeq: 50 })
+    expect(chat.getState().status).toBe('idle')
+    expect(chat.getState().messages.map((m) => m.serverId)).toEqual(['row-50', 'row-51', 'row-52'])
+    expect(calls.some((c) => c.url.pathname.includes('getupdate_sse'))).toBe(false)
+    // The chat is free: a message sent now starts its own turn rather than being refused.
+    expect(chat.getState().error).toBeUndefined()
+  })
+
+  test('a stream that keeps ending before the job completes hands the turn to polling', async () => {
+    let streams = 0
+    const { fetch } = fetchMock(
+      run,
+      // A gateway that answers 200 and closes cleanly, carrying nothing to the end.
+      (c) => (c.url.pathname === streamPath ? (streams++, sse([])) : undefined),
+      (c) =>
+        c.url.pathname.endsWith('/get_result_maybe/job-1')
+          ? json({ completed: true, success: true, result: { windmill_chat_answer: 'polled' } })
+          : undefined,
+      (c) =>
+        c.url.pathname.endsWith('/messages')
+          ? json([messageRow(81, 'user', 'hi'), messageRow(82, 'assistant', 'polled')])
+          : undefined,
+      (c) => (c.url.pathname === '/api/w/ws/flow_conversations/list' ? json([]) : undefined)
+    )
+    const chat = createChat(options({}, fetch))
+    await chat.sendMessage('hi')
+    expect(streams).toBe(3)
+    expect(chat.getState().status).toBe('idle')
+    expect(chat.getState().messages.map((m) => m.content)).toEqual(['hi', 'polled'])
+  }, 15000)
+
   test('resuming a turn whose message is off the first page replays it without duplicating rows', async () => {
     const { fetch } = fetchMock(
       (c) =>

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

@@ -124,6 +124,7 @@
 			disposeHost: (host) => host.dispose(),
 			hasUnsentDraft: (host) => host.hasUnsentDraft,
 			resumeTurn: (host, turn) => host.resumeTurn(turn),
+			moveUnsentDraft: (from, to) => to.adoptUnsentDraft(from.takeUnsentDraft()),
 			isRunFinished: async (jobId) => (await api.getCompletedResult(jobId)).completed
 		})
 		const unsubscribeList = createdList.subscribe((s) => (listState = s))

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

@@ -60,7 +60,7 @@ function pool(
 	} = {}
 ) {
 	const chats: ReturnType<typeof fakeChat>[] = []
-	const hosts = { resumeTurn: vi.fn(), dispose: vi.fn() }
+	const hosts = { resumeTurn: vi.fn(), dispose: vi.fn(), moveUnsentDraft: vi.fn() }
 	const created = new FlowChatPool<{ chat: Chat }>({
 		createChat: () => {
 			const fake = fakeChat()
@@ -71,6 +71,7 @@ function pool(
 		disposeHost: hosts.dispose,
 		hasUnsentDraft: options.hasUnsentDraft ?? (() => false),
 		resumeTurn: (host, turn) => hosts.resumeTurn(host.chat, turn),
+		moveUnsentDraft: (from, to) => hosts.moveUnsentDraft(from.chat, to.chat),
 		isRunFinished: options.isRunFinished ?? (async () => false),
 		keepSettled: options.keepSettled,
 		pollMs: 10
@@ -162,6 +163,24 @@ describe('FlowChatPool', () => {
 		p.destroy()
 	})
 
+	it("hands a withdrawn chat's unsent draft to the new chat that took its place", async () => {
+		const { pool: p, fakeOf, hosts } = pool()
+		const withdrawn = p.selected
+		const { set } = fakeOf(withdrawn.chat)
+		set({ conversationId: 'new-1' })
+		// The reader opens another new chat while the first message is still uploading.
+		const kept = p.newChat()
+		expect(kept).not.toBe(withdrawn)
+		set({ conversationId: undefined })
+		// Not released yet: the send reports what it could not do after this.
+		expect(withdrawn.chat.destroy).not.toHaveBeenCalled()
+		await new Promise((resolve) => setTimeout(resolve, 0))
+		expect(hosts.moveUnsentDraft).toHaveBeenCalledWith(withdrawn.chat, kept.chat)
+		expect(withdrawn.chat.destroy).toHaveBeenCalled()
+		expect(p.selected).toBe(kept)
+		p.destroy()
+	})
+
 	it('keeps a settled chat that still holds something typed and never sent', () => {
 		let held = ''
 		const { pool: p, chatOf } = pool({

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

@@ -57,6 +57,8 @@ export interface FlowChatPoolOptions<H> {
 	hasUnsentDraft(host: H): boolean
 	/** Follows a turn another page started, through the host so what it queues waits for it. */
 	resumeTurn(host: H, turn: RunningTurn): void
+	/** Hands what was typed in a chat being released to the one taking its place. */
+	moveUnsentDraft(from: H, to: H): void
 	/** Whether a run has ended, for a running conversation this page holds no chat for. */
 	isRunFinished(jobId: string): Promise<boolean>
 	/** Settled conversations kept in memory beside the shown and the busy ones. */
@@ -93,6 +95,8 @@ export class FlowChatPool<H> {
 	readonly #running = new Map<string, RunningTurn>()
 	/** Failed reads in a row, per conversation, for a run this pool has no chat for. */
 	readonly #pollFailures = new Map<string, number>()
+	/** Chats on their way out, kept until the send that withdrew them has settled. */
+	readonly #retiring = new Set<Entry<H>>()
 	readonly #unread = new Map<string, number>()
 	readonly #listeners = new Set<(state: FlowChatPoolState) => void>()
 	#selectedId: string | undefined
@@ -198,7 +202,9 @@ export class FlowChatPool<H> {
 		this.#destroyed = true
 		clearInterval(this.#poll)
 		for (const entry of this.#entries.values()) this.#release(entry)
+		for (const entry of this.#retiring) this.#release(entry)
 		if (this.#draft) this.#release(this.#draft)
+		this.#retiring.clear()
 		this.#entries.clear()
 		this.#draft = undefined
 		this.#listeners.clear()
@@ -266,9 +272,22 @@ export class FlowChatPool<H> {
 			this.#running.delete(key)
 			if (this.#selectedId === key) this.#selectedId = undefined
 		}
-		// A new chat opened meanwhile is the draft now; this one has nothing left to show.
-		if (this.#draft) this.#release(entry)
-		else this.#draft = entry
+		if (!this.#draft) {
+			this.#draft = entry
+			this.#publish()
+			return
+		}
+		// A new chat opened meanwhile is the draft now, so this chat has nowhere to show. It
+		// is released only once its send has reported what it could not do — the refusal
+		// reaches it after this — and what the reader typed moves to the chat in its place.
+		const kept = this.#draft
+		this.#retiring.add(entry)
+		setTimeout(() => {
+			if (this.#destroyed || !this.#retiring.delete(entry)) return
+			this.#options.moveUnsentDraft(entry.host, kept.host)
+			this.#release(entry)
+			this.#publish()
+		}, 0)
 		this.#publish()
 	}
 

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

@@ -610,6 +610,25 @@ export class FlowChatViewHost implements ChatViewHost {
 		this.#queue = emptyQueue()
 		return taken
 	}
+	/**
+	 * Everything typed here and not sent, handed to the chat that takes this one's place: a
+	 * new chat whose first message never ran leaves with nothing of its own, and what the
+	 * reader wrote belongs in the composer they are looking at rather than in a released host.
+	 */
+	takeUnsentDraft(): Queue {
+		const queued = this.#takeQueue()
+		const returned = this.#returned
+		this.#returned = emptyQueue()
+		return {
+			text: [returned.text, queued.text].filter(Boolean).join('\n'),
+			images: [...returned.images, ...queued.images],
+			blobs: [...returned.blobs, ...queued.blobs]
+		}
+	}
+	adoptUnsentDraft({ text, images, blobs }: Queue) {
+		if (!text && images.length === 0 && blobs.length === 0) return
+		this.#returnDraft(text, images, blobs)
+	}
 	setComposerStaged = () => {}
 	clearComposerStaged = () => {}
 	attachmentBytesExcluding = () => 0