diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index b08a048d6b..f69cd1389d 100644 --- a/backend/windmill-api/openapi.yaml +++ b/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: diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 3dccea8dfd..de15f937c4 100644 --- a/chat-sdk/src/chat.ts +++ b/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. diff --git a/chat-sdk/src/follow.ts b/chat-sdk/src/follow.ts index 14511f1e2f..4578da5f43 100644 --- a/chat-sdk/src/follow.ts +++ b/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) diff --git a/chat-sdk/test/chat.test.ts b/chat-sdk/test/chat.test.ts index dfa437329c..c917dd82c0 100644 --- a/chat-sdk/test/chat.test.ts +++ b/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) => diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index 54048bd83c..cacdf0b2f4 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/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)) diff --git a/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts b/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts index 6b3c1c8c5f..d3e6cbdc10 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatPool.test.ts @@ -60,7 +60,7 @@ function pool( } = {} ) { const chats: ReturnType[] = [] - 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({ diff --git a/frontend/src/lib/components/flows/conversations/flowChatPool.ts b/frontend/src/lib/components/flows/conversations/flowChatPool.ts index 8345eb4db5..738c940ce7 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatPool.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatPool.ts @@ -57,6 +57,8 @@ export interface FlowChatPoolOptions { 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 /** Settled conversations kept in memory beside the shown and the busy ones. */ @@ -93,6 +95,8 @@ export class FlowChatPool { readonly #running = new Map() /** Failed reads in a row, per conversation, for a run this pool has no chat for. */ readonly #pollFailures = new Map() + /** Chats on their way out, kept until the send that withdrew them has settled. */ + readonly #retiring = new Set>() readonly #unread = new Map() readonly #listeners = new Set<(state: FlowChatPoolState) => void>() #selectedId: string | undefined @@ -198,7 +202,9 @@ export class FlowChatPool { 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 { 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() } diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts index 772bc2b978..4374bc6fce 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts +++ b/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