starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: store chat uploads under windmill_uploads and withdraw refused sends cleanly

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guilhem Lemouel
2026-09-16 20:50:57 +02:00
co-authored by Claude Opus 5
parent 43591fe0e9
commit 5cd8488df7
10 changed files with 105 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chat-sdk/README.md
+3 -2
View File
@@ -243,12 +243,13 @@ await chat.sendMessage('What does this contract say?', {
```
Each file is uploaded to the workspace's object storage under
`windmill_chat_uploads/<turn>/<index>/<name>` and handed to the input as `{ s3, filename }`
`windmill_uploads/chat/<turn>/<index>/<name>` and handed to the input as `{ s3, filename }`
objects (the object for a single-file input). The name's extension is corrected to the
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, and the token needs to be allowed to upload.
storage set up. With Enterprise advanced storage permissions, the user needs read and
write on `windmill_uploads/*`, which the default rules grant.
## History
chat-sdk/src/attachments.ts
+7 -2
View File
@@ -1,8 +1,13 @@
import type { WindmillChatApi } from './api'
import type { ChatAttachment } from './types'
/** Where a chat's uploads live in the workspace's object storage. */
export const CHAT_UPLOADS_PREFIX = 'windmill_chat_uploads'
/**
* Where a chat's uploads live in the workspace's object storage. Under `windmill_uploads/`
* because the default Enterprise storage permissions grant every user write and read there
* and deny any other top-level prefix: a key outside it is refused for non-admins, both on
* upload and when the agent's job reads the file back.
*/
export const CHAT_UPLOADS_PREFIX = 'windmill_uploads/chat'
/** What an AI agent step reads out of `user_attachments`. */
export interface UploadedAttachment {
chat-sdk/src/chat.ts
+16 -12
View File
@@ -577,25 +577,29 @@ class ChatImpl implements Chat {
/**
* Undo what `sendMessage` showed for a turn that never ran: its user message, and the
* conversation it opened when there was none. Only while that conversation is still the
* one on screen; a switch meanwhile has already left it behind.
* conversation it opened when there was none. Also after a switch away mid-upload, which
* has already written the pending message to local history and kept the conversation listed.
*/
#withdrawTurn(turn: Turn): void {
if (this.#state.conversationId !== turn.conversationId) return
const messages = this.#state.messages.filter((m) => m.id !== turn.userMessageId)
const id = turn.conversationId
const onScreen = this.#state.conversationId === id
const withoutTurn = (messages: ChatMessage[]) => messages.filter((m) => m.id !== turn.userMessageId)
if (turn.isNew) {
if (this.#state.history === 'local') this.#local.deleteConversation(turn.conversationId)
if (this.#state.history === 'local') this.#local.deleteConversation(id)
this.#set({
conversationId: undefined,
conversations: this.#state.conversations.filter((c) => c.id !== turn.conversationId),
messages,
status: 'idle',
error: undefined
conversations: this.#state.conversations.filter((c) => c.id !== id),
...(onScreen
? { conversationId: undefined, messages: withoutTurn(this.#state.messages), status: 'idle', error: undefined }
: {})
})
return
}
this.#set({ messages, status: 'idle', error: undefined })
this.#persistLocal()
if (onScreen) {
this.#set({ messages: withoutTurn(this.#state.messages), status: 'idle', error: undefined })
this.#persistLocal()
} else if (this.#state.history === 'local') {
this.#local.saveMessages(id, withoutTurn(this.#local.getMessages(id)))
}
}
#failTurn(turn: Turn, e: unknown): void {
chat-sdk/test/attachments.test.ts
+30 -2
View File
@@ -88,8 +88,8 @@ describe('sendMessage with attachments', () => {
const keys = uploads(calls).map((c) => c.url.searchParams.get('file_key')!)
expect(keys).toHaveLength(3)
const prefix = keys[0].split('/').slice(0, 2).join('/')
expect(prefix).toMatch(/^windmill_chat_uploads\/[0-9a-f-]{36}$/)
const prefix = keys[0].split('/').slice(0, 3).join('/')
expect(prefix).toMatch(/^windmill_uploads\/chat\/[0-9a-f-]{36}$/)
expect(keys).toEqual([
`${prefix}/0/photo.png`,
`${prefix}/1/contract.pdf`,
@@ -201,4 +201,32 @@ describe('sendMessage with attachments', () => {
conversationId: undefined
})
})
test('a switch away mid-upload leaves no conversation behind', async () => {
const storage = memoryStorage()
const { fetch, calls } = fetchMock(
(c) =>
c.url.pathname === UPLOAD_PATH
? new Promise((_, reject) =>
c.signal!.addEventListener('abort', () => reject(abortError()))
)
: undefined,
run,
answer
)
const chat = createChat({ ...options(fetch), storage })
const sending = chat.sendMessage('never runs', {
attachments: [{ name: 'contract.pdf', data: pdf }],
attachmentsInput: { name: 'files', multiple: true }
})
await new Promise((r) => setTimeout(r, 0))
const opened = chat.getState().conversationId!
chat.newConversation()
await expect(sending).rejects.toMatchObject({ name: 'AbortError' })
expect(runs(calls)).toHaveLength(0)
expect(chat.getState().conversations.map((c) => c.id)).not.toContain(opened)
const reloaded = createChat({ ...options(fetch), storage })
expect((await reloaded.loadConversations()).map((c) => c.id)).not.toContain(opened)
})
})
frontend/src/lib/components/copilot/chat/AIChatInput.svelte
+1 -3
View File
@@ -493,9 +493,7 @@
*/
export async function addNonImageFiles(files: File[]) {
if (files.length === 0) return
// Attaching can be off despite the chat taking attachments — no object storage to
// upload to, say. The `+` renders disabled with the reason; a drop and a paste reach
// here instead, and would otherwise become a chip that only fails once sent.
// Same reason as in addImages.
const unavailable = chatHost.attachmentsUnavailableReason
if (unavailable) {
sendUserToast(unavailable, true)
frontend/src/lib/components/copilot/chat/blobUtils.test.ts
+1 -24
View File
@@ -1,33 +1,10 @@
import { describe, expect, it } from 'vitest'
import { dataUrlToBlob, matchesAccept } from './blobUtils'
import { matchesAccept } from './blobUtils'
function file(name: string, type: string): File {
return new File(['x'], name, { type })
}
describe('dataUrlToBlob', () => {
// The bytes are re-uploaded verbatim, so a decode that drops or shifts one is a
// corrupted file the reader only discovers downstream.
it('decodes base64 back to the exact bytes', async () => {
const bytes = new Uint8Array([0x00, 0xff, 0x10, 0x89, 0x50])
const b64 = btoa(String.fromCharCode(...bytes))
const blob = dataUrlToBlob(`data:application/pdf;base64,${b64}`)
expect(blob.type).toBe('application/pdf')
expect(new Uint8Array(await blob.arrayBuffer())).toEqual(bytes)
})
it('percent-decodes a url that is not base64', async () => {
const blob = dataUrlToBlob('data:text/plain,hello%20world')
expect(blob.type).toBe('text/plain')
expect(await blob.text()).toBe('hello world')
})
it('falls back to a media type when the url names none', async () => {
expect(dataUrlToBlob('data:;base64,QQ==').type).toBe('application/octet-stream')
expect(dataUrlToBlob('data:;base64,QQ==', 'image/png').type).toBe('image/png')
})
})
describe('matchesAccept', () => {
it('matches an extension, a type wildcard and an exact media type', () => {
expect(matchesAccept(file('report.PDF', ''), '.pdf')).toBe(true)
frontend/src/lib/components/copilot/chat/blobUtils.ts
-16
View File
@@ -61,19 +61,3 @@ export async function fileToAttachedBlob(file: File): Promise<AttachedBlob> {
size: file.size
}
}
/** The bytes behind a `data:` URL, for a host that has to re-upload them. */
export function dataUrlToBlob(dataUrl: string, fallbackType = 'application/octet-stream'): Blob {
const comma = dataUrl.indexOf(',')
const header = dataUrl.slice(5, comma)
const isBase64 = header.endsWith(';base64')
const mediaType = (isBase64 ? header.slice(0, -';base64'.length) : header) || fallbackType
const payload = dataUrl.slice(comma + 1)
if (!isBase64) {
return new Blob([decodeURIComponent(payload)], { type: mediaType })
}
const binary = atob(payload)
const bytes = new Uint8Array(binary.length)
for (let i = 0; i < binary.length; i++) bytes[i] = binary.charCodeAt(i)
return new Blob([bytes], { type: mediaType })
}
frontend/src/lib/components/flows/conversations/agentAttachmentInput.ts
+7 -20
View File
@@ -33,35 +33,23 @@ export function agentSteps(modules: FlowModule[] | undefined): FlowModule[] {
}
/**
* AI agent inputs the chat composer can drive.
*
* The composer never edits the flow: an agent field is reachable only when the author
* wired a flow input to it, so what the composer writes is a run input like any other. That
* also keeps it working on the deployed chat, where the reader has no write access, and
* on a step linked to an `ai_agent` resource, where every field but `user_message` /
* `user_attachments` comes from the resource and is not overridable at all.
*
* Only what the person chatting legitimately owns turn to turn belongs here, which today
* is the files they attach and nothing else. `system_prompt`, `temperature` and
* `max_completion_tokens` shape how the agent behaves for everyone who runs the flow —
* surfacing them per conversation invites tuning the flow from the chat instead of
* fixing it in the editor. They stay flow settings, reachable through Configure inputs
* when the author deliberately exposes them. `max_iterations` is absent for the same
* reason, and because it caps the tool-use loop rather than a single generation.
* AI agent inputs the chat composer drives, through the flow input the author wired to each:
* the composer writes run inputs, never the flow. Only what the person chatting owns turn to
* turn belongs here. `system_prompt`, `temperature` and the like shape the agent for everyone
* who runs the flow, so they stay in the Configure-inputs modal.
*/
export const AGENT_CHAT_INPUT_KEYS = ['user_attachments'] as const
export type AgentChatInputKey = (typeof AGENT_CHAT_INPUT_KEYS)[number]
/** The key that rides one message rather than the conversation, which the composer
* clears on send. A later key of the other sort would not be this one. */
/** The key that rides one message rather than the conversation, cleared on send. */
export const PER_TURN_AGENT_CHAT_INPUT_KEY: AgentChatInputKey = 'user_attachments'
export type AgentChatInput = {
/** Flow input property feeding the agent field. */
name: string
key: AgentChatInputKey
/** The flow input's own schema entry — the chip renders it with the same editor the modal would. */
/** The flow input's own schema entry. */
property: Record<string, any>
}
@@ -118,8 +106,7 @@ export function isEmptyAgentChatInputValue(value: any): boolean {
/**
* The flow inputs that an AI agent step reads directly into one of its chat-relevant
* fields. Several agents may resolve to the same flow input; it is one chip either way,
* and one that stays unambiguous however many agents read it.
* fields. Several agents may read the same flow input; it is promoted once.
*/
export function resolveAgentChatInputs(
modules: FlowModule[] | undefined,
frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts
+10 -27
View File
@@ -8,7 +8,7 @@ import type { AIAutonomyMode } from '$lib/components/copilot/chat/AIChatManager.
import { isPlanCardTool } from '$lib/components/copilot/chat/planMode'
import { AttachedFilesStore } from '$lib/components/copilot/chat/files/attachedFiles.svelte'
import { SessionArtifactsStore } from '$lib/components/copilot/chat/artifacts/artifactsState.svelte'
import { dataUrlToBlob, type AttachedBlob } from '$lib/components/copilot/chat/blobUtils'
import type { AttachedBlob } from '$lib/components/copilot/chat/blobUtils'
import type { AttachedImage } from '$lib/components/copilot/chat/imageUtils'
import type { AttachedTextFile } from '$lib/components/copilot/chat/textFileUtils'
import { sendUserToast } from '$lib/toast'
@@ -242,10 +242,8 @@ export class FlowChatViewHost implements ChatViewHost {
}
const target = this.#options.attachmentsTarget?.()
const inputs = { ...(this.#options.additionalInputs?.() ?? {}) }
// Where the paperclip is the input's editor, the stored settings have no say over it:
// a value saved while the modal owned it — before this workspace had object storage —
// would otherwise ride along on every later message. The chat sets it from the
// attachments, or not at all.
// The attachments are this input's only editor: a value stored for it in the inputs
// modal would otherwise ride along on every message.
if (target) delete inputs[target.name]
// The per-turn cap again, at the place the truncation would happen: the composer
// enforces it as files are attached, but a queue built over several turns arrives
@@ -263,12 +261,11 @@ export class FlowChatViewHost implements ChatViewHost {
true
)
}
// The bytes go up as picked (or as the composer re-encoded them, for an image): the
// chat uploads them to the workspace's object storage and names them by their type.
const attachments: ChatAttachment[] = target
? [...images, ...blobs].map((attachment, index) => ({
name: attachment.name ?? `attachment-${index + 1}`,
data: dataUrlToBlob(attachment.dataUrl, attachment.mediaType)
data: attachment.dataUrl,
mediaType: attachment.mediaType
}))
: []
this.#automaticScroll = true
@@ -291,30 +288,16 @@ export class FlowChatViewHost implements ChatViewHost {
true
)
}
this.#restoreToComposer(text, images, blobs)
// What was queued behind it comes back too, after it: the chat publishes `idle`
// when it withdraws the turn, and a queue left in place would be flushed as if
// the turn had run.
this.dequeueMessage()
this.#aiChatInput?.prependText(text, images, [], blobs)
})
this.#turnDone = turn
await turn
return true
}
/**
* Give a spent draft back after a send that did not run. Back to the front of the queue
* whenever one is waiting: anything queued was typed later, and the next settled turn
* sends the queue whole — so putting the older draft in the composer would run the two
* out of order.
*/
#restoreToComposer(text: string, images: AttachedImage[], blobs: AttachedBlob[]) {
const queue = this.#queue
if (!queue.text && queue.images.length === 0 && queue.blobs.length === 0) {
this.#aiChatInput?.prependText(text, images, [], blobs)
return
}
this.#queue = {
text: queue.text ? `${text}\n${queue.text}` : text,
images: [...images, ...queue.images],
blobs: [...blobs, ...queue.blobs]
}
}
/** Settles when the chat has released the last turn this host started. */
#turnDone: Promise<unknown> = Promise.resolve()
cancel = () => {
frontend/src/lib/components/flows/conversations/flowChatViewHost.test.ts
+30 -7
View File
@@ -290,7 +290,7 @@ describe('FlowChatViewHost', () => {
expect(single.attachmentsUnavailableReason).toBe('no storage')
})
it('hands the attachments to the chat as bytes, and drops a stored value for their input', async () => {
it('hands the attachments to the chat, and drops a stored value for their input', async () => {
const { chat } = fakeChat()
const host = new FlowChatViewHost(chat, {
additionalInputs: () => ({ tone: 'brief', files: [{ s3: 'stale' }] }),
@@ -300,13 +300,10 @@ describe('FlowChatViewHost', () => {
const [, options] = chat.sendMessage.mock.calls[0] as any
expect(options.inputs).toEqual({ tone: 'brief' })
expect(options.attachmentsInput).toBe(listInput)
expect(options.attachments.map((a: any) => [a.name, a.data.type])).toEqual([
['shot.webp', 'image/png'],
['contract.pdf', 'application/pdf']
expect(options.attachments).toEqual([
{ name: 'shot.webp', data: PNG, mediaType: 'image/png' },
{ name: 'contract.pdf', data: pdf.dataUrl, mediaType: 'application/pdf' }
])
expect(new Uint8Array(await options.attachments[0].data.arrayBuffer())).toEqual(
new Uint8Array([0x89, 0x50, 0x4e, 0x47])
)
})
// The composer caps as files are attached, but a queue built over several turns
@@ -335,6 +332,32 @@ describe('FlowChatViewHost', () => {
host.dispose()
})
// The chat settles a withdrawn turn as `idle`, which reads like a turn that ran; a queue
// left waiting would then go out with the failed draft merged in front of it.
it('does not send what was queued behind an upload that failed', async () => {
const { chat, set } = fakeChat(
idleState({ messages: [message({ role: 'user', content: 'earlier' })] })
)
let refuse = (_e: Error) => {}
chat.sendMessage.mockImplementationOnce(
() => new Promise<void>((_, reject) => (refuse = reject))
)
const host = new FlowChatViewHost(chat, { attachmentsTarget: () => listInput })
const prependText = vi.fn()
host.setAiChatInput({ prependText } as any)
void host.sendRequest({ instructions: 'A', blobs: [pdf] })
set({ status: 'submitted' })
host.queueMessage('B')
set({ status: 'idle' })
refuse(new Error('upload failed (500)'))
await new Promise((resolve) => setTimeout(resolve, 0))
expect(chat.sendMessage).toHaveBeenCalledTimes(1)
expect(host.queuedMessage).toBe('')
expect(prependText.mock.calls.map((c) => c[0])).toEqual(['B', 'A'])
expect(prependText).toHaveBeenLastCalledWith('A', [], [], [pdf])
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 })