fix(chat): give each attachment its own object, and read a long turn whole

Two attachments can arrive under one name — a reader picking `report.pdf` from
two folders. Keyed on the name alone they raced to a single object, so the agent
read whichever landed last twice and never saw the other. Each gets a segment of
its own inside the turn's prefix; the name stays the last segment, so anything
reading a name off the key still sees what was attached.

The messages endpoint answers oldest-first with a limit, so one request only ever
reached the start of a turn that wrote more rows than a page — an agent calling
several tools a round. Its answer was among the rows left behind, and the sweep
then dropped the temp rows standing in for it, so the answer vanished until the
conversation was next loaded. The poll reads on until a page comes back short.

Leaving uploaded objects behind when a send does not run is deliberate, and now
says so: deleting them is a request that can fail on a path already failing, and
a resend uploads its own, so a lost send costs one prefix rather than a growing
number.

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

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

@@ -75,6 +75,11 @@ const FOLLOW_RETRY_DELAY_MS = 500
 /** How often a turn with no stream asks its job whether the run is over. */
 const SETTLE_POLL_MS = 2000
 
+/** A poll reads the conversation a page at a time, and stops after this many — enough for a
+ * long tool-using turn, bounded so a page that stops advancing cannot loop. */
+const POLL_PAGE_SIZE = 50
+const POLL_MAX_PAGES = 20
+
 /**
  * The agent events the worker streams, in the shape the transcript applies. The SDK names
  * the same six events after the wire protocol; this is the rest of the app's vocabulary.
@@ -922,18 +927,31 @@ export class FlowChatManager {
 		if (!this.#workspace()) return
 
 		try {
-			const lastSeq = this.getLastPersistedMessageSeq(conversationId)
 			const startedIn = this.#generation
-			const response = await FlowConversationsService.listConversationMessages({
-				workspace: this.#workspace()!,
-				conversationId: conversationId,
-				page: 1,
-				perPage: 50,
-				afterSeq: lastSeq
-			})
-			// An interval tick already dispatched outlives `clearInterval`, and a turn's final
-			// poll outlives its abort — either would put a forgotten flow's rows back.
-			if (startedIn !== this.#generation) return
+			// Paged, not one request: the endpoint answers oldest-first with a limit, so a turn
+			// that wrote more rows than a page — an agent calling several tools a round — would
+			// hand back its earliest and leave its answer behind, and the sweep below drops the
+			// temp rows that were standing in for it.
+			const response: ChatMessage[] = []
+			let afterSeq = this.getLastPersistedMessageSeq(conversationId)
+			for (let page = 0; page < POLL_MAX_PAGES; page++) {
+				const batch = await FlowConversationsService.listConversationMessages({
+					workspace: this.#workspace()!,
+					conversationId: conversationId,
+					page: 1,
+					perPage: POLL_PAGE_SIZE,
+					afterSeq
+				})
+				// An interval tick already dispatched outlives `clearInterval`, and a turn's final
+				// poll outlives its abort — either would put a forgotten flow's rows back.
+				if (startedIn !== this.#generation) return
+				response.push(...batch)
+				if (batch.length < POLL_PAGE_SIZE) break
+				const furthest = Math.max(...batch.map((m) => m.created_seq))
+				// A page that moved nothing would ask for the same rows forever.
+				if (afterSeq !== undefined && furthest <= afterSeq) break
+				afterSeq = furthest
+			}
 
 			if (options?.isNewConversation) {
 				await this.refreshConversations()

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

@@ -172,6 +172,37 @@ describe('a chat re-pointed at another flow', () => {
 	})
 })
 
+/**
+ * The messages endpoint answers oldest-first with a limit, so one request only reaches the
+ * start of a turn that wrote a lot of rows — an agent calling several tools a round. Its
+ * answer is among the rows that would be left behind.
+ */
+describe('reading a turn longer than one page', () => {
+	const assistantRows = (from: number, count: number) =>
+		Array.from({ length: count }, (_, i) => ({
+			id: `m${from + i}`,
+			conversation_id: 'a',
+			message_type: 'assistant',
+			content: `row ${from + i}`,
+			created_at: new Date().toISOString(),
+			created_seq: from + i
+		}))
+
+	it('keeps reading until a page comes back short', async () => {
+		vi.mocked(FlowConversationsService.listConversationMessages)
+			.mockReset()
+			.mockResolvedValueOnce(assistantRows(1, 50) as any)
+			.mockResolvedValueOnce(assistantRows(51, 12) as any)
+		const manager = managerWithRows()
+		manager.selectedConversationId = 'a'
+
+		await (manager as any).pollConversationMessages('a', {})
+
+		expect(manager.messages).toHaveLength(62)
+		expect(manager.messages.at(-1)?.content).toBe('row 62')
+	})
+})
+
 /**
  * A queued message goes out when the turn ahead of it reaches a terminal state — not
  * merely when the chat stops looking busy. The stream dropping is the case that separates

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

@@ -466,9 +466,13 @@ export class FlowChatViewHost implements ChatViewHost {
 		)
 		if (!started) {
 			// The upload succeeded and the run did not, so the composer's draft was spent on
-			// nothing. The uploaded objects stay where they are — a resend uploads its own,
-			// under its own prefix — but what the reader wrote comes back, to the chat it was
-			// written in rather than the one open by now.
+			// nothing. What the reader wrote comes back, to the chat it was written in rather
+			// than the one open by now.
+			//
+			// The uploaded objects are deliberately left in place. Deleting them is itself a
+			// request that can fail, on a path that is already failing, and a resend uploads
+			// its own under a fresh prefix — so a lost send costs one prefix, not a growing
+			// number. The workspace's own storage retention is what collects them.
 			this.#restoreToComposer({ ...options, conversationId })
 			return false
 		}
@@ -522,8 +526,12 @@ export class FlowChatViewHost implements ChatViewHost {
 	): Promise<S3Attachment[]> {
 		const workspace = this.#options.workspace?.()
 		if (!workspace) throw new Error('no workspace')
-		// One prefix per turn keeps a re-attached filename from overwriting the copy an
-		// earlier message still refers to.
+		// A prefix per turn, and a segment per attachment inside it. The turn's prefix keeps a
+		// re-attached filename off the copy an earlier message still points at; the segment
+		// does the same within one turn, where two files can arrive under one name and would
+		// otherwise race to a single key and leave the agent reading one of them twice. The
+		// name itself stays the last segment, so anything that reads a name off the key
+		// still sees what the reader attached.
 		const prefix = `windmill_chat_uploads/${randomUUID()}`
 		return Promise.all(
 			attachments.map(async (attachment, index) => {
@@ -534,7 +542,7 @@ export class FlowChatViewHost implements ChatViewHost {
 				)
 				const { file_key } = await HelpersService.fileUpload({
 					workspace,
-					fileKey: `${prefix}/${filename}`,
+					fileKey: `${prefix}/${index}/${filename}`,
 					contentType: blob.type,
 					requestBody: blob
 				})

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

@@ -87,6 +87,41 @@ describe('a send whose attachments are still uploading', () => {
 	})
 
 	// Stop has no job to cancel yet, so it has to be honoured when the upload lands —
+
+	/**
+	 * Two files can arrive under one name. Keyed on the name alone they race to the same
+	 * object, and the agent reads whichever landed last — twice — while the other is gone.
+	 */
+	it('gives two attachments sharing a name their own objects', async () => {
+		const manager = stubManager('a')
+		const keys: string[] = []
+		vi.mocked(HelpersService.fileUpload).mockImplementation((async (args: any) => {
+			keys.push(args.fileKey)
+			return { file_key: args.fileKey }
+		}) as any)
+
+		await host(manager).sendRequest({
+			instructions: 'read both',
+			blobs: [
+				{
+					name: 'report.pdf',
+					dataUrl: 'data:application/pdf;base64,AAA',
+					mediaType: 'application/pdf'
+				},
+				{
+					name: 'report.pdf',
+					dataUrl: 'data:application/pdf;base64,BBB',
+					mediaType: 'application/pdf'
+				}
+			] as any
+		})
+
+		expect(keys).toHaveLength(2)
+		expect(new Set(keys).size).toBe(2)
+		// The name a reader sees is still the one they attached.
+		expect(keys.every((k) => k.endsWith('/report.pdf'))).toBe(true)
+	})
+
 	// otherwise the run starts and the reader watches a message they took back execute.
 	it('does not run after a Stop pressed while it uploaded', async () => {
 		const manager = stubManager('a')