From 16d5d85c6b1c05ef273e92858ba4bb0f1d3001ca Mon Sep 17 00:00:00 2001 From: owenob1 Date: Tue, 7 Jul 2026 00:42:34 +0800 Subject: [PATCH] feat: implement the full Cloudflare email SDK surface (v1.0.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds out @mvrx/mail from a thin @mvrx/aecs re-export into the complete AECS-SDK-1 surface. All modules implement against the interfaces already defined in adapters.ts. Core (package root): - storage: d1Init, d1Store, getThread, getMessage, listMessages (fixed §3.7 DDL, cursor-based §3.8 pagination, raw D1 prepared stmts) - send: sendEmail - rules: evaluateRules + Rule/Condition/Action types + D1 rule storage (§15) Subpath entry points: - /transports: cfTransport, smtpTransport, r2BlobStore (§3.5) - /providers: 8 AI connectors — cf, openai, anthropic, gemini, mistral, azure, ollama, openai-compat (§6) - /tools: deterministic analysis — extractAddresses, detectIntent, requiresReply, extractDates, extractLinks (§7.1) - /ai-tools: summarize, classify, sentiment, extractEntities, extractAction, ask (§7.2) - /compose: draft, reply, improve, tone, translate, suggestSubjects, send, createCompose (§8) - /attachments: attachmentsForAI, AttachmentProcessor, processors (chain, storeToR2, pdfToText, ocr, transcribe, cfPdfExtractor, runOcr, runTranscribe) (§9) - /hub: UserHub Durable Object + publishEvent/hubRouter/hubBus, typed MailEvent union, SSE fan-out, fire-and-forget (§16) Tests: 111 passing via @cloudflare/vitest-pool-workers with real D1/R2/DO bindings in workerd (not mocks). Example worker demonstrates the full receive -> store -> rules -> events -> classify -> reply loop and typechecks against the built package. Version 0.1.0 -> 1.0.0 (Appendix C: full surface stable). Signed-off-by: owenob1 --- .github/workflows/ci.yml | 5 +- examples/basic-worker/package.json | 16 + examples/basic-worker/src/index.ts | 104 +- examples/basic-worker/tsconfig.json | 13 + examples/basic-worker/wrangler.jsonc | 2 +- packages/mail/package.json | 66 +- packages/mail/src/ai-tools/index.ts | 239 ++++ packages/mail/src/attachments/index.ts | 343 +++++ packages/mail/src/compose/index.ts | 456 +++++++ packages/mail/src/hub/index.ts | 189 +++ packages/mail/src/index.ts | 7 + packages/mail/src/providers/index.ts | 190 +++ packages/mail/src/rules/index.ts | 267 ++++ packages/mail/src/send.ts | 13 + packages/mail/src/storage.ts | 521 ++++++++ packages/mail/src/tools.ts | 236 ++++ packages/mail/src/transports/index.ts | 259 ++++ packages/mail/test/ai-tools.test.ts | 202 +++ packages/mail/test/attachments.test.ts | 373 ++++++ packages/mail/test/compose.test.ts | 329 +++++ packages/mail/test/env.d.ts | 9 + packages/mail/test/hub.test.ts | 77 ++ packages/mail/test/providers.test.ts | 258 ++++ packages/mail/test/rules.test.ts | 413 +++++++ packages/mail/test/storage.test.ts | 268 ++++ packages/mail/test/tools.test.ts | 157 +++ packages/mail/test/transports.test.ts | 141 +++ packages/mail/test/worker.ts | 9 + packages/mail/vitest.config.ts | 22 + pnpm-lock.yaml | 1581 ++++++++++++++++++++++++ 30 files changed, 6743 insertions(+), 22 deletions(-) create mode 100644 examples/basic-worker/package.json create mode 100644 examples/basic-worker/tsconfig.json create mode 100644 packages/mail/src/ai-tools/index.ts create mode 100644 packages/mail/src/attachments/index.ts create mode 100644 packages/mail/src/compose/index.ts create mode 100644 packages/mail/src/hub/index.ts create mode 100644 packages/mail/src/providers/index.ts create mode 100644 packages/mail/src/rules/index.ts create mode 100644 packages/mail/src/send.ts create mode 100644 packages/mail/src/storage.ts create mode 100644 packages/mail/src/tools.ts create mode 100644 packages/mail/src/transports/index.ts create mode 100644 packages/mail/test/ai-tools.test.ts create mode 100644 packages/mail/test/attachments.test.ts create mode 100644 packages/mail/test/compose.test.ts create mode 100644 packages/mail/test/env.d.ts create mode 100644 packages/mail/test/hub.test.ts create mode 100644 packages/mail/test/providers.test.ts create mode 100644 packages/mail/test/rules.test.ts create mode 100644 packages/mail/test/storage.test.ts create mode 100644 packages/mail/test/tools.test.ts create mode 100644 packages/mail/test/transports.test.ts create mode 100644 packages/mail/test/worker.ts create mode 100644 packages/mail/vitest.config.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fd7e995..2c284d8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,5 +21,8 @@ jobs: cache: pnpm - run: pnpm install --frozen-lockfile - - run: pnpm -r typecheck + # Build first so workspace packages that consume a sibling's built output + # (e.g. examples/basic-worker importing @mvrx/mail subpaths, whose types + # resolve to dist/) can be type-checked. - run: pnpm -r build + - run: pnpm -r typecheck diff --git a/examples/basic-worker/package.json b/examples/basic-worker/package.json new file mode 100644 index 0000000..6ca4aa6 --- /dev/null +++ b/examples/basic-worker/package.json @@ -0,0 +1,16 @@ +{ + "name": "@mvrx/example-basic-worker", + "version": "0.0.0", + "private": true, + "type": "module", + "scripts": { + "typecheck": "tsc --noEmit" + }, + "dependencies": { + "@mvrx/mail": "workspace:*" + }, + "devDependencies": { + "@cloudflare/workers-types": "^4.20250109.0", + "typescript": "^5.8.0" + } +} diff --git a/examples/basic-worker/src/index.ts b/examples/basic-worker/src/index.ts index d7ae424..37cb720 100644 --- a/examples/basic-worker/src/index.ts +++ b/examples/basic-worker/src/index.ts @@ -1,21 +1,103 @@ -import { parse, wrappers, type CloudflareMailEnv } from "@mvrx/mail"; +import { parse, d1Init, d1Store, loadRules, evaluateRules } from "@mvrx/mail"; +import { cfTransport } from "@mvrx/mail/transports"; +import { cfProvider } from "@mvrx/mail/providers"; +import { classify } from "@mvrx/mail/ai-tools"; +import { compose } from "@mvrx/mail/compose"; +import { processors } from "@mvrx/mail/attachments"; +import { publishEvent, hubRouter } from "@mvrx/mail/hub"; + +// Register the UserHub Durable Object (backs real-time SSE events). +export { UserHub } from "@mvrx/mail/hub"; + +interface Env { + DB: D1Database; + BLOBS: R2Bucket; + AI: Ai; + EMAIL: SendEmail; + HUB: DurableObjectNamespace; + AGENT_MODEL_CLASSIFY: string; + AGENT_MODEL_CHAT: string; +} export default { - async email(message: ForwardableEmailMessage, env: CloudflareMailEnv) { + // Inbound: parse (+ archive attachments to R2 and extract PDF text) → store → + // run rules → notify connected clients → AI classify → auto-acknowledge. + async email(message: ForwardableEmailMessage, env: Env) { const email = await parse(message, { - wrapper: wrappers.xml("email"), + // Attachment handlers run during parse: store bytes to R2, then pull text + // out of PDFs via Workers AI so it's queryable / AI-ready. + onAttachment: processors.chain( + processors.storeToR2(env.BLOBS, { keyPrefix: "att" }), + processors.pdfToText({ extractor: processors.cfPdfExtractor(env.AI) }) + ), }); - console.log({ - messageId: email.messageId, - threadId: email.threadId, - from: email.metadata.from.email, - subject: email.metadata.subject, - forAI: email.content.forAI, + await d1Init(env.DB); + await d1Store(env.DB, email); + + // Single-tenant default: the recipient address is the userId. + const userId = message.to; + + // Evaluate stored rules (forward/auto-reply fire through the transport). + const rules = await loadRules(env.DB); + const results = await evaluateRules(email, rules, cfTransport(env.EMAIL)); + for (const r of results) { + if (!r.matched) continue; + await publishEvent(env.HUB, userId, { + type: "rule_fired", + payload: { + ruleId: r.ruleId, + messageId: email.messageId, + threadId: email.threadId, + actions: r.actions.map((a) => a.type), + }, + }); + } + + // Push a real-time "new message" event to any connected SSE clients. + await publishEvent(env.HUB, userId, { + type: "new_message", + payload: { + messageId: email.messageId, + threadId: email.threadId, + from: email.metadata.from, + subject: email.metadata.subject, + }, }); + + // Classify with Workers AI. + const ai = cfProvider(env.AI); + const { category } = await classify(email, ai, { model: env.AGENT_MODEL_CLASSIFY }); + console.log({ messageId: email.messageId, category }); + + // Auto-acknowledge with an AI-drafted reply, threaded correctly. + const { body } = await compose.reply(email, ai, { + intent: "acknowledge receipt and say we'll respond within one business day", + tone: "friendly", + model: env.AGENT_MODEL_CHAT, + }); + + await compose.send( + { + from: { name: "Support", email: "support@example.com" }, + to: [email.metadata.from], + subject: `Re: ${email.metadata.subject ?? ""}`, + inReplyTo: email.messageId, + references: [...email.thread.references, email.messageId], + }, + body, + cfTransport(env.EMAIL) + ); }, - async fetch(): Promise { - return new Response("AECS basic Worker example"); + // Mount the real-time SSE endpoint: clients connect with `new EventSource("/hub")`. + async fetch(req: Request, env: Env): Promise { + const url = new URL(req.url); + if (url.pathname === "/hub") { + // Derive the userId from your auth in production; single-tenant demo below. + const userId = url.searchParams.get("user") ?? "demo"; + return hubRouter(req, env.HUB, userId); + } + return new Response("AECS mail Worker — receive, store, rules, events, classify, reply"); }, }; diff --git a/examples/basic-worker/tsconfig.json b/examples/basic-worker/tsconfig.json new file mode 100644 index 0000000..43cf44a --- /dev/null +++ b/examples/basic-worker/tsconfig.json @@ -0,0 +1,13 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "lib": ["ES2022"], + "types": ["@cloudflare/workers-types"], + "strict": true, + "noEmit": true, + "skipLibCheck": true + }, + "include": ["src/**/*.ts"] +} diff --git a/examples/basic-worker/wrangler.jsonc b/examples/basic-worker/wrangler.jsonc index ec19652..a1a3f26 100644 --- a/examples/basic-worker/wrangler.jsonc +++ b/examples/basic-worker/wrangler.jsonc @@ -25,7 +25,7 @@ "durable_objects": { "bindings": [ { - "name": "USER_HUB", + "name": "HUB", "class_name": "UserHub" } ] diff --git a/packages/mail/package.json b/packages/mail/package.json index e003b15..a66f5e9 100644 --- a/packages/mail/package.json +++ b/packages/mail/package.json @@ -1,12 +1,14 @@ { "name": "@mvrx/mail", - "version": "0.1.0", + "version": "1.0.0", "description": "Cloudflare Email Routing SDK: send, receive, and store AI-ready email using the @mvrx/aecs standard.", "license": "AGPL-3.0-only", "type": "module", "main": "./dist/index.js", "types": "./dist/index.d.ts", - "files": ["dist"], + "files": [ + "dist" + ], "exports": { ".": { "import": "./dist/index.js", @@ -35,12 +37,41 @@ "./types": { "import": "./dist/types.js", "types": "./dist/types.d.ts" + }, + "./transports": { + "import": "./dist/transports/index.js", + "types": "./dist/transports/index.d.ts" + }, + "./providers": { + "import": "./dist/providers/index.js", + "types": "./dist/providers/index.d.ts" + }, + "./tools": { + "import": "./dist/tools.js", + "types": "./dist/tools.d.ts" + }, + "./ai-tools": { + "import": "./dist/ai-tools/index.js", + "types": "./dist/ai-tools/index.d.ts" + }, + "./compose": { + "import": "./dist/compose/index.js", + "types": "./dist/compose/index.d.ts" + }, + "./attachments": { + "import": "./dist/attachments/index.js", + "types": "./dist/attachments/index.d.ts" + }, + "./hub": { + "import": "./dist/hub/index.js", + "types": "./dist/hub/index.d.ts" } }, "scripts": { "typecheck": "tsc --noEmit", "build": "tsc", - "test": "node --test test/*.test.mjs" + "test": "vitest run", + "test:watch": "vitest" }, "dependencies": { "@mvrx/aecs": "github:mvrxapp/aecs#main" @@ -52,16 +83,33 @@ "zod": "^3.0.0" }, "peerDependenciesMeta": { - "@cloudflare/workers-types": { "optional": true }, - "drizzle-orm": { "optional": true }, - "hono": { "optional": true }, - "zod": { "optional": true } + "@cloudflare/workers-types": { + "optional": true + }, + "drizzle-orm": { + "optional": true + }, + "hono": { + "optional": true + }, + "zod": { + "optional": true + } }, "devDependencies": { + "@cloudflare/vitest-pool-workers": "^0.18.0", "@cloudflare/workers-types": "^4.20250109.0", - "typescript": "^5.8.0" + "typescript": "^5.8.0", + "vitest": "^4.1.0" }, - "keywords": ["email", "ai", "mime", "aecs", "cloudflare", "workers"], + "keywords": [ + "email", + "ai", + "mime", + "aecs", + "cloudflare", + "workers" + ], "repository": { "type": "git", "url": "https://github.com/mvrxapp/mail.git", diff --git a/packages/mail/src/ai-tools/index.ts b/packages/mail/src/ai-tools/index.ts new file mode 100644 index 0000000..3388e55 --- /dev/null +++ b/packages/mail/src/ai-tools/index.ts @@ -0,0 +1,239 @@ +import type { NormalizedEmail } from "@mvrx/aecs"; +import type { AiChatProvider, AiMessage } from "../adapters.js"; + +/** + * AI-powered analysis tools (AECS-SDK-1 §7.2). + * + * Every tool builds a prompt from `email.content.forAI` (falling back to + * `content.clean` / `content.text` if `forAI` is null), optionally appends + * attachment `extractedText`, calls the supplied `AiChatProvider`, and + * returns a parsed result. + */ + +/** Default model used when `options.model` is not supplied. Matches the + * documented default for `cfProvider` (§6.2), the SDK's first-class, + * zero-latency provider. Callers should override via `options.model` when + * using a different provider. */ +const DEFAULT_MODEL = "@cf/meta/llama-3.3-70b-instruct"; + +export interface AiToolOptions { + /** Override the model used for this call. Defaults to `DEFAULT_MODEL`. */ + model?: string; + /** Append `att.extractedText` from all attachments as additional LLM context. Default: false. */ + includeAttachments?: boolean; +} + +export interface SummarizeOptions extends AiToolOptions { + /** Target sentence count for the summary. Default: 2. */ + maxSentences?: number; +} + +export interface ClassifyOptions extends AiToolOptions { + /** Candidate categories to classify into. */ + categories?: string[]; +} + +export interface ClassifyResult { + category: string; + confidence: number; +} + +export interface ExtractActionResult { + action: string; + params: Record; +} + +export interface SentimentResult { + sentiment: "positive" | "neutral" | "negative"; + confidence: number; +} + +export interface ExtractEntitiesResult { + people: string[]; + companies: string[]; + products: string[]; + amounts: string[]; +} + +export interface AskOptions extends AiToolOptions { + /** The question to answer about the email (and its attachments). */ + question: string; +} + +// ── Internal helpers ───────────────────────────────────────────────────────── + +function baseText(email: NormalizedEmail): string { + return email.content.forAI ?? email.content.clean ?? email.content.text ?? ""; +} + +function buildContext(email: NormalizedEmail, options?: AiToolOptions): string { + let context = baseText(email); + + if (options?.includeAttachments) { + const attachmentTexts = (email.attachments ?? []) + .filter((att) => !!att.extractedText) + .map((att) => `--- Attachment: ${att.filename} ---\n${att.extractedText}`); + + if (attachmentTexts.length > 0) { + context = `${context}\n\n${attachmentTexts.join("\n\n")}`.trim(); + } + } + + return context; +} + +function resolveModel(options?: AiToolOptions): string { + return options?.model ?? DEFAULT_MODEL; +} + +function parseJson(text: string, fallback: T): T { + try { + const cleaned = text + .trim() + .replace(/^```(?:json)?\s*/i, "") + .replace(/```\s*$/i, ""); + return JSON.parse(cleaned) as T; + } catch { + return fallback; + } +} + +async function run( + provider: AiChatProvider, + options: AiToolOptions | undefined, + messages: AiMessage[] +): Promise { + const { text } = await provider.run(resolveModel(options), messages); + return text; +} + +// ── Tools ──────────────────────────────────────────────────────────────────── + +/** Summarise the email in `options.maxSentences` sentences (default: 2). */ +export async function summarize( + email: NormalizedEmail, + provider: AiChatProvider, + options?: SummarizeOptions +): Promise { + const maxSentences = options?.maxSentences ?? 2; + const messages: AiMessage[] = [ + { + role: "system", + content: `You are an email summarization assistant. Summarize the email below in at most ${maxSentences} sentence(s). Respond with only the summary text, no preamble or labels.`, + }, + { role: "user", content: buildContext(email, options) }, + ]; + const text = await run(provider, options, messages); + return text.trim(); +} + +/** Classify the email into one of `options.categories`. */ +export async function classify( + email: NormalizedEmail, + provider: AiChatProvider, + options?: ClassifyOptions +): Promise { + const categories = options?.categories; + const instruction = categories?.length + ? `Classify the email below into exactly one of these categories: ${categories.join(", ")}.` + : "Classify the email below into an appropriate short category label."; + + const messages: AiMessage[] = [ + { + role: "system", + content: `${instruction} Respond with ONLY strict JSON matching this shape, no other text: {"category": string, "confidence": number between 0 and 1}.`, + }, + { role: "user", content: buildContext(email, options) }, + ]; + const text = await run(provider, options, messages); + return parseJson(text, { category: categories?.[0] ?? "other", confidence: 0 }); +} + +/** Extract the primary actionable item (task, meeting, request, etc.) from the email. */ +export async function extractAction( + email: NormalizedEmail, + provider: AiChatProvider, + options?: AiToolOptions +): Promise { + const messages: AiMessage[] = [ + { + role: "system", + content: + 'Extract the primary actionable item from the email below. Respond with ONLY strict JSON matching this shape, no other text: {"action": string (a snake_case action identifier, e.g. "schedule_meeting"), "params": object (relevant parameters such as date, time, participants)}. If there is no clear action, respond with {"action": "none", "params": {}}.', + }, + { role: "user", content: buildContext(email, options) }, + ]; + const text = await run(provider, options, messages); + return parseJson(text, { action: "none", params: {} }); +} + +/** Detect the overall sentiment of the email. */ +export async function sentiment( + email: NormalizedEmail, + provider: AiChatProvider, + options?: AiToolOptions +): Promise { + const messages: AiMessage[] = [ + { + role: "system", + content: + 'Analyze the sentiment of the email below. Respond with ONLY strict JSON matching this shape, no other text: {"sentiment": "positive" | "neutral" | "negative", "confidence": number between 0 and 1}.', + }, + { role: "user", content: buildContext(email, options) }, + ]; + const text = await run(provider, options, messages); + return parseJson(text, { sentiment: "neutral", confidence: 0 }); +} + +/** Extract key entities (people, companies, products, amounts) from the email. */ +export async function extractEntities( + email: NormalizedEmail, + provider: AiChatProvider, + options?: AiToolOptions +): Promise { + const messages: AiMessage[] = [ + { + role: "system", + content: + 'Extract key entities from the email below. Respond with ONLY strict JSON matching this shape, no other text: {"people": string[], "companies": string[], "products": string[], "amounts": string[]}. Use an empty array for any category with no matches.', + }, + { role: "user", content: buildContext(email, options) }, + ]; + const text = await run(provider, options, messages); + return parseJson(text, { + people: [], + companies: [], + products: [], + amounts: [], + }); +} + +/** Answer a free-form question about the email (and its attachments). */ +export async function ask( + email: NormalizedEmail, + provider: AiChatProvider, + options: AskOptions +): Promise { + const messages: AiMessage[] = [ + { + role: "system", + content: + "Answer the question below using only information contained in the email context provided. If the answer cannot be found, say so plainly. Respond with a concise natural-language answer, not JSON.", + }, + { + role: "user", + content: `${buildContext(email, options)}\n\nQuestion: ${options.question}`, + }, + ]; + const text = await run(provider, options, messages); + return text.trim(); +} + +export const aiTools = { + summarize, + classify, + extractAction, + sentiment, + extractEntities, + ask, +}; diff --git a/packages/mail/src/attachments/index.ts b/packages/mail/src/attachments/index.ts new file mode 100644 index 0000000..0bacc8b --- /dev/null +++ b/packages/mail/src/attachments/index.ts @@ -0,0 +1,343 @@ +import type { Attachment, AttachmentHandler, ForAIWrapper, NormalizedEmail } from "@mvrx/aecs"; +import { wrappers } from "@mvrx/aecs/wrappers"; +import type { BlobPutOptions, BlobStore } from "../adapters.js"; +import { r2BlobStore } from "../transports/index.js"; + +/** + * Attachment processing pipeline (AECS-SDK-1 §9.3–9.6). + * + * The spec marks §9.3–9.8 "Status: Roadmap" — this module is that pipeline's + * implementation: a composable `AttachmentProcessor`/`AttachmentHandler` + * chain (store to R2, extract PDF text, OCR images, transcribe audio) plus + * `attachmentsForAI`, which aggregates whatever `att.extractedText` the + * chain populated into a single LLM-ready string. + * + * DECISION: `ocr`, `transcribe`, `cfPdfExtractor`, `runOcr`, and + * `runTranscribe` all take the raw Cloudflare `Ai` binding rather than the + * SDK's `AiChatProvider` (see `../adapters.js` / `cfProvider` in + * `../providers/index.js`). `AiChatProvider.run(model, messages)` is a + * text-only chat interface — there is no way to attach image or audio bytes + * to an `AiMessage`. Vision/audio inference on Workers AI requires calling + * `Ai.run(model, inputs)` directly with model-specific input shapes (e.g. + * `{ image: number[] }`, `{ audio: number[] }`), so these processors bypass + * `AiChatProvider` entirely and depend on `Ai` instead. + */ + +// ── AttachmentProcessor (§9.5) ─────────────────────────────────────────────── + +/** + * Object-form processor: `accepts` gates whether `process` runs for a given + * attachment. Compose one or more into a single `AttachmentHandler` with + * `processors.chain(...)`. + */ +export interface AttachmentProcessor { + accepts(att: Attachment): boolean; + process(att: Attachment): Promise | void; +} + +function isAttachmentProcessor( + proc: AttachmentProcessor | AttachmentHandler +): proc is AttachmentProcessor { + return typeof proc !== "function"; +} + +// ── processors.chain (§9.4) ────────────────────────────────────────────────── + +/** + * Composes any mix of `AttachmentProcessor` objects and bare + * `AttachmentHandler` functions into a single `AttachmentHandler`, run in + * order against `parse()`'s `onAttachment` option. Processor objects only + * run when `accepts(att)` returns true; handler functions always run. + */ +export function chain(...procs: (AttachmentProcessor | AttachmentHandler)[]): AttachmentHandler { + return async (att, ctx) => { + for (const proc of procs) { + if (isAttachmentProcessor(proc)) { + if (proc.accepts(att)) await proc.process(att); + } else { + await proc(att, ctx); + } + } + }; +} + +// ── processors.storeToR2 (§9.3) ────────────────────────────────────────────── + +export interface StoreToR2Options { + /** Prefix for the stored key, before `/`. Default: "att". */ + keyPrefix?: string; + /** Derive a public/signed URL for the stored key. See NOTE below. */ + publicUrl?: (key: string) => string; +} + +/** + * Stores `att.content()` bytes to a `BlobStore` (or a raw Cloudflare + * `R2Bucket`, which is wrapped via `r2BlobStore()`), keyed + * `${keyPrefix}//`, and records that key on + * `att.blobKey`. + * + * NOTE: §9.3 of the spec shows `publicUrl` populating an `att.url` field, + * but the `Attachment` type (`@mvrx/aecs`) has no `url` property — only + * `blobKey`. `publicUrl`, when supplied, is still invoked and forwarded to + * `BlobStore.put` as `BlobPutOptions.publicUrl` (for stores that record it), + * but there is nowhere on `Attachment` to persist the resulting URL, so only + * `att.blobKey` is set here. + */ +export function storeToR2(store: BlobStore | R2Bucket, options?: StoreToR2Options): AttachmentHandler { + const blobStore: BlobStore = isRawR2Bucket(store) ? r2BlobStore(store) : store; + const keyPrefix = options?.keyPrefix ?? "att"; + + return async (att, ctx) => { + const key = `${keyPrefix}/${ctx.messageId}/${att.filename}`; + const bytes = await att.content(); + + const putOptions: BlobPutOptions = { contentType: att.contentType }; + if (options?.publicUrl) putOptions.publicUrl = options.publicUrl(key); + + await blobStore.put(key, bytes, putOptions); + att.blobKey = key; + }; +} + +/** `R2Bucket.head()` has no equivalent on `BlobStore`, so its presence + * distinguishes a raw R2 binding from an already-wrapped `BlobStore`. */ +function isRawR2Bucket(store: BlobStore | R2Bucket): store is R2Bucket { + return typeof (store as R2Bucket).head === "function"; +} + +// ── processors.pdfToText (§9.4) ────────────────────────────────────────────── + +export interface PdfToTextOptions { + /** Extracts text from PDF bytes. Use `processors.cfPdfExtractor(env.AI)` or a custom extractor. */ + extractor: (bytes: Uint8Array) => Promise; +} + +/** Sets `att.extractedText` for `application/pdf` attachments via the supplied extractor. */ +export function pdfToText(options: PdfToTextOptions): AttachmentHandler { + return async (att) => { + if (att.contentType !== "application/pdf") return; + att.extractedText = await options.extractor(await att.content()); + }; +} + +// ── processors.cfPdfExtractor (§9.4) ───────────────────────────────────────── + +/** + * PDF-to-text extractor backed by Cloudflare Workers AI document conversion + * (`Ai.toMarkdown`, which accepts `{ name, blob }` documents and returns + * `{ format: "markdown", data }` — or `{ format: "error", error }` — per + * attachment). Returns the converted markdown, or `null` on conversion + * failure. + */ +export function cfPdfExtractor(ai: Ai): (bytes: Uint8Array) => Promise { + return async (bytes: Uint8Array): Promise => { + const blob = new Blob([bytes], { type: "application/pdf" }); + const [result] = await ai.toMarkdown([{ name: "document.pdf", blob }]); + if (!result || result.format !== "markdown") return null; + return result.data; + }; +} + +// ── processors.ocr (§9.4) ──────────────────────────────────────────────────── + +const DEFAULT_OCR_MODEL = "@cf/llava-hf/llava-1.5-7b-hf"; +const DEFAULT_OCR_PROMPT = "Extract all text visible in this image."; + +export interface OcrOptions { + ai: Ai; + /** Vision model to run. Default: "@cf/llava-hf/llava-1.5-7b-hf". */ + model?: string; + /** Instruction passed to the vision model. Default: "Extract all text visible in this image." */ + prompt?: string; +} + +/** Sets `att.extractedText` for `image/*` attachments via Workers AI OCR (see `runOcr`). */ +export function ocr(options: OcrOptions): AttachmentHandler { + return async (att) => { + if (!att.contentType.startsWith("image/")) return; + att.extractedText = await runOcr(options.ai, await att.content(), { + model: options.model, + prompt: options.prompt, + }); + }; +} + +/** + * Runs OCR on raw image bytes directly against the `Ai` binding (real + * Workers AI vision inference — not verified by these offline tests; see + * test/attachments.test.ts). Uses `ai.run(model, inputs)` rather than + * `AiChatProvider`, since chat providers are text-only (see file-level + * DECISION comment above). + * + * `model`/`prompt` default to the same values as `ocr()`'s options so this + * can also be called directly (e.g. from a Queue consumer doing async + * extraction, §9.8) with just `(ai, bytes)`. + */ +export async function runOcr( + ai: Ai, + bytes: Uint8Array, + options?: { model?: string; prompt?: string } +): Promise { + const model: string = options?.model ?? DEFAULT_OCR_MODEL; + const prompt: string = options?.prompt ?? DEFAULT_OCR_PROMPT; + // Workers AI vision models take raw image bytes as a plain number array. + const input: Record = { image: Array.from(bytes), prompt }; + const result = (await ai.run(model, input)) as { description?: string }; + return result.description ?? null; +} + +// ── processors.transcribe (§9.4) ───────────────────────────────────────────── + +const DEFAULT_TRANSCRIBE_MODEL = "@cf/openai/whisper"; + +export interface TranscribeOptions { + ai: Ai; + /** Speech-to-text model to run. Default: "@cf/openai/whisper". */ + model?: string; + /** BCP-47 language hint (only honored by models that accept it, e.g. whisper-large-v3-turbo). */ + language?: string; +} + +/** Sets `att.extractedText` for `audio/*` attachments via Workers AI transcription (see `runTranscribe`). */ +export function transcribe(options: TranscribeOptions): AttachmentHandler { + return async (att) => { + if (!att.contentType.startsWith("audio/")) return; + att.extractedText = await runTranscribe(options.ai, await att.content(), { + model: options.model, + language: options.language, + }); + }; +} + +/** + * Transcribes raw audio bytes directly against the `Ai` binding (real + * Workers AI transcription — not verified by these offline tests; see + * test/attachments.test.ts). Uses `ai.run(model, inputs)` rather than + * `AiChatProvider`, for the same reason as `runOcr` (see file-level + * DECISION comment above). + * + * `model`/`language` default to the same values as `transcribe()`'s options + * so this can also be called directly with just `(ai, bytes)` (e.g. from a + * Queue consumer, §9.8). + */ +export async function runTranscribe( + ai: Ai, + bytes: Uint8Array, + options?: { model?: string; language?: string } +): Promise { + const model: string = options?.model ?? DEFAULT_TRANSCRIBE_MODEL; + // Workers AI whisper models take raw audio bytes as a plain number array. + const input: Record = { audio: Array.from(bytes) }; + if (options?.language) input.language = options.language; + const result = (await ai.run(model, input)) as { text?: string }; + return result.text ?? null; +} + +// ── processors namespace ───────────────────────────────────────────────────── + +export const processors = { + chain, + storeToR2, + pdfToText, + cfPdfExtractor, + ocr, + transcribe, + runOcr, + runTranscribe, +}; + +// ── attachmentsForAI (§9.6) ─────────────────────────────────────────────────── + +const DEFAULT_MAX_CHARS_PER_ATTACHMENT = 4_000; +const DEFAULT_MAX_TOTAL_CHARS = 16_000; + +export interface AttachmentsForAIOptions { + /** Max characters per attachment. Default: 4_000. */ + maxCharsPerAttachment?: number; + /** Max total characters across all attachments. Default: 16_000. */ + maxTotalChars?: number; + /** + * Wrap each attachment's text block. Default: `wrappers.xml("attachment")`. + * Set to `null` to disable wrapping. + */ + wrapper?: ForAIWrapper | null; + /** + * Which content types to include. Accepts exact types or `type/*` glob + * patterns (e.g. `["application/pdf", "image/*", "audio/*"]`). + * Default: include all attachments that have `extractedText` set. + */ + include?: string[]; + /** Label for each attachment block. Default: `(att) => att.filename`. */ + label?: (att: Attachment) => string; +} + +function matchesContentType(pattern: string, contentType: string): boolean { + if (pattern.endsWith("/*")) return contentType.startsWith(pattern.slice(0, -1)); + return pattern === contentType; +} + +/** + * Aggregates `att.extractedText` across `attachments` into a single + * LLM-ready string, once processors (`processors.pdfToText`/`ocr`/ + * `transcribe`/a custom `AttachmentProcessor`) have populated it. Attachments + * without `extractedText` are skipped. Returns `null` if no attachment + * contributed a block. + * + * Self-contained: unlike the rest of this module, this does not reuse any + * `@mvrx/aecs` aggregation logic (there is none for attachments) — it's a + * standalone implementation of the format documented in AECS-SDK-1 §9.6. + * + * Each block's body is `name="