fix: list a chat conversation only once its run starts and require files where the flow does

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

chat-sdk/src/attachments.ts

@@ -1,6 +1,6 @@
 import type { WindmillChatApi } from './api'
 import type { ChatAttachment } from './types'
-import { isAbortError } from './utils'
+import { abortError, isAbortError } from './utils'
 
 /**
  * Where a chat's uploads live in the workspace's object storage. Under `windmill_uploads/`
@@ -91,6 +91,7 @@ export async function uploadAttachments(
   // 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.
+  if (signal?.aborted) throw abortError()
   const batch = new AbortController()
   const abortBatch = () => batch.abort()
   signal?.addEventListener('abort', abortBatch, { once: true })

chat-sdk/src/chat.ts

@@ -140,7 +140,6 @@ class ChatImpl implements Chat {
     const touched = { ...conversation, updatedAt: timestamp }
     this.#set({
       conversationId,
-      conversations: [touched, ...this.#state.conversations.filter((c) => c.id !== conversationId)],
       messages: [
         ...this.#state.messages,
         { id: turn.userMessageId, role: 'user', content, success: true, createdAt: timestamp, pending: true }
@@ -148,7 +147,6 @@ class ChatImpl implements Chat {
       status: 'submitted',
       error: undefined
     })
-    this.#rememberConversation()
 
     try {
       const args: Record<string, unknown> = { ...this.#config.inputs, ...options.inputs, user_message: content }
@@ -159,6 +157,12 @@ class ChatImpl implements Chat {
         args[attachmentsInput.name] = attachmentsInput.multiple ? uploaded : uploaded[0]
       }
       turn.started = true
+      // Listed only once the run is asked for: a send that never runs (an upload that failed
+      // or was stopped) then has no conversation entry to take back.
+      this.#set({
+        conversations: [touched, ...this.#state.conversations.filter((c) => c.id !== conversationId)]
+      })
+      this.#rememberConversation()
       const context = { memoryId: conversationId, conversationId, signal: turn.controller.signal }
       turn.jobId = this.#config.run
         ? await this.#config.run(args, context)
@@ -583,36 +587,32 @@ 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. Also after a switch away mid-upload, which
-   * has already written the pending message to local history and kept the conversation listed.
+   * Take back the user message of a turn that never ran. A conversation it would have opened
+   * was never listed (see `sendMessage`), so only the message goes, and, while it is the turn
+   * on screen, the busy status. A switch away mid-upload has already written the message to
+   * local history, so it is removed there too.
    */
   #withdrawTurn(turn: Turn): void {
     if (turn.withdrawn) return
     turn.withdrawn = true
     const id = turn.conversationId
-    // 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 && !(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),
-        ...(onScreen
-          ? { conversationId: undefined, messages: withoutTurn(this.#state.messages), status: 'idle', error: undefined }
-          : {})
-      })
-      return
-    }
-    if (onScreen) {
-      this.#set({ messages: withoutTurn(this.#state.messages), status: 'idle', error: undefined })
+    if (this.#state.conversationId === id) {
+      const messages = withoutTurn(this.#state.messages)
+      // A turn started since, such as a resend right after Stop, owns the status.
+      const newerTurn = this.#turn !== undefined && this.#turn !== turn
+      if (newerTurn) {
+        this.#set({ messages })
+      } else {
+        const unopened = turn.isNew && messages.length === 0
+        this.#set({ messages, status: 'idle', error: undefined, ...(unopened ? { conversationId: 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)))
+    }
+    if (this.#state.history === 'local' && this.#state.conversationId !== id) {
+      const stored = withoutTurn(this.#local.getMessages(id))
+      if (stored.length > 0) this.#local.saveMessages(id, stored)
+      else if (!this.#state.conversations.some((c) => c.id === id)) this.#local.deleteConversation(id)
     }
   }
 

chat-sdk/test/attachments.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, test } from 'bun:test'
-import { storedAttachmentName } from '../src/attachments'
+import { WindmillChatApi } from '../src/api'
+import { storedAttachmentName, uploadAttachments } from '../src/attachments'
 import { createChat } from '../src/chat'
 import type { ChatOptions } from '../src/types'
 import { abortError } from '../src/utils'
@@ -319,4 +320,37 @@ describe('sendMessage with attachments', () => {
     await next
     expect(runs(calls)).toHaveLength(2)
   })
+
+  test('a conversation is listed only once its run starts', async () => {
+    let failUpload: (r: Response) => void = () => {}
+    const { fetch } = fetchMock(
+      (c) =>
+        c.url.pathname === UPLOAD_PATH
+          ? new Promise<Response>((resolve) => (failUpload = resolve))
+          : undefined,
+      run,
+      answer
+    )
+    const chat = createChat(options(fetch))
+    const sending = chat.sendMessage('read this', {
+      attachments: [{ name: 'a.pdf', data: pdf }],
+      attachmentsInput: { name: 'files', multiple: true }
+    })
+    await new Promise((r) => setTimeout(r, 0))
+    expect(chat.getState()).toMatchObject({ status: 'submitted', conversations: [] })
+    failUpload(text('boom', 500))
+    await expect(sending).rejects.toThrow('boom')
+    expect(chat.getState().conversations).toEqual([])
+  })
+
+  test('an already aborted signal uploads nothing', async () => {
+    const { fetch, calls } = fetchMock(upload)
+    const api = new WindmillChatApi({ baseUrl: BASE, workspace: 'ws', token: 'tok', fetch })
+    const controller = new AbortController()
+    controller.abort()
+    await expect(
+      uploadAttachments(api, [{ name: 'a.pdf', data: pdf }], 'turn', controller.signal)
+    ).rejects.toMatchObject({ name: 'AbortError' })
+    expect(calls).toHaveLength(0)
+  })
 })

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

@@ -54,13 +54,17 @@
 	})
 
 	// The composer's attachments feed this input, and the paperclip is its whole editor.
-	const attachmentsTarget = $derived(
-		attachmentsTargetFor(
+	const attachmentsTarget = $derived.by(() => {
+		const target = attachmentsTargetFor(
 			resolveAgentChatInputs(flowModules, additionalInputsSchema).find(
 				(input) => input.key === PER_TURN_AGENT_CHAT_INPUT_KEY
 			)
 		)
-	)
+		const required: unknown = additionalInputsSchema?.required
+		return target && Array.isArray(required) && required.includes(target.name)
+			? { ...target, required: true }
+			: target
+	})
 	// Uploading needs the workspace's object storage; without one the `+` is drawn disabled
 	// saying so, since the modal could not upload either.
 	const workspaceStorage = useWorkspaceStorageConfigured(() => workspace)

frontend/src/lib/components/flows/conversations/agentAttachmentInput.ts

@@ -78,8 +78,9 @@ function holdsS3File(property: Record<string, any> | undefined): boolean {
 	)
 }
 
-/** The flow input the composer's attachments feed, and whether it holds a list. */
-export type AttachmentsTarget = { name: string; multiple: boolean }
+/** The flow input the composer's attachments feed, whether it holds a list, and whether the
+ * flow requires it (so a message without a file cannot run). */
+export type AttachmentsTarget = { name: string; multiple: boolean; required?: boolean }
 
 /**
  * Where the composer's attachments go, or nothing when there is nowhere they fit.

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

@@ -243,6 +243,12 @@ export class FlowChatViewHost implements ChatViewHost {
 			return false
 		}
 		const target = this.#options.attachmentsTarget?.()
+		// The inputs modal does not ask for this input, so a required one is enforced here.
+		if (target?.required && images.length === 0 && blobs.length === 0) {
+			sendUserToast('This chat needs a file with each message. Attach one to send.', true)
+			this.#aiChatInput?.prependText(text, images, [], blobs)
+			return false
+		}
 		const inputs = { ...(this.#options.additionalInputs?.() ?? {}) }
 		// The attachments are this input's only editor: a value stored for it in the inputs
 		// modal would otherwise ride along on every message.

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

@@ -404,6 +404,21 @@ describe('FlowChatViewHost', () => {
 		host.dispose()
 	})
 
+	it('refuses a message without a file when the flow requires one', async () => {
+		const { chat } = fakeChat()
+		const host = new FlowChatViewHost(chat, {
+			attachmentsTarget: () => ({ ...listInput, required: true })
+		})
+		const prependText = vi.fn()
+		host.setAiChatInput({ prependText } as any)
+		expect(await host.sendRequest({ instructions: 'no file' })).toBe(false)
+		expect(chat.sendMessage).not.toHaveBeenCalled()
+		expect(prependText).toHaveBeenCalledWith('no file', [], [], [])
+		expect(await host.sendRequest({ instructions: 'with file', blobs: [pdf] })).toBe(true)
+		expect(chat.sendMessage).toHaveBeenCalledTimes(1)
+		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 })