From 30f202ce511bcf9fd1d7f989c5ffd8b04db14245 Mon Sep 17 00:00:00 2001 From: Guilhem Lemouel Date: Fri, 18 Sep 2026 14:39:47 +0200 Subject: [PATCH] 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) --- backend/windmill-api/openapi.yaml | 20 +++++++ .../windmill-common/src/flow_conversations.rs | 6 ++ .../flows/conversations/FlowChat.svelte | 2 +- .../flows/conversations/flowChatPool.test.ts | 49 +++++++++++++++- .../flows/conversations/flowChatPool.ts | 56 +++++++++++++++++-- .../conversations/flowChatViewHost.svelte.ts | 11 +++- .../conversations/flowChatViewHost.test.ts | 3 + .../components/sessions/SessionPicker.svelte | 4 +- 8 files changed, 138 insertions(+), 13 deletions(-) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index fee3e62782..b08a048d6b 100644 --- a/backend/windmill-api/openapi.yaml +++ b/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: diff --git a/backend/windmill-common/src/flow_conversations.rs b/backend/windmill-common/src/flow_conversations.rs index 646d2437e9..582ea3bd36 100644 --- a/backend/windmill-common/src/flow_conversations.rs +++ b/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], diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index 22eee594b7..54048bd83c 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/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 }) diff --git a/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts b/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts index 9a3b86dd1f..6b3c1c8c5f 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts @@ -53,7 +53,11 @@ function conversation(id: string, extra: Partial = {}): Conversati } function pool( - options: { keepSettled?: number; isRunFinished?: (jobId: string) => Promise } = {} + options: { + keepSettled?: number + isRunFinished?: (jobId: string) => Promise + hasUnsentDraft?: (host: { chat: Chat }) => boolean + } = {} ) { const chats: ReturnType[] = [] 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') diff --git a/frontend/src/lib/components/flows/conversations/flowChatPool.ts b/frontend/src/lib/components/flows/conversations/flowChatPool.ts index 445b3018bc..8345eb4db5 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatPool.ts +++ b/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 { 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 { #draft: Entry | undefined /** Turns running in conversations this pool is not following, as the list reported them. */ readonly #running = new Map() + /** Failed reads in a row, per conversation, for a run this pool has no chat for. */ + readonly #pollFailures = new Map() readonly #unread = new Map() readonly #listeners = new Set<(state: FlowChatPoolState) => void>() #selectedId: string | undefined @@ -226,7 +232,14 @@ export class FlowChatPool { 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 { this.#publish() } + /** Takes an entry back out of the list of conversations, as the chat that starts one. */ + #undoNewConversation(entry: Entry): 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 { ([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 { async #pollRuns(): Promise { 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 diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts index 7f3a07b383..772bc2b978 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts +++ b/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(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 diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts index 1cd44eaff9..03252b84d2 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts +++ b/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() }) diff --git a/frontend/src/lib/components/sessions/SessionPicker.svelte b/frontend/src/lib/components/sessions/SessionPicker.svelte index c634129f74..caac58019a 100644 --- a/frontend/src/lib/components/sessions/SessionPicker.svelte +++ b/frontend/src/lib/components/sessions/SessionPicker.svelte @@ -802,7 +802,7 @@ {#if draft} {/if} - + {/if} @@ -1069,7 +1069,7 @@ {#if draft} {/if} - + {/if}