fix: keep a chat that holds an unsent draft, and take one back when its first turn is withdrawn

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

backend/windmill-api/openapi.yaml

@@ -11286,6 +11286,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_wait_result/fv/{version}:
     post:
@@ -11326,6 +11331,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" } }`.
 
     get:
       summary: run flow by version with GET and wait until completion
@@ -14985,6 +14995,11 @@ paths:
               schema:
                 type: string
                 format: uuid
+        "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/batch_rerun_jobs:
     post:
@@ -15512,6 +15527,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/dynamic_select:
     post:

backend/windmill-common/src/flow_conversations.rs

@@ -144,6 +144,12 @@ async fn refuse_running_turn(
     ))
 }
 
+/// The running turn of each of `conversation_ids` that has one.
+///
+/// It answers for whatever ids it is given and checks no permission of its own, so the
+/// executor must be one the caller is entitled to read those conversations through: a
+/// `user_db` transaction under RLS, or a transaction holding ids the caller has already
+/// authorized. Handed a raw pool and ids from a request, it would report other users' jobs.
 pub async fn running_turns<'e, E: sqlx::PgExecutor<'e>>(
     executor: E,
     conversation_ids: &[Uuid],

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

@@ -122,7 +122,7 @@
 			createChat: () => createChat(options),
 			createHost: (chat) => new FlowChatViewHost(chat),
 			disposeHost: (host) => host.dispose(),
-			hasQueued: (host) => host.queuedMessage !== '',
+			hasUnsentDraft: (host) => host.hasUnsentDraft,
 			resumeTurn: (host, turn) => host.resumeTurn(turn),
 			isRunFinished: async (jobId) => (await api.getCompletedResult(jobId)).completed
 		})

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

@@ -53,7 +53,11 @@ function conversation(id: string, extra: Partial<Conversation> = {}): Conversati
 }
 
 function pool(
-	options: { keepSettled?: number; isRunFinished?: (jobId: string) => Promise<boolean> } = {}
+	options: {
+		keepSettled?: number
+		isRunFinished?: (jobId: string) => Promise<boolean>
+		hasUnsentDraft?: (host: { chat: Chat }) => boolean
+	} = {}
 ) {
 	const chats: ReturnType<typeof fakeChat>[] = []
 	const hosts = { resumeTurn: vi.fn(), dispose: vi.fn() }
@@ -65,14 +69,15 @@ function pool(
 		},
 		createHost: (chat) => ({ chat }),
 		disposeHost: hosts.dispose,
-		hasQueued: () => false,
+		hasUnsentDraft: options.hasUnsentDraft ?? (() => false),
 		resumeTurn: (host, turn) => hosts.resumeTurn(host.chat, turn),
 		isRunFinished: options.isRunFinished ?? (async () => false),
 		keepSettled: options.keepSettled,
 		pollMs: 10
 	})
 	const chatOf = (id: string) => chats.find((c) => c.chat.getState().conversationId === id)!
-	return { pool: created, chatOf, hosts }
+	const fakeOf = (chat: Chat) => chats.find((c) => c.chat === chat)!
+	return { pool: created, chatOf, fakeOf, hosts }
 }
 
 describe('FlowChatPool', () => {
@@ -137,6 +142,44 @@ describe('FlowChatPool', () => {
 		p.destroy()
 	})
 
+	it('takes a chat back as the new one when its first message never ran', () => {
+		const { pool: p, fakeOf } = pool()
+		const draft = p.selected
+		const { set } = fakeOf(draft.chat)
+		// A new chat's first message names its conversation, then is withdrawn: the upload
+		// failed, or Stop was pressed while it ran, so that conversation was never created.
+		set({ conversationId: 'new-1' })
+		expect(p.getState().selectedId).toBe('new-1')
+		set({ conversationId: undefined })
+		expect(p.getState().selectedId).toBeUndefined()
+		expect(p.get('new-1')).toBeUndefined()
+		expect(p.getState().activity).toEqual({})
+		// The next message mints its own id on that same chat, and the pool follows it there.
+		expect(p.selected).toBe(draft)
+		set({ conversationId: 'new-2' })
+		expect(p.getState().selectedId).toBe('new-2')
+		expect(p.get('new-2')).toBe(draft)
+		p.destroy()
+	})
+
+	it('keeps a settled chat that still holds something typed and never sent', () => {
+		let held = ''
+		const { pool: p, chatOf } = pool({
+			keepSettled: 1,
+			hasUnsentDraft: (host) => host.chat.getState().conversationId === held
+		})
+		p.select('typed')
+		held = 'typed'
+		for (const id of ['b', 'c', 'd']) p.select(id)
+		expect(p.get('typed')).toBeDefined()
+		expect(chatOf('typed').chat.destroy).not.toHaveBeenCalled()
+		// Once it is sent or taken back, the chat is releasable like any other.
+		held = ''
+		p.select('e')
+		expect(p.get('typed')).toBeUndefined()
+		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

@@ -27,6 +27,9 @@ export function isBusy(status: ChatState['status']): boolean {
 	return status === 'submitted' || status === 'streaming'
 }
 
+/** Reads of one run that may fail before its row stops saying the turn is running. */
+const POLL_GIVE_UP = 3
+
 /** What a conversation's row says about it. */
 export type ConversationActivity = 'running' | 'error' | 'idle'
 
@@ -49,8 +52,9 @@ export interface FlowChatPoolOptions<H> {
 	createChat(): Chat
 	createHost(chat: Chat): H
 	disposeHost(host: H): void
-	/** Whether a message typed during the turn waits in the host to go out. */
-	hasQueued(host: H): boolean
+	/** Whether the host holds text that was typed and never sent: queued behind the turn,
+	 * or handed back by a turn that refused it. Such a chat is never released. */
+	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
 	/** Whether a run has ended, for a running conversation this page holds no chat for. */
@@ -87,6 +91,8 @@ export class FlowChatPool<H> {
 	#draft: Entry<H> | undefined
 	/** Turns running in conversations this pool is not following, as the list reported them. */
 	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>()
 	readonly #unread = new Map<string, number>()
 	readonly #listeners = new Set<(state: FlowChatPoolState) => void>()
 	#selectedId: string | undefined
@@ -226,7 +232,14 @@ export class FlowChatPool<H> {
 			if (this.#selectedId === undefined) this.#selectedId = state.conversationId
 		}
 		const id = state.conversationId
-		if (id === undefined) return
+		if (id === undefined) {
+			// The chat gave its conversation back: a new chat's first message never ran, so the
+			// conversation the id named was never created. Held under that id, the entry would
+			// answer for a conversation that does not exist and mint a second one on the next
+			// message, so it goes back to being the chat a new conversation starts on.
+			if (entry !== this.#draft) this.#undoNewConversation(entry)
+			return
+		}
 		const busy = isBusy(state.status)
 		if (busy) this.#running.delete(id)
 		else if (entry.busy) entry.settledAt = ++this.#clock
@@ -244,6 +257,21 @@ export class FlowChatPool<H> {
 		this.#publish()
 	}
 
+	/** Takes an entry back out of the list of conversations, as the chat that starts one. */
+	#undoNewConversation(entry: Entry<H>): void {
+		for (const [key, held] of this.#entries) {
+			if (held !== entry) continue
+			this.#entries.delete(key)
+			this.#unread.delete(key)
+			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
+		this.#publish()
+	}
+
 	#activity(id: string): ConversationActivity {
 		const state = this.#entries.get(id)?.chat.getState()
 		if (this.#running.has(id) || (state && isBusy(state.status))) return 'running'
@@ -257,7 +285,7 @@ export class FlowChatPool<H> {
 			([id, entry]) =>
 				id !== this.#selectedId &&
 				!isBusy(entry.chat.getState().status) &&
-				!this.#options.hasQueued(entry.host)
+				!this.#options.hasUnsentDraft(entry.host)
 		)
 		settled.sort(([, a], [, b]) => b.lastShownAt - a.lastShownAt)
 		for (const [id, entry] of settled.slice(this.#options.keepSettled ?? 5)) {
@@ -285,8 +313,24 @@ export class FlowChatPool<H> {
 	async #pollRuns(): Promise<void> {
 		await Promise.all(
 			[...this.#running].map(async ([id, turn]) => {
-				const finished = await this.#options.isRunFinished(turn.jobId).catch(() => false)
-				if (finished && this.#running.get(id) === turn) this.#running.delete(id)
+				const finished = await this.#options
+					.isRunFinished(turn.jobId)
+					.then((done) => {
+						this.#pollFailures.delete(id)
+						return done
+					})
+					.catch(() => {
+						// A run whose job cannot be read — purged, refused, gone — would otherwise
+						// keep its row running and its poll going for the life of the page. After a
+						// few tries the row goes quiet; opening the conversation reads its rows.
+						const failures = (this.#pollFailures.get(id) ?? 0) + 1
+						this.#pollFailures.set(id, failures)
+						return failures >= POLL_GIVE_UP
+					})
+				if (finished && this.#running.get(id) === turn) {
+					this.#running.delete(id)
+					this.#pollFailures.delete(id)
+				}
 			})
 		)
 		if (this.#destroyed) return

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

@@ -540,7 +540,7 @@ export class FlowChatViewHost implements ChatViewHost {
 	 * does. This host outlives the panel, so a turn that refuses its message after the reader
 	 * has moved on has nowhere to put it back until then.
 	 */
-	#returned: Queue = emptyQueue()
+	#returned = $state<Queue>(emptyQueue())
 	#returnDraft(text: string, images: AttachedImage[] = [], blobs: AttachedBlob[] = []) {
 		if (this.#aiChatInput) {
 			this.#aiChatInput.prependText(text, images, [], blobs)
@@ -559,6 +559,15 @@ export class FlowChatViewHost implements ChatViewHost {
 	get queuedMessage(): string {
 		return this.#queue.text
 	}
+	/**
+	 * Something typed here has not been sent: waiting for the turn, or handed back by a turn
+	 * that refused it while no composer was mounted to take it. Either way this host is the
+	 * only place it exists, so nothing may release it.
+	 */
+	get hasUnsentDraft(): boolean {
+		const held = [this.#queue, this.#returned]
+		return held.some((q) => q.text !== '' || q.images.length > 0 || q.blobs.length > 0)
+	}
 	queuedContext = undefined
 	get queuedImages(): AttachedImage[] {
 		return this.#queue.images

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

@@ -475,9 +475,12 @@ describe('FlowChatViewHost', () => {
 		const host = new FlowChatViewHost(chat)
 		host.queueMessage('typed before leaving')
 		host.cancel()
+		// Held by the host alone until a composer takes it, so the pool must not release it.
+		expect(host.hasUnsentDraft).toBe(true)
 		const prependText = vi.fn()
 		host.setAiChatInput({ prependText } as any)
 		expect(prependText).toHaveBeenCalledWith('typed before leaving', [], [], [])
+		expect(host.hasUnsentDraft).toBe(false)
 		host.dispose()
 	})
 

frontend/src/lib/components/sessions/SessionPicker.svelte

@@ -802,7 +802,7 @@
 													{#if draft}
 														<PencilLine class="w-3 h-3 text-tertiary" aria-label="Unsent draft" />
 													{/if}
-													<UnreadCountBadge count={unread} />
+													<UnreadCountBadge count={unread} class="min-w-4 h-4 text-[10px]" />
 												</span>
 											{/if}
 										</MenuItem>
@@ -1069,7 +1069,7 @@
 										{#if draft}
 											<PencilLine class="w-3 h-3 text-tertiary" aria-label="Unsent draft" />
 										{/if}
-										<UnreadCountBadge count={unread} />
+										<UnreadCountBadge count={unread} class="min-w-4 h-4 text-[10px]" />
 									</span>
 								{/if}
 							</button>