diff --git a/chat-sdk/README.md b/chat-sdk/README.md index fb5c9752c6..899ca49acb 100644 --- a/chat-sdk/README.md +++ b/chat-sdk/README.md @@ -248,8 +248,10 @@ objects (the object for a single-file input). The name's extension is corrected file's media type for PNG, JPEG and PDF, because the worker reads the type off the key. A failed upload rejects `sendMessage` before any run starts, and `stop()` during the upload aborts it; both leave the transcript as it was. The workspace needs object -storage set up. With Enterprise advanced storage permissions, the user needs read and -write on `windmill_uploads/*`, which the default rules grant. +storage set up. With Enterprise advanced storage permissions, the user needs read, write +and delete on `windmill_uploads/*`, which the default rules grant. The upload goes through +`job_helpers`, so a restricted token needs `job_helpers:write`; a sandboxed raw app cannot +request that scope today, so attachments are not available there yet. ## History diff --git a/chat-sdk/src/api.ts b/chat-sdk/src/api.ts index 18e457c6c6..577abb02a3 100644 --- a/chat-sdk/src/api.ts +++ b/chat-sdk/src/api.ts @@ -214,6 +214,11 @@ export class WindmillChatApi { return (await res.json()) as { file_key: string } } + /** Removes an object from the workspace's object storage. */ + async deleteFile(fileKey: string): Promise { + await this.#request('job_helpers/delete_s3_file', { method: 'DELETE', query: { file_key: fileKey } }) + } + async deleteConversation(conversationId: string): Promise { await this.#request(`flow_conversations/delete/${encodeURIComponent(conversationId)}`, { method: 'DELETE' diff --git a/chat-sdk/src/attachments.ts b/chat-sdk/src/attachments.ts index 0c9319c35f..0eeca9f992 100644 --- a/chat-sdk/src/attachments.ts +++ b/chat-sdk/src/attachments.ts @@ -1,5 +1,6 @@ import type { WindmillChatApi } from './api' import type { ChatAttachment } from './types' +import { isAbortError } from './utils' /** * Where a chat's uploads live in the workspace's object storage. Under `windmill_uploads/` @@ -87,15 +88,39 @@ export async function uploadAttachments( signal?: AbortSignal ): Promise { const prefix = `${CHAT_UPLOADS_PREFIX}/${turnId}` - return Promise.all( - attachments.map(async (attachment, index) => { - const blob = attachmentBlob(attachment) - const filename = storedAttachmentName(attachment.name || `attachment-${index + 1}`, blob.type) - const { file_key } = await api.uploadFile(`${prefix}/${index}/${filename}`, blob, { - contentType: blob.type, - signal + // One failed upload aborts the rest, and whatever already landed is deleted: no run will + // read it, and a resend uploads under a fresh prefix. Best effort, so a delete that fails + // leaves that object behind rather than masking the upload error. + const batch = new AbortController() + const abortBatch = () => batch.abort() + signal?.addEventListener('abort', abortBatch, { once: true }) + try { + const results = await Promise.allSettled( + attachments.map(async (attachment, index) => { + try { + const blob = attachmentBlob(attachment) + const filename = storedAttachmentName( + attachment.name || `attachment-${index + 1}`, + blob.type + ) + const { file_key } = await api.uploadFile(`${prefix}/${index}/${filename}`, blob, { + contentType: blob.type, + signal: batch.signal + }) + return { s3: file_key, filename } + } catch (e) { + batch.abort() + throw e + } }) - return { s3: file_key, filename } - }) - ) + ) + const uploaded = results.flatMap((r) => (r.status === 'fulfilled' ? [r.value] : [])) + const reasons = results.flatMap((r) => (r.status === 'rejected' ? [r.reason] : [])) + if (reasons.length === 0) return uploaded + await Promise.all(uploaded.map((u) => api.deleteFile(u.s3).catch(() => {}))) + // The failure that started it, not the aborts it caused in the other uploads. + throw reasons.find((reason) => !isAbortError(reason)) ?? reasons[0] + } finally { + signal?.removeEventListener('abort', abortBatch) + } } diff --git a/chat-sdk/src/chat.ts b/chat-sdk/src/chat.ts index 54892ec086..85f15924b4 100644 --- a/chat-sdk/src/chat.ts +++ b/chat-sdk/src/chat.ts @@ -45,6 +45,8 @@ interface Turn { isNew: boolean /** The run was asked for. Before that, a failure or a stop withdraws the turn instead of failing it. */ started: boolean + /** Both `stop()` and the send's own rejection withdraw; only the first may. */ + withdrawn: boolean jobId?: string /** The flow job and its step jobs; a persisted answer carries one of them as `job_id`. */ jobIds?: Set @@ -126,6 +128,7 @@ class ChatImpl implements Chat { userMessageId: `pending-${randomId()}`, isNew, started: false, + withdrawn: false, streamedText: false } this.#turn = turn @@ -585,10 +588,15 @@ class ChatImpl implements Chat { * has already written the pending message to local history and kept the conversation listed. */ #withdrawTurn(turn: Turn): void { + if (turn.withdrawn) return + turn.withdrawn = true const id = turn.conversationId - const onScreen = this.#state.conversationId === id + // A turn started since (a resend right after Stop, or after switching back) owns the + // conversation and the status; this one only takes its own message away. + const newerTurn = this.#turn !== undefined && this.#turn !== turn + const onScreen = this.#state.conversationId === id && !newerTurn const withoutTurn = (messages: ChatMessage[]) => messages.filter((m) => m.id !== turn.userMessageId) - if (turn.isNew) { + if (turn.isNew && !(newerTurn && this.#turn?.conversationId === id)) { if (this.#state.history === 'local') this.#local.deleteConversation(id) this.#set({ conversations: this.#state.conversations.filter((c) => c.id !== id), @@ -601,6 +609,8 @@ class ChatImpl implements Chat { if (onScreen) { this.#set({ messages: withoutTurn(this.#state.messages), status: 'idle', error: undefined }) this.#persistLocal() + } else if (this.#state.conversationId === id) { + this.#set({ messages: withoutTurn(this.#state.messages) }) } else if (this.#state.history === 'local') { this.#local.saveMessages(id, withoutTurn(this.#local.getMessages(id))) } diff --git a/chat-sdk/test/attachments.test.ts b/chat-sdk/test/attachments.test.ts index a35c209b42..114b92d0e3 100644 --- a/chat-sdk/test/attachments.test.ts +++ b/chat-sdk/test/attachments.test.ts @@ -245,4 +245,78 @@ describe('sendMessage with attachments', () => { const reloaded = createChat({ ...options(fetch), storage }) expect((await reloaded.loadConversations()).map((c) => c.id)).not.toContain(opened) }) + + test('a failed upload deletes the files of the same batch that did land', async () => { + let first: (r: Response) => void = () => {} + const { fetch, calls } = fetchMock( + (c) => { + if (c.url.pathname !== UPLOAD_PATH) return undefined + const key = c.url.searchParams.get('file_key')! + // The first file lands after the second has already failed. + if (key.includes('/0/')) return new Promise((resolve) => (first = resolve)) + setTimeout(() => first(json({ file_key: keys()[0] })), 5) + return text('quota exceeded', 507) + }, + (c) => + c.method === 'DELETE' && c.url.pathname === '/api/w/ws/job_helpers/delete_s3_file' + ? json('deleted') + : undefined, + run, + answer + ) + const keys = () => uploads(calls).map((c) => c.url.searchParams.get('file_key')!) + const chat = createChat(options(fetch)) + await expect( + chat.sendMessage('read these', { + attachments: [ + { name: 'a.pdf', data: pdf }, + { name: 'b.png', data: png } + ], + attachmentsInput: { name: 'files', multiple: true } + }) + ).rejects.toThrow('quota exceeded') + const deletes = calls + .filter((c) => c.method === 'DELETE') + .map((c) => c.url.searchParams.get('file_key')) + expect(deletes).toEqual([keys()[0]]) + expect(runs(calls)).toHaveLength(0) + }) + + test('a send made right after stop() is not reset by the stopped upload', async () => { + let releaseRun: (r: Response) => void = () => {} + const { fetch, calls } = fetchMock( + (c) => + c.url.pathname === UPLOAD_PATH + ? new Promise((_, reject) => + c.signal!.addEventListener('abort', () => reject(abortError())) + ) + : undefined, + (c) => + c.method === 'POST' && c.url.pathname === `/api/w/ws/jobs/run/f/${FLOW}` + ? new Promise((resolve) => (releaseRun = resolve)) + : undefined, + answer + ) + const chat = createChat(options(fetch)) + // An existing conversation, so the stopped turn and the next one share it. + const first = chat.sendMessage('first') + await new Promise((r) => setTimeout(r, 0)) + releaseRun(text('job-1')) + await first + const stopped = chat.sendMessage('with a file', { + attachments: [{ name: 'a.pdf', data: pdf }], + attachmentsInput: { name: 'files', multiple: true } + }) + await new Promise((r) => setTimeout(r, 0)) + void chat.stop() + const next = chat.sendMessage('right after') + await expect(stopped).rejects.toMatchObject({ name: 'AbortError' }) + await new Promise((r) => setTimeout(r, 0)) + expect(chat.getState().status).toBe('submitted') + expect(chat.getState().messages.map((m) => m.content)).toContain('right after') + expect(chat.getState().messages.map((m) => m.content)).not.toContain('with a file') + releaseRun(text('job-1')) + await next + expect(runs(calls)).toHaveLength(2) + }) }) diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts index 96049e6d28..c01f2991c9 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts @@ -171,6 +171,8 @@ export class FlowChatViewHost implements ChatViewHost { if (previous.conversationId !== state.conversationId) { // A conversation opens at its end, whatever the reader was doing in the last one. this.#automaticScroll = true + // A conversation reopened later comes back from the server under other message ids. + this.#sentAttachments.clear() // The queue was typed into the conversation that just went away; a message sent // after the switch would ride out of the wrong one, so it goes back to the composer. this.dequeueMessage() @@ -304,6 +306,14 @@ export class FlowChatViewHost implements ChatViewHost { }) this.#turnDone = turn await turn + // The files are kept only for a turn that failed, the one Retry is offered on: a base64 + // payload per sent file would otherwise pile up for as long as the panel lives. + if (sentId) { + const index = this.#state.messages.findIndex((m) => m.id === sentId) + if (index === -1 || !turnFailed(this.#state.messages, index)) { + this.#sentAttachments.delete(sentId) + } + } return true } /** Settles when the chat has released the last turn this host started. */ diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts index 741a076e77..318f2bdd49 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts @@ -367,13 +367,13 @@ describe('FlowChatViewHost', () => { chat.sendMessage.mockImplementationOnce(async () => {}) const host = new FlowChatViewHost(chat, { attachmentsTarget: () => listInput }) const sending = host.sendRequest({ instructions: 'read', images: [image], blobs: [pdf] }) - await sending set({ messages: [ message({ id: 'u1', role: 'user', content: 'read' }), message({ role: 'assistant', content: 'boom', success: false }) ] }) + await sending host.retryRequest(0) await new Promise((resolve) => setTimeout(resolve, 0)) const [text, options] = chat.sendMessage.mock.calls[1] as any @@ -382,6 +382,28 @@ describe('FlowChatViewHost', () => { host.dispose() }) + it('lets go of the files of a turn that succeeded', async () => { + const { chat, set } = fakeChat(idleState({ messages: [] })) + chat.sendMessage.mockImplementationOnce(async () => { + set({ messages: [message({ id: 'u1', role: 'user', content: 'read', pending: true })] }) + }) + chat.sendMessage.mockImplementationOnce(async () => {}) + const host = new FlowChatViewHost(chat, { attachmentsTarget: () => listInput }) + const sending = host.sendRequest({ instructions: 'read', blobs: [pdf] }) + set({ + messages: [ + message({ id: 'u1', role: 'user', content: 'read' }), + message({ role: 'assistant', content: 'done' }) + ] + }) + await sending + host.retryRequest(0) + await new Promise((resolve) => setTimeout(resolve, 0)) + const [, options] = chat.sendMessage.mock.calls[1] as any + expect(options.attachments).toEqual([]) + host.dispose() + }) + it('queues attachments with the text and sends them together', async () => { const { chat, set } = fakeChat(idleState({ status: 'streaming' })) const host = new FlowChatViewHost(chat, { attachmentsTarget: () => listInput })