Cloudflare Artifact + Blob Store

withGenerationPersistence(persistence) needs only stores.generationRuns to track a generation's lifecycle. Add stores.artifacts (metadata) and stores.blobs (the bytes) — both, or neither — and the middleware also persists the generated media: image/audio/TTS/video/transcription bytes land at blob key artifacts/<runId>/<artifactId>, with an ArtifactRecord row describing each.

The deliverable is one file in the Worker — e.g. src/lib/generation-persistence.ts — exporting a factory that builds an AIPersistence from the request's R2 + D1 bindings, plus a GET route that serves artifact bytes with retrieveArtifact / retrieveBlob.

Read the sibling ai-persistence/build-cloudflare-adapter skill for the per-request-binding rule, wrangler config shape, D1 migration workflow, and the chat (generation-run/message) side. This skill covers only the two byte-storage stores and how to compose them.

The two contracts, verbatim

Both come from @tanstack/ai-persistence. defineBlobStore / defineArtifactStore type an object literal inline (autocomplete + contract checking, no separate annotation).

// BlobStore — the byte layer. R2 backs it.
interface BlobStore {
  put(
    key: string,
    body: BlobBody,
    options?: BlobPutOptions,
  ): Promise<BlobRecord>
  get(key: string): Promise<BlobObject | null> // metadata + byte accessors
  head(key: string): Promise<BlobRecord | null> // metadata only
  delete(key: string): Promise<void> // no-op if absent
  list(options?: BlobListOptions): Promise<BlobListPage>
}

// ArtifactStore — the metadata layer. D1 (or KV) backs it.
interface ArtifactStore {
  save(record: ArtifactRecord): Promise<void> // insert or overwrite
  get(artifactId: string): Promise<ArtifactRecord | null>
  list(runId: string): Promise<Array<ArtifactRecord>> // [] when none
  delete(artifactId: string): Promise<void>
  deleteForRun(runId: string): Promise<void>
}

BlobBody is ReadableStream<Uint8Array> | ArrayBuffer | ArrayBufferView | string | Blob — which is exactly what R2Bucket.put accepts, so the body flows straight through with no conversion. BlobPutOptions is { contentType?, customMetadata? }; BlobListOptions is { prefix?, cursor?, limit? }; BlobListPage is { objects: BlobRecord[], cursor?, truncated? }.

1. BlobStore backed by R2

R2Object carries size, etag, httpMetadata.contentType, customMetadata, and uploaded (a Date). BlobRecord wants createdAt / updatedAt as epoch ms — R2 tracks only the single uploaded instant, so map it to both. get / head are the byte-body vs metadata-only split; R2ObjectBody already exposes body, arrayBuffer(), and text(), so a BlobObject is essentially the R2 object plus the mapped metadata.

import { defineBlobStore } from '@tanstack/ai-persistence'
import type { BlobObject, BlobRecord } from '@tanstack/ai-persistence'

function toRecord(obj: R2Object): BlobRecord {
  const uploaded = obj.uploaded.getTime()
  return {
    key: obj.key,
    size: obj.size,
    etag: obj.etag,
    ...(obj.httpMetadata?.contentType
      ? { contentType: obj.httpMetadata.contentType }
      : {}),
    ...(obj.customMetadata ? { customMetadata: obj.customMetadata } : {}),
    createdAt: uploaded,
    updatedAt: uploaded,
  }
}

export function r2BlobStore(bucket: R2Bucket) {
  return defineBlobStore({
    async put(key, body, options) {
      const obj = await bucket.put(key, body, {
        ...(options?.contentType
          ? { httpMetadata: { contentType: options.contentType } }
          : {}),
        ...(options?.customMetadata
          ? { customMetadata: options.customMetadata }
          : {}),
      })
      // R2.put returns null only when an onlyIf precondition fails — not used here.
      if (!obj) throw new Error(`R2 put failed for ${key}`)
      return toRecord(obj)
    },

    async get(key): Promise<BlobObject | null> {
      const obj = await bucket.get(key)
      if (!obj) return null
      return {
        ...toRecord(obj),
        body: obj.body,
        arrayBuffer: () => obj.arrayBuffer(),
        text: () => obj.text(),
      }
    },

    async head(key) {
      const obj = await bucket.head(key)
      return obj ? toRecord(obj) : null
    },

    async delete(key) {
      await bucket.delete(key)
    },

    async list(options) {
      // R2 reads `limit: 0` as "use the default", so short-circuit it.
      if (options?.limit === 0) {
        return { objects: [] }
      }
      const page = await bucket.list({
        ...(options?.prefix !== undefined ? { prefix: options.prefix } : {}),
        ...(options?.cursor !== undefined ? { cursor: options.cursor } : {}),
        ...(options?.limit !== undefined ? { limit: options.limit } : {}),
        // R2 omits httpMetadata/customMetadata from list rows unless asked.
        include: ['httpMetadata', 'customMetadata'],
      })
      return {
        objects: page.objects.map(toRecord),
        ...(page.truncated ? { cursor: page.cursor, truncated: true } : {}),
      }
    },
  })
}

Invariants that matter (asserted by the conformance testkit):

• get / head return null for a missing key; delete is a silent no-op.
• put overwrites an existing key.
• list filters by prefix literally (R2 prefix is a literal byte prefix — no glob), returns keys in ascending order, and pages via the opaque cursor when truncated. R2's own cursor is opaque and satisfies this directly. limit: 0 must yield an empty, untruncated page — R2 treats limit: 0 as "use the default", so special-case it: if (options?.limit === 0) return { objects: [] }.

2. ArtifactStore backed by D1

ArtifactRecord is { artifactId, runId, threadId, name, mimeType, size, sourceUrl?, createdAt } (createdAt epoch ms). One flat table, keyed by artifact_id, indexed by run_id for list.

CREATE TABLE IF NOT EXISTS generation_artifacts (
  artifact_id  text PRIMARY KEY NOT NULL,
  run_id       text NOT NULL,
  thread_id    text NOT NULL,
  blob_key     text,
  name         text NOT NULL,
  mime_type    text NOT NULL,
  size         integer NOT NULL,
  source_url   text,
  created_at   integer NOT NULL
);
CREATE INDEX IF NOT EXISTS generation_artifacts_run ON generation_artifacts (run_id);

import { defineArtifactStore } from '@tanstack/ai-persistence'
import type { ArtifactRecord } from '@tanstack/ai-persistence'

interface ArtifactRow {
  artifact_id: string
  run_id: string
  thread_id: string
  blob_key: string | null
  name: string
  mime_type: string
  size: number
  source_url: string | null
  created_at: number
}

function fromRow(row: ArtifactRow): ArtifactRecord {
  return {
    artifactId: row.artifact_id,
    runId: row.run_id,
    threadId: row.thread_id,
    ...(row.blob_key != null ? { blobKey: row.blob_key } : {}),
    name: row.name,
    mimeType: row.mime_type,
    size: row.size,
    ...(row.source_url != null ? { sourceUrl: row.source_url } : {}),
    createdAt: row.created_at,
  }
}

export function d1ArtifactStore(db: D1Database) {
  return defineArtifactStore({
    async save(record) {
      // Insert or overwrite (artifact ids are unique).
      await db
        .prepare(
          `INSERT INTO generation_artifacts
             (artifact_id, run_id, thread_id, blob_key, name, mime_type, size, source_url, created_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
           ON CONFLICT(artifact_id) DO UPDATE SET
             run_id = excluded.run_id, thread_id = excluded.thread_id,
             blob_key = excluded.blob_key, name = excluded.name,
             mime_type = excluded.mime_type, size = excluded.size,
             source_url = excluded.source_url, created_at = excluded.created_at`,
        )
        .bind(
          record.artifactId,
          record.runId,
          record.threadId,
          record.blobKey ?? null,
          record.name,
          record.mimeType,
          record.size,
          record.sourceUrl ?? null,
          record.createdAt,
        )
        .run()
    },

    async get(artifactId) {
      const row = await db
        .prepare(`SELECT * FROM generation_artifacts WHERE artifact_id = ?`)
        .bind(artifactId)
        .first<ArtifactRow>()
      return row ? fromRow(row) : null
    },

    async list(runId) {
      const { results } = await db
        .prepare(`SELECT * FROM generation_artifacts WHERE run_id = ?`)
        .bind(runId)
        .all<ArtifactRow>()
      return results.map(fromRow)
    },

    async delete(artifactId) {
      await db
        .prepare(`DELETE FROM generation_artifacts WHERE artifact_id = ?`)
        .bind(artifactId)
        .run()
    },

    async deleteForRun(runId) {
      await db
        .prepare(`DELETE FROM generation_artifacts WHERE run_id = ?`)
        .bind(runId)
        .run()
    },
  })
}

Omitting source_url / blob_key from the record when the column is NULL keeps records comparing cleanly against the reference in-memory store. Persist blob_key verbatim: a storageKey mapper can put the bytes anywhere, so a reader cannot recompute the path — resolveArtifactBlobKey(record) falls back to the default convention only for rows written before the column existed. KV alternative: if you have no D1, back save/get with KV.put(artifactId, JSON.stringify(record)) / KV.get(artifactId, 'json'), and maintain a run:<runId> index key (a JSON array of artifact ids) for list — KV has no query, so list needs that secondary index.

3. Compose and wire

Bindings are per-request on Workers, so export a factory. Combine the byte stores with a generation-run store (and, if this Worker also does chat, the chat stores). Either build the whole AIPersistence with defineAIPersistence, or layer the artifact stores onto an existing chat persistence with composePersistence:

import {
  defineAIPersistence,
  composePersistence,
  withGenerationPersistence,
} from '@tanstack/ai-persistence'
import { r2BlobStore } from './r2-blob-store'
import { d1ArtifactStore } from './d1-artifact-store'
import { d1GenerationRunStore } from './d1-job-store' // your GenerationRunStore

/** Call inside a request handler — bindings are not available at module scope. */
export function generationPersistence(env: Env) {
  return defineAIPersistence({
    stores: {
      generationRuns: d1GenerationRunStore(env.DB),
      artifacts: d1ArtifactStore(env.DB),
      blobs: r2BlobStore(env.ARTIFACTS_BUCKET),
    },
  })
}

// …or add bytes to a persistence that already has chat + generation runs:
// composePersistence(chatAndJobsPersistence(env), {
//   overrides: {
//     artifacts: d1ArtifactStore(env.DB),
//     blobs: r2BlobStore(env.ARTIFACTS_BUCKET),
//   },
// })

withGenerationPersistence throws if exactly one of artifacts / blobs is present — provide both or neither. Wire it as generation middleware:

import { generateImage, toServerSentEventsResponse } from '@tanstack/ai'
import { openaiImage } from '@tanstack/ai-openai'
import { withGenerationPersistence } from '@tanstack/ai-persistence'
import { generationPersistence } from './lib/generation-persistence'

export default {
  async fetch(request: Request, env: Env) {
    const { prompt, threadId } = await request.json()
    const stream = generateImage({
      adapter: openaiImage('gpt-image-1'),
      prompt,
      threadId, // the slot recorded on the job + artifacts
      stream: true,
      middleware: [
        withGenerationPersistence(generationPersistence(env), { threadId }),
      ],
    })
    return toServerSentEventsResponse(stream)
  },
}

4. Serve the bytes back

A GET route resolves an artifactId to its record and its stored bytes. retrieveArtifact returns the ArtifactRecord (or null → 404); retrieveBlob returns the BlobObject (metadata + a streamable body). Both key off artifactBlobKey({ runId, artifactId }) internally, so you never build the key yourself.

import { retrieveArtifact, retrieveBlob } from '@tanstack/ai-persistence'
import { generationPersistence } from './lib/generation-persistence'

export async function GET(request: Request, env: Env) {
  const artifactId = new URL(request.url).searchParams.get('id') ?? ''
  const persistence = generationPersistence(env)

  // Authorize before serving — derive the owner from the session, never trust
  // a client-supplied id. (The record carries runId/threadId to check against.)
  const record = await retrieveArtifact(persistence, artifactId)
  if (!record) return new Response('Not found', { status: 404 })

  const blob = await retrieveBlob(persistence, record) // pass the record: no 2nd lookup
  if (!blob?.body) return new Response('Not found', { status: 404 })

  return new Response(blob.body, {
    headers: {
      'content-type': record.mimeType,
      'content-length': String(record.size),
    },
  })
}

To hydrate a server-driven generation client (persistence: true + a stable threadId) on mount, also expose reconstructGeneration(persistence, request) on a GET that reads ?threadId= / ?runId= — see ai-core/client-persistence.

Other backends — the contract is tiny, here's how each maps

BlobStore is five methods over an object store. Any of these backs it; swap the factory, keep everything else. put maps to the SDK's upload, get to a download that exposes body/arrayBuffer/text, head to a metadata fetch, delete to a delete, list to a prefixed, cursor-paged list.

For each: contentType and customMetadata ride the SDK's own metadata fields; BlobRecord.createdAt/updatedAt come from the object's stored timestamps (epoch ms); return null from get/head on a not-found rather than throwing.

Verify

The shared runPersistenceConformance testkit covers all seven stores, including generationRuns, artifacts, and blobs — point it at your factory rather than hand-writing these assertions:

import { runPersistenceConformance } from '@tanstack/ai-persistence/testkit'
import { env } from 'cloudflare:test'
import { generationPersistence } from '../src/lib/generation-persistence'

// A generation-only Worker declares the chat state stores it does not provide.
runPersistenceConformance('app-r2', () => generationPersistence(env), {
  skip: ['messages', 'runs', 'interrupts', 'metadata'],
})

Run it against a Miniflare R2 + D1 binding with the migration applied, reset between runs (see ai-persistence/build-cloudflare-adapter for the cloudflare:test harness pattern). It exercises, among the rest:

• put then get round-trips bytes and metadata; get/head return null for a missing key; delete is a silent no-op on an absent key.
• put overwrites an existing key (and its contentType/customMetadata).
• list filters by prefix literally and case-sensitively, returns ascending keys, pages through the cursor when truncated without gaps or repeats, and returns an empty untruncated page for limit: 0.
• The ArtifactStore: save is insert-or-overwrite, get returns null when absent, list(runId) returns [] for an unknown run, and delete / deleteForRun remove exactly the expected rows.
• The GenerationRunStore: createOrResume idempotency, no-op update on an unknown id, and findLatestForThread returning the most recently started linked run (terminal ones included).

An end-to-end check is the strongest signal: run generateImage through withGenerationPersistence(generationPersistence(env), { threadId }), then confirm the blob exists at artifacts/<runId>/<artifactId> and retrieveBlob streams it back.