Add attachment support to email_reply outbound sends

docs/contributing/architecture/primitives.yaml

@@ -497,7 +497,9 @@ primitives:
     name: Email blob storage (R2)
     summary:
       EMAIL_BLOBS R2 bucket holding raw email MIME under the frozen key contract
-      email-raw:v1:{userId}/{messageId}; D1 email_messages keeps metadata plus
+      email-raw:v1:{userId}/{messageId} and standalone outbound attachment bytes
+      under email-attachment:v1:{userId}/{messageId}/{attachmentId}
+      (storage_kind 'external'); D1 email_messages keeps metadata plus
       raw_mime_key, with inline raw_mime as the fallback when the binding is
       unavailable. Covered by account deletion and retention.
     code:

docs/use/email-primitives.md

@@ -46,7 +46,11 @@ Use the MCP `email` domain:
 - `email_send` sends a notification email to your own account email address
   (notify-self only; any other recipient is rejected).
 - `email_reply` replies to a stored inbound message. The recipient always comes
-  from the stored message.
+  from the stored message. Optional `attachments` (up to 10 of
+  `{ filename, content_type, content_base64 }`) are sent with the reply and
+  stored as `external` attachments readable later via `email_attachment_get`.
+  With attachments, the whole message (bodies plus decoded attachment bytes)
+  must fit the plan's `email_message_bytes` per-message cap.
 - `email_attachment_get` returns stored attachment bytes by attachment id.
 - `email_message_list` lists stored inbound and outbound messages.
 - `email_message_search` searches stored messages by case-insensitive substring

packages/shared/src/outbound-email.ts

@@ -2,6 +2,7 @@ import {
 	array,
 	createSchema,
 	fail,
+	literal,
 	object,
 	optional,
 	string,
@@ -36,6 +37,19 @@ const optionalHeadersSchema = createSchema<
 	return { value: headers }
 })
 
+const outboundEmailAttachmentSchema = object({
+	/** Base64-encoded bytes, as the Cloudflare Email REST API expects. */
+	content: nonEmptyStringSchema,
+	filename: nonEmptyStringSchema,
+	type: nonEmptyStringSchema,
+	disposition: union([literal('attachment'), literal('inline')]),
+	contentId: optional(nonEmptyStringSchema),
+})
+
+export type OutboundEmailAttachment = InferOutput<
+	typeof outboundEmailAttachmentSchema
+>
+
 const outboundEmailSchema = object({
 	from: nonEmptyStringSchema,
 	to: union([nonEmptyStringSchema, array(nonEmptyStringSchema)]),
@@ -44,6 +58,7 @@ const outboundEmailSchema = object({
 	text: optional(nonEmptyStringSchema),
 	replyTo: optional(nonEmptyStringSchema),
 	headers: optionalHeadersSchema,
+	attachments: optional(array(outboundEmailAttachmentSchema)),
 })
 
 export type OutboundEmail = InferOutput<typeof outboundEmailSchema>

packages/worker/src/app/account-deletion.ts

@@ -95,7 +95,7 @@ type UserDeletionInventory = {
 	vectorIds: Array<string>
 	storageIds: Array<string>
 	bundleKvKeys: Array<string>
-	emailRawMimeKeys: Array<string>
+	emailBlobKeys: Array<string>
 	sourceSnapshots: Array<UserSourceSnapshot>
 	savedPackages: Array<UserSavedPackageSnapshot>
 	repoSessions: Array<UserRepoSessionSnapshot>
@@ -283,15 +283,30 @@ async function listUserPackageServices(env: Env, userId: string) {
 	}))
 }
 
-async function listUserEmailRawMimeKeys(env: Env, userId: string) {
-	const rows = await env.APP_DB.prepare(
-		`SELECT raw_mime_key
-		FROM email_messages
-		WHERE user_id = ? AND raw_mime_key IS NOT NULL`,
-	)
-		.bind(userId)
-		.all<{ raw_mime_key: string }>()
-	return uniqueStrings((rows.results ?? []).map((row) => row.raw_mime_key))
+async function listUserEmailBlobKeys(env: Env, userId: string) {
+	const [rawMimeRows, attachmentRows] = await Promise.all([
+		env.APP_DB.prepare(
+			`SELECT raw_mime_key
+			FROM email_messages
+			WHERE user_id = ? AND raw_mime_key IS NOT NULL`,
+		)
+			.bind(userId)
+			.all<{ raw_mime_key: string }>(),
+		// Externally stored attachments (outbound mail) have their own R2
+		// objects, separate from any raw-MIME blob.
+		env.APP_DB.prepare(
+			`SELECT attachment.storage_key AS storage_key
+			FROM email_attachments attachment
+			JOIN email_messages message ON message.id = attachment.message_id
+			WHERE message.user_id = ? AND attachment.storage_key IS NOT NULL`,
+		)
+			.bind(userId)
+			.all<{ storage_key: string }>(),
+	])
+	return uniqueStrings([
+		...(rawMimeRows.results ?? []).map((row) => row.raw_mime_key),
+		...(attachmentRows.results ?? []).map((row) => row.storage_key),
+	])
 }
 
 async function listUserCommunityListingIds(env: Env, userId: string) {
@@ -342,7 +357,7 @@ async function collectUserDeletionInventory(input: {
 	const [
 		vectorIds,
 		storageIds,
-		emailRawMimeKeys,
+		emailBlobKeys,
 		sourceSnapshots,
 		savedPackages,
 		repoSessions,
@@ -361,11 +376,9 @@ async function collectUserDeletionInventory(input: {
 			input.warnings.push(`Failed to enumerate storage ids: ${message}`)
 			return [] as Array<string>
 		}),
-		listUserEmailRawMimeKeys(input.env, input.userId).catch((error) => {
-			const message = error instanceof Error ? error.message : String(error)
-			input.warnings.push(
-				`Failed to enumerate email raw-MIME blob keys: ${message}`,
-			)
+		listUserEmailBlobKeys(input.env, input.userId).catch((error) => {
+			const message = getErrorMessage(error)
+			input.warnings.push(`Failed to enumerate email blob keys: ${message}`)
 			return [] as Array<string>
 		}),
 		listUserSourceSnapshots(input.env, input.userId).catch((error) => {
@@ -418,7 +431,7 @@ async function collectUserDeletionInventory(input: {
 		vectorIds,
 		storageIds,
 		bundleKvKeys,
-		emailRawMimeKeys,
+		emailBlobKeys,
 		sourceSnapshots,
 		savedPackages,
 		repoSessions,
@@ -1068,7 +1081,7 @@ export async function deleteUserAccount(input: {
 
 	result.deletedEmailBlobs = await deleteEmailBlobs({
 		blobs: input.env.EMAIL_BLOBS,
-		keys: inventory.emailRawMimeKeys,
+		keys: inventory.emailBlobKeys,
 		warnings,
 	})
 

packages/worker/src/app/email/cloudflare-email.ts

@@ -130,13 +130,14 @@ async function sendViaCloudflareApi(
 
 export async function sendCloudflareEmail(
 	config: CloudflareEmailClientConfig,
-	message: Omit<OutboundEmail, 'replyTo' | 'headers'> &
-		Partial<Pick<OutboundEmail, 'replyTo' | 'headers'>>,
+	message: Omit<OutboundEmail, 'replyTo' | 'headers' | 'attachments'> &
+		Partial<Pick<OutboundEmail, 'replyTo' | 'headers' | 'attachments'>>,
 ): Promise<CloudflareSendResult> {
 	const normalized = normalizeEmailPayload({
 		...message,
 		replyTo: message.replyTo,
 		headers: message.headers,
+		attachments: message.attachments,
 	})
 	const apiBaseUrl =
 		typeof config.apiBaseUrl === 'string' && config.apiBaseUrl.trim().length > 0

packages/worker/src/app/retention.node.test.ts

@@ -753,6 +753,16 @@ test('email message retention deletes old user rows, R2 blobs, attachments, and
 		) VALUES ('att-old', 'msg-old-blob', 'text/plain', 1, 'raw-mime', ?)`,
 		)
 		.run(daysAgo(emailMessageRetentionDays + 1))
+	// Externally stored attachment bytes (outbound mail) live in their own
+	// R2 object that must be deleted alongside the raw-MIME blobs.
+	sqlite
+		.prepare(
+			`INSERT INTO email_attachments (
+			id, message_id, content_type, size, storage_kind, storage_key, created_at
+		) VALUES ('att-ext', 'msg-old-plain', 'text/plain', 1, 'external',
+			'email-attachment:v1:user-1/msg-old-plain/att-ext', ?)`,
+		)
+		.run(daysAgo(emailMessageRetentionDays + 1))
 	const blobDelete = vi.fn(async () => undefined)
 
 	const result = await pruneUserEmailMessagesForRetention({
@@ -763,14 +773,18 @@ test('email message retention deletes old user rows, R2 blobs, attachments, and
 
 	expect(result).toEqual({
 		deletedMessages: 2,
-		deletedAttachments: 1,
+		deletedAttachments: 2,
 		deletedThreads: 1,
 		deletedRawMimeBlobs: 1,
+		deletedAttachmentBlobs: 1,
 		blobDeleteErrors: 0,
 		hasMore: false,
 		nextCursor: null,
 	})
-	expect(blobDelete).toHaveBeenCalledWith(['raw-mime/user-1/msg-old-blob'])
+	expect(blobDelete).toHaveBeenCalledWith([
+		'raw-mime/user-1/msg-old-blob',
+		'email-attachment:v1:user-1/msg-old-plain/att-ext',
+	])
 	expect(idsForTable(sqlite, 'email_messages')).toEqual([
 		'msg-boundary',
 		'msg-fresh',

packages/worker/src/app/retention.ts

@@ -26,6 +26,12 @@ export const retentionCronGateMinutes = 5
 export const retentionCronIntervalMinutes = 60
 export const retentionDefaultBatchSize = 250
 export const retentionDeleteIdsMaxParameters = 100
+/**
+ * R2 bulk deletes accept up to 1000 keys per call. A 250-message batch with
+ * raw MIME plus up to ten external attachment keys each can exceed that, so
+ * blob deletes are chunked below this ceiling.
+ */
+export const retentionBlobDeleteMaxKeys = 1000
 export const publishedBundleArtifactRetentionBatchSize = 100
 /**
  * Total wall-clock budget for one retention cron run. Each hourly run keeps
@@ -202,6 +208,7 @@ export type RetentionPruneResult = {
 		deletedAttachments: number
 		deletedThreads: number
 		deletedRawMimeBlobs: number
+		deletedAttachmentBlobs: number
 		blobDeleteErrors: number
 	}
 	entitlementDailyCounters: number
@@ -714,6 +721,7 @@ export type EmailMessageRetentionResult = {
 	deletedAttachments: number
 	deletedThreads: number
 	deletedRawMimeBlobs: number
+	deletedAttachmentBlobs: number
 	blobDeleteErrors: number
 	hasMore: boolean
 	nextCursor: EmailMessageRetentionCursor | null
@@ -760,27 +768,87 @@ export async function pruneUserEmailMessagesForRetention(input: {
 		deletedAttachments: 0,
 		deletedThreads: 0,
 		deletedRawMimeBlobs: 0,
+		deletedAttachmentBlobs: 0,
 		blobDeleteErrors: 0,
 		hasMore: false,
 		nextCursor: null,
 	}
-	const rowsWithBlob = rows.filter((row) => row.raw_mime_key !== null)
-	let deletableRows = rows
-	if (rowsWithBlob.length > 0) {
-		// Never delete a row whose R2 blob could not be deleted first: once
-		// the row is gone its raw_mime_key is lost and the blob is orphaned
-		// forever. Failed rows are skipped and retried on a later run.
+	// Externally stored attachments (outbound mail) have their own R2
+	// objects that must be deleted along with the raw-MIME blobs.
+	const attachmentKeysByMessageId = new Map<string, Array<string>>()
+	for (
+		let index = 0;
+		index < rows.length;
+		index += retentionDeleteIdsMaxParameters
+	) {
+		const chunk = rows.slice(index, index + retentionDeleteIdsMaxParameters)
+		const chunkPlaceholders = chunk.map(() => '?').join(', ')
+		const attachmentRows = await runD1WithRetry(() =>
+			input.db
+				.prepare(
+					`SELECT message_id, storage_key
+				FROM email_attachments
+				WHERE storage_key IS NOT NULL AND message_id IN (${chunkPlaceholders})`,
+				)
+				.bind(...chunk.map((row) => row.id))
+				.all<{ message_id: string; storage_key: string }>(),
+		)
+		for (const attachmentRow of attachmentRows.results ?? []) {
+			const keys = attachmentKeysByMessageId.get(attachmentRow.message_id)
+			if (keys) keys.push(attachmentRow.storage_key)
+			else {
+				attachmentKeysByMessageId.set(attachmentRow.message_id, [
+					attachmentRow.storage_key,
+				])
+			}
+		}
+	}
+	const blobKeysForRow = (row: (typeof rows)[number]) => [
+		...(row.raw_mime_key !== null ? [row.raw_mime_key] : []),
+		...(attachmentKeysByMessageId.get(row.id) ?? []),
+	]
+	const rowsWithBlob = rows.filter((row) => blobKeysForRow(row).length > 0)
+	// Never delete a row whose R2 blobs could not be deleted first: once
+	// the row is gone its blob keys are lost and the blobs are orphaned
+	// forever. Failed rows are skipped and retried on a later run. Deletes
+	// are chunked below R2's bulk-delete key limit, keeping each message's
+	// keys inside one call so a failure skips whole messages.
+	const failedRowIds = new Set<string>()
+	let chunkRows: Array<(typeof rows)[number]> = []
+	let chunkKeys: Array<string> = []
+	const deleteChunk = async () => {
+		if (chunkKeys.length === 0) return
 		try {
-			await input.blobs.delete(
-				rowsWithBlob.map((row) => row.raw_mime_key as string),
+			await input.blobs.delete(chunkKeys)
+			result.deletedRawMimeBlobs += chunkRows.filter(
+				(row) => row.raw_mime_key !== null,
+			).length
+			result.deletedAttachmentBlobs += chunkRows.reduce(
+				(count, row) =>
+					count + (attachmentKeysByMessageId.get(row.id)?.length ?? 0),
+				0,
 			)
-			result.deletedRawMimeBlobs = rowsWithBlob.length
 		} catch (error) {
-			result.blobDeleteErrors = rowsWithBlob.length
-			deletableRows = rows.filter((row) => row.raw_mime_key === null)
-			console.warn('retention-email-raw-mime-delete-failed', { error })
+			result.blobDeleteErrors += chunkKeys.length
+			for (const row of chunkRows) failedRowIds.add(row.id)
+			console.warn('retention-email-blob-delete-failed', { error })
+		}
+		chunkRows = []
+		chunkKeys = []
+	}
+	for (const row of rowsWithBlob) {
+		const keys = blobKeysForRow(row)
+		if (
+			chunkKeys.length > 0 &&
+			chunkKeys.length + keys.length > retentionBlobDeleteMaxKeys
+		) {
+			await deleteChunk()
 		}
+		chunkRows.push(row)
+		chunkKeys.push(...keys)
 	}
+	await deleteChunk()
+	const deletableRows = rows.filter((row) => !failedRowIds.has(row.id))
 	const messageIds = deletableRows.map((row) => row.id)
 	// email_attachments rows reference email_messages via FK, so delete the
 	// dependent attachment rows first.
@@ -930,6 +998,7 @@ export async function pruneRetention(input: {
 			deletedAttachments: 0,
 			deletedThreads: 0,
 			deletedRawMimeBlobs: 0,
+			deletedAttachmentBlobs: 0,
 			blobDeleteErrors: 0,
 		},
 		entitlementDailyCounters: 0,
@@ -1027,6 +1096,8 @@ export async function pruneRetention(input: {
 				result.emailMessages.deletedAttachments += batch.deletedAttachments
 				result.emailMessages.deletedThreads += batch.deletedThreads
 				result.emailMessages.deletedRawMimeBlobs += batch.deletedRawMimeBlobs
+				result.emailMessages.deletedAttachmentBlobs +=
+					batch.deletedAttachmentBlobs
 				result.emailMessages.blobDeleteErrors += batch.blobDeleteErrors
 				emailMessagesCursor = batch.nextCursor
 				return batch.hasMore