fix: discard uploads when a send is stopped as its attachments are announced

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

chat-sdk/src/attachments.ts

@@ -70,6 +70,11 @@ function dataUrlToBlob(dataUrl: string, fallbackType: string): Blob {
  * The key's turn prefix and per-file index keep two files with the same name, in this turn or
  * an earlier one, from overwriting each other; the name stays the last segment.
  */
+/** Delete uploads no run will read. Best effort: a delete that fails leaves that object behind. */
+export async function discardUploads(api: WindmillChatApi, uploaded: UploadedAttachment[]): Promise<void> {
+  await Promise.all(uploaded.map((u) => api.deleteFile(u.s3).catch(() => {})))
+}
+
 export async function uploadAttachments(
   api: WindmillChatApi,
   attachments: AttachmentUpload[],
@@ -108,7 +113,7 @@ export async function uploadAttachments(
     const reasons = results.flatMap((r) => (r.status === 'rejected' ? [r.reason] : []))
     // A stop that lands once every upload has answered still withdraws the batch.
     if (reasons.length === 0 && !signal?.aborted) return uploaded
-    await Promise.all(uploaded.map((u) => api.deleteFile(u.s3).catch(() => {})))
+    await discardUploads(api, uploaded)
     // The failure that started it, not the aborts it caused in the other uploads.
     throw reasons.find((reason) => !isAbortError(reason)) ?? reasons[0] ?? abortError()
   } finally {

chat-sdk/src/chat.ts

@@ -7,7 +7,7 @@ import {
 import { resolveConfig, type ResolvedConfig } from './config'
 import { followJob } from './follow'
 import { createLocalHistory, type LocalHistory } from './history'
-import { uploadAttachments } from './attachments'
+import { discardUploads, uploadAttachments } from './attachments'
 import type { AgentStreamEvent } from './stream'
 import type {
   Chat,
@@ -167,6 +167,11 @@ class ChatImpl implements Chat {
             messages: this.#state.messages.map((m) => (m.id === turn.userMessageId ? { ...m, attachments: carried } : m))
           })
         }
+        // A subscriber told of those attachments may stop the turn right here.
+        if (turn.controller.signal.aborted) {
+          await discardUploads(this.#api, uploaded)
+          throw abortError()
+        }
       }
       // Nothing may start once stop() or a conversation switch has withdrawn the turn.
       if (turn.controller.signal.aborted) throw abortError()

chat-sdk/test/attachments.test.ts

@@ -440,4 +440,32 @@ describe('sendMessage with attachments', () => {
     ).toEqual([key])
     expect(chat.getState()).toMatchObject({ status: 'idle', messages: [], conversations: [] })
   })
+
+  test('a subscriber stopping when the attachments appear still discards the uploads', async () => {
+    const { fetch, calls } = fetchMock(
+      upload,
+      (c) =>
+        c.method === 'DELETE' && c.url.pathname === '/api/w/ws/job_helpers/delete_s3_file'
+          ? json('deleted')
+          : undefined,
+      run,
+      answer
+    )
+    const chat = createChat(options(fetch))
+    chat.subscribe((s) => {
+      if (s.messages.some((m) => m.attachments)) void chat.stop()
+    })
+    await expect(
+      chat.sendMessage('read this', {
+        attachments: [{ name: 'contract.pdf', data: pdf }],
+        attachmentsInput: { name: 'files', multiple: true }
+      })
+    ).rejects.toMatchObject({ name: 'AbortError' })
+    expect(runs(calls)).toHaveLength(0)
+    const key = uploads(calls)[0].url.searchParams.get('file_key')
+    expect(
+      calls.filter((c) => c.method === 'DELETE').map((c) => c.url.searchParams.get('file_key'))
+    ).toEqual([key])
+    expect(chat.getState()).toMatchObject({ status: 'idle', messages: [], conversations: [] })
+  })
 })