diff --git a/.changeset/client-typed-storage-adapter-defaults.md b/.changeset/client-typed-storage-adapter-defaults.md index 983ba5753..e515fdabe 100644 --- a/.changeset/client-typed-storage-adapter-defaults.md +++ b/.changeset/client-typed-storage-adapter-defaults.md @@ -3,15 +3,16 @@ --- `localStoragePersistence` / `sessionStoragePersistence` / `indexedDBPersistence` -default their `TValue` back to `ChatPersistedState` instead of `any`. +are no longer generic. Each returns a `ChatStorageAdapter`, +and `WebStoragePersistenceOptions` types its `serialize` / `deserialize` codec +over `ChatPersistedState`. -The `any` default was justified by a claim that "a bare call works for both the -chat **and generation** `persistence` options with no type argument". That is no -longer true: generation `persistence` is now `boolean` (server-driven only), so -chat is the sole `persistence` option that takes a storage adapter — and the -`any` default erased `getItem` / `setItem` type safety for chat users in exchange -for nothing. +The type parameter existed so one adapter could back both the chat and the +generation `persistence` option. Generation `persistence` is now `boolean` +(server-driven only), so chat is the sole option that takes a storage adapter and +the parameter had no second value to hold. -A bare `localStoragePersistence()` still needs no type argument. Only a -standalone store holding something other than a chat transcript needs the -explicit one, e.g. `localStoragePersistence()`. +A bare `localStoragePersistence()` is unchanged. A call that passed an explicit +type argument for a standalone store, `localStoragePersistence()`, no +longer compiles: build that store with your own object literal, since these +factories are for chat state. diff --git a/.changeset/generation-mount-hydration-and-speech-restore.md b/.changeset/generation-mount-hydration-and-speech-restore.md index 13b7b3e02..b80ae37b2 100644 --- a/.changeset/generation-mount-hydration-and-speech-restore.md +++ b/.changeset/generation-mount-hydration-and-speech-restore.md @@ -16,7 +16,7 @@ results. together re-fired the hydrate GET on every discarded/speculative render, flooding the connection pool (`ERR_INSUFFICIENT_RESOURCES`). Hydration now runs once from `mountDevtools` (the hooks' commit-phase mount effect), guarded by `serverHydrationStarted`. - `initialResumeSnapshot` still seeds SSR/first paint. Note for direct + Note for direct (non-framework) `GenerationClient`/`VideoGenerationClient` users: mount hydration and the "missing `hydrateGeneration` handler" warning now fire from `mountDevtools()` rather than the constructor, so call `mountDevtools()` (as diff --git a/.changeset/generation-persistence-server-only.md b/.changeset/generation-persistence-server-only.md index fa29823c1..63c3a1978 100644 --- a/.changeset/generation-persistence-server-only.md +++ b/.changeset/generation-persistence-server-only.md @@ -23,14 +23,15 @@ restored differently: a client snapshot can never hold the generated bytes, so `result` came back `null` from storage but whole from the server. One mode removes that split. -Gone from `@tanstack/ai-client`: the `GenerationPersistence` type, the storage -read/write path in `GenerationClient` and `VideoGenerationClient`, and the -adapter arm of `GenerationPersistenceOption`. `persistence: true` still requires -a stable `threadId` at the type level, and still needs a `hydrateGeneration` -handler (every built-in connection has one) plus a `reconstructGeneration` route. +Gone from `@tanstack/ai-client`: the `GenerationPersistence` type and the storage +read/write path in `GenerationClient` and `VideoGenerationClient`. +`persistence: true` still requires a stable `threadId` at the type level, and +still needs a `hydrateGeneration` handler (every built-in connection has one) +plus a `reconstructGeneration` route. -`initialResumeSnapshot` is unchanged, so an app that wants to manage its own -storage can still seed the client from it. +The `initialResumeSnapshot` option went with it: it seeded the storage mode that +no longer exists, so the server hydration handler is the only way a run is +restored. **None of this touches chat.** `useChat` keeps both modes, and `localStoragePersistence` / `sessionStoragePersistence` / `indexedDBPersistence` diff --git a/.changeset/generation-persistence.md b/.changeset/generation-persistence.md index 68af8bcab..84027ac52 100644 --- a/.changeset/generation-persistence.md +++ b/.changeset/generation-persistence.md @@ -17,7 +17,7 @@ Add generation persistence, mirroring chat: media generation runs survive a relo **Server-side load (`reconstructGeneration`).** A new `reconstructGeneration(persistence, request, options?)` server helper — the generation parallel of `reconstructChat` — reads a `?runId=` (or `?threadId=`) from the request, authorizes it via an `authorize` callback, and returns `{ resumeSnapshot, activeRun }` JSON so a server-authoritative client restores the last run on mount. Requires the `generationRuns` store. `authorize` is optional at the type level for single-user and prototype routes, but any multi-user deployment must pass it: the run and thread ids arrive from the caller, so identity has to be derived from server-side session state and ownership checked before the helper reads persistence. The same applies to a route that serves artifact bytes by id. -**Media byte storage (server).** When the backend also provides both an `artifacts` (`ArtifactStore`) and a `blobs` (`BlobStore`) store, `withGenerationPersistence` writes each generated file's bytes to the blob store (key `artifacts//`), records an `ArtifactRecord`, and attaches `PersistedArtifactRef`s to the result and the run record. A new `artifactUrl` option stamps a durable app-origin serve URL onto each ref (a new `PersistedArtifactRef.url`) and rewrites the live result's media URL to it, so live and restored results both render media from your own origin instead of the provider's expiring link. Extraction is customizable via `extractArtifacts` / `nameArtifact`; `retrieveArtifact` / `retrieveBlob` (and the shared `artifactBlobKey`) serve the bytes back. Prompt media referenced by **URL** is not downloaded: the URL is caller-supplied, so fetching it server-side would be an SSRF vector, and the bytes are redundant. Opt in per-app with `allowInputUrl` (a predicate, so the check can't be skipped). Every artifact fetch is limited to `http:`/`https:`, timed out (`artifactFetchTimeoutMs`, default 30s) and size-capped (`maxArtifactBytes`, default 100 MiB); input fetches additionally block loopback/private/link-local hosts and refuse redirects. `artifactFetch` injects the `fetch` used, for routing downloads through an egress-restricted proxy. `memoryPersistence()` ships in-memory `artifacts`/`blobs` stores; the generation activities gained `threadId` / `runId` options. `@tanstack/ai-utils` adds `base64ToUint8Array`. +**Media byte storage (server).** When the backend also provides both an `artifacts` (`ArtifactStore`) and a `blobs` (`BlobStore`) store, `withGenerationPersistence` writes each generated file's bytes to the blob store (key `artifacts//`), records an `ArtifactRecord`, and attaches `PersistedArtifactRef`s to the result and the run record. A new `artifactUrl` option stamps a durable app-origin serve URL onto each ref (a new `PersistedArtifactRef.url`) and rewrites the live result's media URL to it, so live and restored results both render media from your own origin instead of the provider's expiring link. Extraction is customizable via `extractArtifacts` / `nameArtifact`; `retrieveArtifact` / `retrieveBlob` (which resolve the key through `resolveArtifactBlobKey`) serve the bytes back. Prompt media referenced by **URL** is not downloaded: the URL is caller-supplied, so fetching it server-side would be an SSRF vector, and the bytes are redundant. Opt in per-app with `allowInputUrl` (a predicate, so the check can't be skipped). Every artifact fetch is limited to `http:`/`https:`, timed out (`artifactFetchTimeoutMs`, default 30s) and size-capped (`maxArtifactBytes`, default 100 MiB); input fetches additionally block loopback/private/link-local hosts and refuse redirects. `artifactFetch` injects the `fetch` used, for routing downloads through an egress-restricted proxy. `memoryPersistence()` ships in-memory `artifacts`/`blobs` stores; the generation activities gained `threadId` / `runId` options. `@tanstack/ai-utils` adds `base64ToUint8Array`. **Client (transparent restore).** Generation hooks (`useGenerateImage`, `useGenerateVideo`, `useGenerateAudio`, `useGenerateSpeech`, `useGeneration`, `useSummarize`, `useTranscription`, and their Solid/Vue/Svelte/Angular equivalents) take a `persistence` option, and it is boolean — server-driven only, with no client-storage adapter arm: `true` hydrates the last run for a stable `threadId` on mount, and the browser caches nothing. Restore is **invisible**: it repaints the normal `result` / `status` / `error` fields as if the run had just finished, and reports the in-flight run's id as `runId` — there is no `resumeSnapshot` / `resumeState` / `pendingArtifacts` / `resultArtifacts` hook field. If a run is still generating when the connection drops or the page reloads, the client re-attaches to it and finishes it in place (via the connection's `joinRun` durability replay), exactly like `useChat`. With byte storage configured, a restored `result` is rebuilt whole, its media resolved to the durable serve URL and its refs on `result.artifacts`; without it, `status` / `error` restore and `result` stays null. The snapshot never holds the generated bytes and never restarts provider work — generation still only begins on `generate(...)`. @@ -27,4 +27,4 @@ Add generation persistence, mirroring chat: media generation runs survive a relo `findLatestForThread` is a **required** method on `GenerationRunStore` — a `?threadId=` lookup is the whole mount-time hydration path, so a store that cannot answer it cannot back generation persistence. TypeScript rejects a store that omits it; a JavaScript adapter that ships without it fails at the call, not silently. -Snapshots arriving from the server are validated with the new `parseGenerationResumeSnapshot` before anything is repainted. +Snapshots arriving from the server are validated before anything is repainted, so a stale or malformed record cannot paint a bogus result. diff --git a/.changeset/generation-run-threadid-required.md b/.changeset/generation-run-threadid-required.md index 41956cafe..e004bb268 100644 --- a/.changeset/generation-run-threadid-required.md +++ b/.changeset/generation-run-threadid-required.md @@ -27,9 +27,9 @@ so a record without one could be written and then never read back. And the client discarded any snapshot that arrived without one. That last disagreement was a silent failure: the server legitimately omitted -`threadId` for a record that had none, and `parseGenerationResumeSnapshot` -responded by dropping the **entire** snapshot — status, result and error along -with the cursor — leaving a blank idle panel with no diagnostic while the +`threadId` for a record that had none, and the client's snapshot validation +responded by dropping the **entire** snapshot (status, result and error along +with the cursor), leaving a blank idle panel with no diagnostic while the provider kept billing. Making the field required removes the disagreement by construction rather than patching one side of it. diff --git a/.changeset/hooks-expose-run-id.md b/.changeset/hooks-expose-run-id.md index 425efb437..eacfc1f59 100644 --- a/.changeset/hooks-expose-run-id.md +++ b/.changeset/hooks-expose-run-id.md @@ -38,9 +38,8 @@ ordinary streaming turn. `runId` tracks every run: it is set when any run starts `injectChat` (Angular) exposed no equivalent field before and now returns `runId` alongside the other frameworks. -`ChatResumeState` and `GenerationResumeState` remain exported — they still -describe the persisted resume snapshot (and `resumeInterruptsUnsafe` still takes -a `ChatResumeState`). They are simply no longer part of a hook's return shape. +`ChatResumeState` remains exported, since `resumeInterruptsUnsafe` still takes +one. It is simply no longer part of a hook's return shape. New docs page: [Id map](https://tanstack.com/ai/latest/docs/persistence/id-map) covers what each id means on chat versus generation, how to choose a `threadId`, diff --git a/.changeset/trim-persistence-public-surface.md b/.changeset/trim-persistence-public-surface.md new file mode 100644 index 000000000..9fb26b38a --- /dev/null +++ b/.changeset/trim-persistence-public-surface.md @@ -0,0 +1,47 @@ +--- +'@tanstack/ai-client': minor +'@tanstack/ai-persistence': minor +'@tanstack/ai-react': minor +'@tanstack/ai-preact': minor +'@tanstack/ai-solid': minor +'@tanstack/ai-svelte': minor +'@tanstack/ai-vue': minor +'@tanstack/ai-angular': minor +--- + +**Breaking:** trim the persistence public API down to what an app actually calls. + +Generation persistence is server-driven, so the types and options that only +existed to support a client-managed copy of a run are gone. + +- **`initialResumeSnapshot` is removed from every generation hook** (`useGeneration`, + `useGenerateImage`, `useGenerateVideo`, `useGenerateAudio`, `useGenerateSpeech`, + `useSummarize`, `useTranscription`, and the Solid / Vue / Svelte / Angular + equivalents) and from `GenerationClient` / `VideoGenerationClient`. A run is + restored by `persistence: true` plus a `hydrateGeneration` handler. **`useChat` + keeps its `initialResumeSnapshot`.** +- **No longer exported from `@tanstack/ai-client`** (they are internals of the + hydration path): `GenerationResumeSnapshot`, `GenerationResumeState`, + `GenerationResumeStatus`, `GenerationResultSnapshot`, `GenerationErrorSnapshot`, + `GenerationEventSnapshot`, `GenerationPendingArtifact`, + `parseGenerationResumeSnapshot`, `updateGenerationResumeSnapshot`, and + `ChatResumeSnapshot`. `GenerationPersistenceOption` (an alias for `boolean`) is + deleted; write `persistence?: boolean`. `GenerationPersistenceOptions`, the + union that requires a `threadId` alongside `persistence`, is unchanged. +- **`ChatResumeSnapshotV1` / `ChatResumeSnapshotV2` are collapsed into one shape** + with no `schemaVersion` field. The two versions were structurally identical, no + reader branched on the version, and only V2 was ever written. +- **The framework packages no longer re-export `PersistedArtifactRef`.** No hook + type refers to it; import it from `@tanstack/ai` where the artifact stores are + defined. +- **`artifactBlobKey` is no longer exported from `@tanstack/ai-persistence`.** Use + `resolveArtifactBlobKey(record)`, which its own docs already recommended for + reads, since a record written with a custom `storageKey` carries its real key. +- **`createInterruptController` and `InterruptController` are deleted.** The + controller only forwarded five calls to the `interrupts` store; call the store + directly (`persistence.stores.interrupts`). +- The ctx-capability plumbing (`PersistenceCapability`, `InterruptsCapability`, + `getPersistence`, `providePersistence`, `getInterrupts`, `provideInterrupts`) is + unchanged and now documented, for middleware that reads the stores + `withPersistence` holds. See + [Persistence internals](https://tanstack.com/ai/latest/docs/persistence/internals). diff --git a/docs/config.json b/docs/config.json index 9bbf14527..e50a13e3f 100644 --- a/docs/config.json +++ b/docs/config.json @@ -289,6 +289,21 @@ "addedAt": "2026-07-24", "updatedAt": "2026-07-31" }, + { + "label": "Build a Chat Adapter", + "to": "persistence/build-your-own-chat-adapter", + "addedAt": "2026-07-31" + }, + { + "label": "Build a Generation Adapter", + "to": "persistence/build-your-own-generation-adapter", + "addedAt": "2026-07-31" + }, + { + "label": "Store Reference", + "to": "persistence/store-reference", + "addedAt": "2026-07-31" + }, { "label": "Migrations", "to": "persistence/migrations", @@ -299,7 +314,7 @@ "label": "Internals", "to": "persistence/internals", "addedAt": "2026-07-22", - "updatedAt": "2026-07-30" + "updatedAt": "2026-07-31" } ] }, diff --git a/docs/persistence/build-your-own-adapter.md b/docs/persistence/build-your-own-adapter.md index c71fbf89a..254212c2c 100644 --- a/docs/persistence/build-your-own-adapter.md +++ b/docs/persistence/build-your-own-adapter.md @@ -15,10 +15,17 @@ store interfaces from `@tanstack/ai-persistence`. Implement the ones you want against your database, hand the result to `withPersistence`, and you are done. The core never inspects your tables, so the schema is yours to shape. -This guide builds a complete SQLite adapter on Node's built-in `node:sqlite`, end -to end, then shows how to map the same contracts onto a database schema you -already have. The runnable version of everything here lives in the -`examples/ts-react-chat` app (`src/lib/sqlite-persistence.ts`). +This page covers the shape every adapter has, which stores your app actually +needs, and how to verify the result. The two walkthroughs build a real one on +Node's built-in `node:sqlite`, end to end: + +- [Build a chat adapter](./build-your-own-chat-adapter): the transcript, run + lifecycle, durable approvals, and key/value state. +- [Build a generation adapter](./build-your-own-generation-adapter): generation + runs plus the artifact and blob stores that keep generated media. + +The runnable version of both lives in the `examples/ts-react-chat` app +(`src/lib/sqlite-persistence.ts`). ## Which stores do you need? @@ -53,8 +60,8 @@ Two pairs cannot be split: nothing describing them, is not a usable combination. So the smallest adapter worth shipping is a single `messages` store, and the -common production shape is `messages` + `runs` + `interrupts`. The guide below -builds them in that order. +common production shape is `messages` + `runs` + `interrupts`. The +[chat walkthrough](./build-your-own-chat-adapter) builds them in that order. ## What an adapter is @@ -86,17 +93,17 @@ For generation: - `generationRuns`: the generation run lifecycle. The counterpart to `runs`, keyed by its own `runId`. - `artifacts` + `blobs`: keep the generated media bytes. See - [Generation & media stores](#generation--media-stores). + [Build a generation adapter](./build-your-own-generation-adapter). The middleware turns on behavior for whatever stores it finds, so a `messages`-only adapter is a valid adapter. -Those seven — `messages`, `runs`, `interrupts`, `metadata`, `generationRuns`, -`artifacts`, `blobs` — are the *only* keys `stores` accepts; anything else -throws `Unknown AIPersistence store key` at construction. Need a mutex across +Those seven keys (`messages`, `runs`, `interrupts`, `metadata`, +`generationRuns`, `artifacts`, `blobs`) are the *only* ones `stores` accepts; +anything else throws `Unknown AIPersistence store key` at construction. Need a mutex across instances? That is `withLocks`; see [Locks](../advanced/locks). -Type each store with its `define*Store` helper, as the sections below do. There +Type each store with its `define*Store` helper, as both walkthroughs do. There is one per store: - `defineMessageStore` @@ -123,13 +130,13 @@ Annotate the value with a named shape: `stores.messages` is possibly `undefined`. Every method signature and invariant is in the -[store interface reference](#store-interface-reference) at the end of this page. +[store reference](./store-reference). The invariants (idempotent creates, insert-if-absent, ordered listings) are what the shared conformance suite checks, and getting one wrong is the usual source of subtle bugs. The records the stores hold form a small schema. The thread is not a table of -its own — it exists as the `thread_id` key the other records hang off — and +its own; it exists as the `thread_id` key the other records hang off. And `metadata` is independent of all of it (its identity is `(namespace, key)`). Note the asymmetry on the generation side. A chat run belongs to a thread, and its own `run_id` is secondary. A generation run is keyed by its own `run_id` @@ -138,11 +145,11 @@ first, and its `thread_id` names the slot the run fills, which is what ```mermaid erDiagram - MESSAGES ||--o{ RUN : "thread_id — a thread has many runs" - RUN ||--o{ INTERRUPT : "run_id — a run may pause on interrupts" + MESSAGES ||--o{ RUN : "thread_id, a thread has many runs" + RUN ||--o{ INTERRUPT : "run_id, a run may pause on interrupts" MESSAGES ||..o{ GENERATION_RUN : "thread_id, the slot a run fills" - GENERATION_RUN ||--o{ ARTIFACT : "run_id — a run produces artifacts" - ARTIFACT ||--|| BLOB : "blob_key — the bytes" + GENERATION_RUN ||--o{ ARTIFACT : "run_id, a run produces artifacts" + ARTIFACT ||--|| BLOB : "blob_key, the bytes" MESSAGES { string thread_id PK @@ -181,955 +188,11 @@ erDiagram } ``` -## New database: a SQLite adapter start to finish - -### 1. The schema - -Four tables. JSON payloads are stored as text (SQLite has no JSON column type), -timestamps as integers (epoch milliseconds), everything keyed the way the store -methods look records up. - -```sql -CREATE TABLE IF NOT EXISTS messages ( - thread_id text PRIMARY KEY NOT NULL, - messages_json text NOT NULL -); -CREATE TABLE IF NOT EXISTS runs ( - run_id text PRIMARY KEY NOT NULL, - thread_id text NOT NULL, - status text NOT NULL, - started_at integer NOT NULL, - finished_at integer, - error text, - usage_json text -); -CREATE TABLE IF NOT EXISTS interrupts ( - interrupt_id text PRIMARY KEY NOT NULL, - run_id text NOT NULL, - thread_id text NOT NULL, - status text NOT NULL, - requested_at integer NOT NULL, - resolved_at integer, - payload_json text NOT NULL, - response_json text -); -CREATE TABLE IF NOT EXISTS metadata ( - scope text NOT NULL, - key text NOT NULL, - value_json text NOT NULL, - PRIMARY KEY (scope, key) -); -``` - -### 2. Messages: full-transcript overwrite - -Two contracts to hold: - -- `saveThread` always receives the complete, authoritative history. It is a - replace, not an append. -- `loadThread` returns `[]` for a thread that was never saved, never `null`. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineMessageStore } from '@tanstack/ai-persistence' -import type { ModelMessage } from '@tanstack/ai' - -// `defineMessageStore` types the object inline against the contract — you get -// autocomplete and checking with no separate `: MessageStore` annotation. -function createMessageStore(db: DatabaseSync) { - const select = db.prepare( - 'SELECT messages_json FROM messages WHERE thread_id = ?', - ) - const upsert = db.prepare( - `INSERT INTO messages (thread_id, messages_json) VALUES (?, ?) - ON CONFLICT(thread_id) DO UPDATE SET messages_json = excluded.messages_json`, - ) - return defineMessageStore({ - async loadThread(threadId) { - const json = select.get(threadId)?.messages_json - // Unknown thread → [] (never null). `node:sqlite` types columns as a - // SQL-value union, so narrow to string before parsing (no cast). - if (typeof json !== 'string') return [] - const parsed: Array = JSON.parse(json) - return parsed - }, - async saveThread(threadId, messages) { - upsert.run(threadId, JSON.stringify(messages)) - }, - }) -} -``` - -The methods are `async`, so `node:sqlite` (a synchronous driver) needs no -`Promise.resolve` wrapper: `async` promotes the returned value to a promise, and -a method that returns nothing resolves to `void`. On an async driver, `await` the -query instead. - -### 3. Runs: idempotent create, patch, get - -Two contracts to hold: - -- `createOrResume` must be idempotent. If the run id already exists, return the - stored record unchanged, so resuming a run never resets its `startedAt` or - status. `INSERT ... ON CONFLICT DO NOTHING` gives you that in one statement. -- `update` on an unknown run id is a no-op. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineRunStore } from '@tanstack/ai-persistence' -import type { RunRecord, RunStatus } from '@tanstack/ai-persistence' - -// The `status` column is text; validate it back into the union (no cast). -function toRunStatus(value: unknown): RunStatus { - switch (value) { - case 'running': - case 'completed': - case 'failed': - case 'interrupted': - return value - default: - throw new TypeError(`Unexpected run status: ${String(value)}`) - } -} - -// `node:sqlite` types columns as a SQL-value union, so coerce/narrow each field -// (String / Number / typeof) rather than casting the whole row. -function mapRun(row: Record): RunRecord { - return { - runId: String(row.run_id), - threadId: String(row.thread_id), - status: toRunStatus(row.status), - startedAt: Number(row.started_at), - ...(row.finished_at != null ? { finishedAt: Number(row.finished_at) } : {}), - ...(typeof row.error === 'string' ? { error: row.error } : {}), - ...(typeof row.usage_json === 'string' - ? { usage: JSON.parse(row.usage_json) } - : {}), - } -} - -function createRunStore(db: DatabaseSync) { - const select = db.prepare('SELECT * FROM runs WHERE run_id = ?') - const insert = db.prepare( - `INSERT INTO runs (run_id, thread_id, status, started_at) VALUES (?, ?, ?, ?) - ON CONFLICT(run_id) DO NOTHING`, - ) - const active = db.prepare( - `SELECT * FROM runs WHERE thread_id = ? AND status = 'running' - ORDER BY started_at DESC LIMIT 1`, - ) - return defineRunStore({ - async createOrResume(input) { - const existing = select.get(input.runId) - if (existing) return mapRun(existing) - const status: RunStatus = input.status ?? 'running' - insert.run(input.runId, input.threadId, status, input.startedAt) - return { - runId: input.runId, - threadId: input.threadId, - status, - startedAt: input.startedAt, - } - }, - async update(runId, patch) { - const sets: Array = [] - const params: Array = [] - if (patch.status !== undefined) { - sets.push('status = ?') - params.push(patch.status) - } - if (patch.finishedAt !== undefined) { - sets.push('finished_at = ?') - params.push(patch.finishedAt) - } - if (patch.error !== undefined) { - sets.push('error = ?') - params.push(patch.error) - } - if (patch.usage !== undefined) { - sets.push('usage_json = ?') - params.push(JSON.stringify(patch.usage)) - } - if (sets.length === 0) return - params.push(runId) - db.prepare(`UPDATE runs SET ${sets.join(', ')} WHERE run_id = ?`).run( - ...params, - ) - }, - async get(runId) { - const row = select.get(runId) - return row ? mapRun(row) : null - }, - // The most recent still-running run for a thread. `reconstructChat` calls - // this so a hydrating client (a reload, another device, or switching back to - // a generating thread) learns there is a live run and tails it. Stub it to - // null and the thread always looks idle on hydrate: the transcript restores, - // but a reply that was mid-stream never resumes. - async findActiveRun(threadId) { - const row = active.get(threadId) - return row ? mapRun(row) : null - }, - }) -} -``` - -`update` builds its `SET` list from only the fields present in the patch, so an -empty patch touches nothing and a partial patch leaves other columns alone. Map -each row back with a small helper that omits absent optional fields and parses -the JSON columns. - -### 4. Interrupts: insert-if-absent, ordered listings - -`create` is insert-if-absent: a duplicate interrupt id must never overwrite an -interrupt that was already resolved. Every `list*` method returns records ordered -by `requested_at` ascending, which the middleware relies on. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineInterruptStore } from '@tanstack/ai-persistence' -import type { - InterruptRecord, - InterruptStatus, -} from '@tanstack/ai-persistence' - -function toInterruptStatus(value: unknown): InterruptStatus { - switch (value) { - case 'pending': - case 'resolved': - case 'cancelled': - return value - default: - throw new TypeError(`Unexpected interrupt status: ${String(value)}`) - } -} - -function mapInterrupt(row: Record): InterruptRecord { - return { - interruptId: String(row.interrupt_id), - runId: String(row.run_id), - threadId: String(row.thread_id), - status: toInterruptStatus(row.status), - requestedAt: Number(row.requested_at), - ...(row.resolved_at != null ? { resolvedAt: Number(row.resolved_at) } : {}), - payload: - typeof row.payload_json === 'string' ? JSON.parse(row.payload_json) : {}, - ...(typeof row.response_json === 'string' - ? { response: JSON.parse(row.response_json) } - : {}), - } -} - -function createInterruptStore(db: DatabaseSync) { - const insert = db.prepare( - `INSERT INTO interrupts - (interrupt_id, run_id, thread_id, status, requested_at, payload_json, response_json) - VALUES (?, ?, ?, 'pending', ?, ?, ?) - ON CONFLICT(interrupt_id) DO NOTHING`, - ) - const resolveRow = db.prepare( - `UPDATE interrupts SET status = 'resolved', resolved_at = ?, response_json = ? - WHERE interrupt_id = ?`, - ) - const cancelRow = db.prepare( - `UPDATE interrupts SET status = 'cancelled', resolved_at = ? WHERE interrupt_id = ?`, - ) - const selectOne = db.prepare('SELECT * FROM interrupts WHERE interrupt_id = ?') - // Every listing is ORDER BY requested_at ASC — the middleware relies on it. - const byThread = db.prepare( - 'SELECT * FROM interrupts WHERE thread_id = ? ORDER BY requested_at ASC', - ) - const pendingByThread = db.prepare( - `SELECT * FROM interrupts WHERE thread_id = ? AND status = 'pending' - ORDER BY requested_at ASC`, - ) - const byRun = db.prepare( - 'SELECT * FROM interrupts WHERE run_id = ? ORDER BY requested_at ASC', - ) - const pendingByRun = db.prepare( - `SELECT * FROM interrupts WHERE run_id = ? AND status = 'pending' - ORDER BY requested_at ASC`, - ) - return defineInterruptStore({ - async create(record) { - // Insert-if-absent: a duplicate id must never clobber an already-resolved - // interrupt back to pending. - insert.run( - record.interruptId, - record.runId, - record.threadId, - record.requestedAt, - JSON.stringify(record.payload), - record.response === undefined ? null : JSON.stringify(record.response), - ) - }, - async resolve(interruptId, response) { - resolveRow.run( - Date.now(), - response === undefined ? null : JSON.stringify(response), - interruptId, - ) - }, - async cancel(interruptId) { - cancelRow.run(Date.now(), interruptId) - }, - async get(interruptId) { - const row = selectOne.get(interruptId) - return row ? mapInterrupt(row) : null - }, - async list(threadId) { - return byThread.all(threadId).map(mapInterrupt) - }, - async listPending(threadId) { - return pendingByThread.all(threadId).map(mapInterrupt) - }, - async listByRun(runId) { - return byRun.all(runId).map(mapInterrupt) - }, - async listPendingByRun(runId) { - return pendingByRun.all(runId).map(mapInterrupt) - }, - }) -} -``` - -### 5. Metadata: reject nullish - -`(scope, key)` is the composite identity. A SQL backend cannot store a nullish -value in a `NOT NULL` text column, so reject `null` and `undefined` with a clear -error instead of a cryptic driver failure. Callers clear a value with `delete`. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineMetadataStore } from '@tanstack/ai-persistence' - -function createMetadataStore(db: DatabaseSync) { - const select = db.prepare( - 'SELECT value_json FROM metadata WHERE scope = ? AND key = ?', - ) - const upsert = db.prepare( - `INSERT INTO metadata (scope, key, value_json) VALUES (?, ?, ?) - ON CONFLICT(scope, key) DO UPDATE SET value_json = excluded.value_json`, - ) - return defineMetadataStore({ - async get(scope, key) { - const json = select.get(scope, key)?.value_json - return typeof json === 'string' ? JSON.parse(json) : null - }, - async set(scope, key, value) { - if (value == null) { - throw new TypeError( - 'Metadata values must be defined, non-null JSON. Use delete() to clear.', - ) - } - upsert.run(scope, key, JSON.stringify(value)) - }, - async delete(scope, key) { - db.prepare('DELETE FROM metadata WHERE scope = ? AND key = ?').run( - scope, - key, - ) - }, - }) -} -``` - -### 6. Assemble the adapter - -Open the database, create the tables, and return the stores as an -`AIPersistence`. `defineAIPersistence` keeps the exact store keys in the type and -rejects unknown keys at runtime. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineAIPersistence } from '@tanstack/ai-persistence' -import type { ChatPersistence } from '@tanstack/ai-persistence' -// The four store factories and the schema string, each from your own module. -import { createInterruptStore } from './interrupt-store' -import { createMessageStore } from './message-store' -import { createMetadataStore } from './metadata-store' -import { createRunStore } from './run-store' -import { SCHEMA_SQL } from './schema' - -export function sqlitePersistence(options: { - url: string - migrate?: boolean -}): ChatPersistence { - const db = new DatabaseSync(options.url) - if (options.migrate) db.exec(SCHEMA_SQL) - return defineAIPersistence({ - stores: { - messages: createMessageStore(db), - runs: createRunStore(db), - interrupts: createInterruptStore(db), - metadata: createMetadataStore(db), - }, - }) -} -``` - -That is a complete backend. If you also need a mutex across workers, add -`withLocks` alongside it; see [Locks](../advanced/locks). - -Wire it into `chat()` exactly like any other persistence: - -```ts -import { - chat, - chatParamsFromRequest, - toServerSentEventsResponse, -} from '@tanstack/ai' -import { openaiText } from '@tanstack/ai-openai' -import { withPersistence } from '@tanstack/ai-persistence' -import { persistence } from './persistence' - -export async function POST(request: Request) { - const params = await chatParamsFromRequest(request) - const stream = chat({ - adapter: openaiText('gpt-5.5'), - messages: params.messages, - threadId: params.threadId, - runId: params.runId, - ...(params.resume ? { resume: params.resume } : {}), - middleware: [withPersistence(persistence)], - }) - return toServerSentEventsResponse(stream) -} -``` - -## Generation & media stores - -Everything above builds a **chat** adapter. -[Media generation](./generation-persistence) persists differently: it does not -use the chat `runs` store at all. - -- **Required:** a `generationRuns` store, a `GenerationRunStore` keyed by - `runId` (the run/request id a generation mints). It is the counterpart to - `runs`. -- **Optional, to keep the generated bytes:** an `artifacts` store (metadata) and - a `blobs` store (the bytes). These two must be provided **together**. - -`threadId` is the slot the run belongs to, recorded on each run record. - -These are three more tables alongside the four from the schema in step 1: - -```sql -CREATE TABLE IF NOT EXISTS generation_runs ( - run_id text PRIMARY KEY NOT NULL, - thread_id text NOT NULL, - activity text NOT NULL, - provider text NOT NULL, - model text NOT NULL, - status text NOT NULL, - started_at integer NOT NULL, - finished_at integer, - error_json text, - result_json text, - artifacts_json text, - usage_json text -); -CREATE TABLE IF NOT EXISTS 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 TABLE IF NOT EXISTS blobs ( - key text PRIMARY KEY NOT NULL, - bytes blob NOT NULL, - size integer NOT NULL, - etag text NOT NULL, - content_type text, - custom_metadata_json text, - created_at integer NOT NULL, - updated_at integer NOT NULL -); -``` - -### Generation runs: idempotent create, patch, latest-for-thread - -`GenerationRunStore` is the generation analogue of `RunStore`. Three contracts -to hold: - -- `createOrResume` is idempotent. A second call for a `runId` returns the stored - record unchanged, so resuming a run never resets its `startedAt`, `activity`, - or status. `INSERT ... ON CONFLICT DO NOTHING` gives you that. -- `update` on an unknown `runId` is a no-op. -- `findLatestForThread` returns the run with the greatest `startedAt` linked to a - thread. `reconstructGeneration` calls it to hydrate the last generation for a - thread on a server-driven client's mount. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineGenerationRunStore } from '@tanstack/ai-persistence' -import type { - GenerationRunRecord, - GenerationRunStatus, -} from '@tanstack/ai-persistence' - -function toGenerationRunStatus(value: unknown): GenerationRunStatus { - switch (value) { - case 'running': - case 'completed': - case 'failed': - case 'interrupted': - return value - default: - throw new TypeError(`Unexpected generation run status: ${String(value)}`) - } -} - -// `node:sqlite` types columns as a SQL-value union, so coerce/narrow each field -// (String / Number / typeof) and JSON-parse the text columns — no cast. -function mapGenerationRun(row: Record): GenerationRunRecord { - return { - runId: String(row.run_id), - threadId: String(row.thread_id), - activity: String(row.activity), - provider: String(row.provider), - model: String(row.model), - status: toGenerationRunStatus(row.status), - startedAt: Number(row.started_at), - ...(row.finished_at != null ? { finishedAt: Number(row.finished_at) } : {}), - ...(typeof row.error_json === 'string' - ? { error: JSON.parse(row.error_json) } - : {}), - ...(typeof row.result_json === 'string' - ? { result: JSON.parse(row.result_json) } - : {}), - ...(typeof row.artifacts_json === 'string' - ? { artifacts: JSON.parse(row.artifacts_json) } - : {}), - ...(typeof row.usage_json === 'string' - ? { usage: JSON.parse(row.usage_json) } - : {}), - } -} - -function createGenerationRunStore(db: DatabaseSync) { - const select = db.prepare('SELECT * FROM generation_runs WHERE run_id = ?') - const insert = db.prepare( - `INSERT INTO generation_runs - (run_id, thread_id, activity, provider, model, status, started_at) - VALUES (?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(run_id) DO NOTHING`, - ) - const latest = db.prepare( - `SELECT * FROM generation_runs WHERE thread_id = ? - ORDER BY started_at DESC LIMIT 1`, - ) - return defineGenerationRunStore({ - async createOrResume(input) { - const existing = select.get(input.runId) - if (existing) return mapGenerationRun(existing) - const status: GenerationRunStatus = input.status ?? 'running' - insert.run( - input.runId, - input.threadId, - input.activity, - input.provider, - input.model, - status, - input.startedAt, - ) - return { - runId: input.runId, - threadId: input.threadId, - activity: input.activity, - provider: input.provider, - model: input.model, - status, - startedAt: input.startedAt, - } - }, - async update(runId, patch) { - const sets: Array = [] - const params: Array = [] - if (patch.status !== undefined) { - sets.push('status = ?') - params.push(patch.status) - } - if (patch.finishedAt !== undefined) { - sets.push('finished_at = ?') - params.push(patch.finishedAt) - } - if (patch.error !== undefined) { - sets.push('error_json = ?') - params.push(JSON.stringify(patch.error)) - } - if (patch.result !== undefined) { - sets.push('result_json = ?') - params.push(JSON.stringify(patch.result)) - } - if (patch.artifacts !== undefined) { - sets.push('artifacts_json = ?') - params.push(JSON.stringify(patch.artifacts)) - } - if (patch.usage !== undefined) { - sets.push('usage_json = ?') - params.push(JSON.stringify(patch.usage)) - } - // Empty patch, or an unknown run id, touches nothing (UPDATE no-ops). - if (sets.length === 0) return - params.push(runId) - db.prepare( - `UPDATE generation_runs SET ${sets.join(', ')} WHERE run_id = ?`, - ).run(...params) - }, - async get(runId) { - const row = select.get(runId) - return row ? mapGenerationRun(row) : null - }, - // The most recent run linked to a thread. `reconstructGeneration` calls this - // so a server-driven client (`persistence: true`) hydrates the last - // generation for its thread by the stable thread id, without a run id. - async findLatestForThread(threadId) { - const row = latest.get(threadId) - return row ? mapGenerationRun(row) : null - }, - }) -} -``` - -### Artifacts: media metadata - -`ArtifactStore` holds one metadata row per generated file: its `runId`, -`mimeType`, `size`, and a `createdAt`. The bytes live in the blob store below. - -- `save` is an upsert. -- `list(runId)` returns every artifact for a run, `[]` when there are none. -- `delete` / `deleteForRun` are required. Retention and erasure are the point of - storing media durably, and they mirror `BlobStore.delete`. - -Persist `blobKey` verbatim. It records where these bytes actually went, and a -`storageKey` mapper can put them anywhere, so a reader cannot recompute the -path — `resolveArtifactBlobKey(record)` falls back to the default convention -only for rows written before the column existed. Drop it and every artifact -stored under a custom key becomes unreadable. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineArtifactStore } from '@tanstack/ai-persistence' -import type { ArtifactRecord } from '@tanstack/ai-persistence' - -function mapArtifact(row: Record): ArtifactRecord { - return { - artifactId: String(row.artifact_id), - runId: String(row.run_id), - threadId: String(row.thread_id), - ...(typeof row.blob_key === 'string' ? { blobKey: row.blob_key } : {}), - name: String(row.name), - mimeType: String(row.mime_type), - size: Number(row.size), - ...(typeof row.source_url === 'string' - ? { sourceUrl: row.source_url } - : {}), - createdAt: Number(row.created_at), - } -} - -function createArtifactStore(db: DatabaseSync) { - const upsert = db.prepare( - `INSERT INTO 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`, - ) - const selectOne = db.prepare('SELECT * FROM artifacts WHERE artifact_id = ?') - const byRun = db.prepare( - 'SELECT * FROM artifacts WHERE run_id = ? ORDER BY created_at ASC', - ) - return defineArtifactStore({ - async save(record) { - upsert.run( - record.artifactId, - record.runId, - record.threadId, - record.blobKey ?? null, - record.name, - record.mimeType, - record.size, - record.sourceUrl ?? null, - record.createdAt, - ) - }, - async get(artifactId) { - const row = selectOne.get(artifactId) - return row ? mapArtifact(row) : null - }, - async list(runId) { - return byRun.all(runId).map(mapArtifact) - }, - async delete(artifactId) { - db.prepare('DELETE FROM artifacts WHERE artifact_id = ?').run(artifactId) - }, - async deleteForRun(runId) { - db.prepare('DELETE FROM artifacts WHERE run_id = ?').run(runId) - }, - }) -} -``` - -### Blobs: the bytes - -`BlobStore` is a small object store. `withGenerationPersistence` writes each -generated file under the key `artifacts//`, so a -prefix-filtered `list({ prefix: 'artifacts//' })` enumerates a run's -media. - -- `put` accepts any `BlobBody`: a stream, buffer, string, or `Blob`. The helper - below normalizes it to bytes. -- `list` matches `prefix` literally and pages with a keyset cursor. - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineBlobStore, resolveBlobRange } from '@tanstack/ai-persistence' -import type { - BlobBody, - BlobObject, - BlobRecord, -} from '@tanstack/ai-persistence' - -async function toBytes(body: BlobBody): Promise { - if (typeof body === 'string') return new TextEncoder().encode(body) - if (body instanceof ArrayBuffer) return new Uint8Array(body.slice(0)) - if (ArrayBuffer.isView(body)) { - return new Uint8Array(body.buffer, body.byteOffset, body.byteLength).slice() - } - if (body instanceof Blob) { - return new Uint8Array(await body.arrayBuffer()) - } - // ReadableStream: drain it into one buffer. - const reader = body.getReader() - const chunks: Array = [] - let total = 0 - for (;;) { - const { done, value } = await reader.read() - if (done) break - chunks.push(value) - total += value.byteLength - } - const bytes = new Uint8Array(total) - let offset = 0 - for (const chunk of chunks) { - bytes.set(chunk, offset) - offset += chunk.byteLength - } - return bytes -} - -function mapBlobRecord(row: Record): BlobRecord { - return { - key: String(row.key), - ...(row.size != null ? { size: Number(row.size) } : {}), - ...(typeof row.etag === 'string' ? { etag: row.etag } : {}), - ...(typeof row.content_type === 'string' - ? { contentType: row.content_type } - : {}), - ...(typeof row.custom_metadata_json === 'string' - ? { customMetadata: JSON.parse(row.custom_metadata_json) } - : {}), - ...(row.created_at != null ? { createdAt: Number(row.created_at) } : {}), - ...(row.updated_at != null ? { updatedAt: Number(row.updated_at) } : {}), - } -} - -function blobObject( - record: BlobRecord, - bytes: Uint8Array, - range?: { offset: number; length: number }, -): BlobObject { - return { - ...record, - // `size` keeps describing the whole object; `range` describes these bytes. - ...(range ? { range } : {}), - body: new ReadableStream({ - start(controller) { - controller.enqueue(bytes.slice()) - controller.close() - }, - }), - arrayBuffer() { - const copy = new ArrayBuffer(bytes.byteLength) - new Uint8Array(copy).set(bytes) - return Promise.resolve(copy) - }, - text: () => Promise.resolve(new TextDecoder().decode(bytes)), - } -} - -function createBlobStore(db: DatabaseSync) { - const upsert = db.prepare( - `INSERT INTO blobs - (key, bytes, size, etag, content_type, custom_metadata_json, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(key) DO UPDATE SET - bytes = excluded.bytes, size = excluded.size, etag = excluded.etag, - content_type = excluded.content_type, - custom_metadata_json = excluded.custom_metadata_json, - updated_at = excluded.updated_at`, - ) - const selectCreated = db.prepare('SELECT created_at FROM blobs WHERE key = ?') - const selectOne = db.prepare('SELECT * FROM blobs WHERE key = ?') - // Metadata without the bytes, and the bounded slice: a ranged read must not - // load the whole object to hand back a piece of it. - const selectMeta = db.prepare( - `SELECT key, size, etag, content_type, custom_metadata_json, - created_at, updated_at - FROM blobs WHERE key = ?`, - ) - const selectSlice = db.prepare( - 'SELECT substr(bytes, ?, ?) AS bytes FROM blobs WHERE key = ?', - ) - return defineBlobStore({ - async put(key, body, options) { - const bytes = await toBytes(body) - const now = Date.now() - const prior = selectCreated.get(key) - const createdAt = - prior && prior.created_at != null ? Number(prior.created_at) : now - const etag = String(now) - upsert.run( - key, - bytes, - bytes.byteLength, - etag, - options?.contentType ?? null, - options?.customMetadata ? JSON.stringify(options.customMetadata) : null, - createdAt, - now, - ) - return { - key, - size: bytes.byteLength, - etag, - createdAt, - updatedAt: now, - ...(options?.contentType !== undefined - ? { contentType: options.contentType } - : {}), - ...(options?.customMetadata !== undefined - ? { customMetadata: options.customMetadata } - : {}), - } - }, - async get(key, options) { - if (!options?.range) { - const row = selectOne.get(key) - if (!row) return null - const bytes = - row.bytes instanceof Uint8Array ? row.bytes : new Uint8Array() - return blobObject(mapBlobRecord(row), bytes) - } - // Metadata first, WITHOUT the bytes, so the clamp costs no I/O... - const meta = selectMeta.get(key) - if (!meta) return null - const served = resolveBlobRange(Number(meta.size), options.range) - // ...then let SQLite cut the slice (`substr` is 1-based and byte-wise - // over a BLOB). Reading the row whole and slicing in JS would load the - // entire object on every video seek — the cost ranges exist to avoid. - const slice = selectSlice.get(served.offset + 1, served.length, key) - if (!slice) return null - const bytes = - slice.bytes instanceof Uint8Array ? slice.bytes : new Uint8Array() - return blobObject(mapBlobRecord(meta), bytes, served) - }, - async head(key) { - // Metadata only: never pull the bytes to answer a question about them. - const row = selectMeta.get(key) - return row ? mapBlobRecord(row) : null - }, - async delete(key) { - db.prepare('DELETE FROM blobs WHERE key = ?').run(key) - }, - async list(options) { - if (options?.limit === 0) return { objects: [], truncated: false } - // Match the prefix with `substr(...) = ?` rather than LIKE: SQLite's LIKE - // is case-INsensitive for ASCII and treats `%`/`_` as wildcards, while the - // contract says a prefix matches literally and case-sensitively. Then page - // with a keyset cursor (keys strictly greater than the last one returned). - const prefix = options?.prefix ?? '' - const params: Array = [prefix, prefix] - let where = 'substr(key, 1, length(?)) = ?' - if (options?.cursor !== undefined) { - where += ' AND key > ?' - params.push(options.cursor) - } - let sql = `SELECT * FROM blobs WHERE ${where} ORDER BY key ASC` - const limit = options?.limit - if (limit !== undefined) { - sql += ' LIMIT ?' // fetch one extra row to detect truncation - params.push(limit + 1) - } - const rows = db - .prepare(sql) - .all(...params) - .map(mapBlobRecord) - if (limit !== undefined && rows.length > limit) { - const page = rows.slice(0, limit) - const cursor = page.at(-1)?.key - return { - objects: page, - truncated: true, - ...(cursor !== undefined ? { cursor } : {}), - } - } - return { objects: rows, truncated: false } - }, - }) -} -``` - -### Assemble a generation adapter - -Hand the three stores to `defineAIPersistence` the same way. `generationRuns` alone is a -valid generation adapter (run records, no byte storage); add `artifacts` + -`blobs` — together — to keep the media: - -```ts -import { DatabaseSync } from 'node:sqlite' -import { defineAIPersistence } from '@tanstack/ai-persistence' -// The three generation store factories and the schema string, from your modules. -import { createArtifactStore } from './artifact-store' -import { createBlobStore } from './blob-store' -import { createGenerationRunStore } from './generation-run-store' -import { GENERATION_SCHEMA_SQL } from './generation-schema' - -export function generationPersistence(options: { - url: string - migrate?: boolean -}) { - const db = new DatabaseSync(options.url) - if (options.migrate) db.exec(GENERATION_SCHEMA_SQL) - return defineAIPersistence({ - stores: { - generationRuns: createGenerationRunStore(db), - artifacts: createArtifactStore(db), - blobs: createBlobStore(db), - }, - }) -} -``` - -Pass the result to `withGenerationPersistence` on a `generateImage` / -`generateVideo` / … call; see [Generation persistence](./generation-persistence). -You can also fold these stores into an existing chat adapter with -`composePersistence`, so one backend serves both `withPersistence` and -`withGenerationPersistence`. - ## Existing database: map the contracts onto your schema -You do not have to create the four tables above. If you already have a database, -map each store method onto the tables and columns you already run. Three things -change from the from-scratch version. +You do not have to create tables at all. If you already have a database, map each +store method onto the tables and columns you already run. Three things change from +the from-scratch walkthroughs. **Your column names, your types.** The core reads and writes only through your store methods, so name columns whatever you like and use your database's native @@ -1143,7 +206,7 @@ threads per user, add `created_at`/`updated_at` audit columns, add a tenant id. Keep added columns nullable or defaulted so the store's inserts still succeed. The TanStack AI stores never read or write columns they do not know about. -**Adopt part of it.** You rarely need all four stores in the same database. Put +**Adopt part of it.** You rarely need every store in the same database. Put `messages` and `runs` in your primary database and nothing else, then fill the rest from another source with `composePersistence`: @@ -1179,9 +242,9 @@ runPersistenceConformance('my sqlite adapter', () => ) ``` -The suite covers all seven stores — the four chat state stores and the three -generation stores from the section above — so an adapter lists whatever it -deliberately omits. A chat-only adapter skips the generation half: +The suite covers all seven stores, the four chat state stores and the three +generation stores, so an adapter lists whatever it deliberately omits. A +chat-only adapter skips the generation half: ```ts import { runPersistenceConformance } from '@tanstack/ai-persistence/testkit' @@ -1222,12 +285,12 @@ exactly this test against its SQLite backend, which provides all seven. ## Let your coding agent write it -You do not have to type this page out. `@tanstack/ai-persistence` ships +You do not have to write any of this by hand. `@tanstack/ai-persistence` ships [Agent Skills](../getting-started/agent-skills) that turn it into a recipe your assistant follows against **your** stack: it reads your existing ORM config, schema file, and database handle, appends the four tables to the schema you already have, and writes a single `src/lib/chat-persistence.ts` exporting the -`ChatPersistence` — no new package, no second database client, and no migration +`ChatPersistence`. No new package, no second database client, and no migration mechanism competing with the one you run. Install the skills with [TanStack Intent](https://tanstack.com/intent/latest/docs/overview), @@ -1239,21 +302,21 @@ pnpm add @tanstack/ai-persistence npx @tanstack/intent@latest install ``` -Then ask for what you want — "add chat persistence to this app" — and the +Then ask for what you want ("add chat persistence to this app") and the matching skill loads itself into context: | Skill | Covers | | ----------------------------------------- | ------------------------------------------------------------------- | -| `ai-persistence` | Entry point — routes to everything below | +| `ai-persistence` | Entry point, routes to everything below | | `ai-persistence/server` | `withPersistence`, run lifecycle, interrupts, `reconstructChat` | | `ai-persistence/stores` | The store contracts and their invariants | | `ai-core/locks` | `LockStore` / `withLocks` coordination (ships in `@tanstack/ai/locks`) | | `ai-persistence/build-drizzle-adapter` | `chat-persistence.ts` for a Drizzle app (SQLite / Postgres / MySQL) | | `ai-persistence/build-prisma-adapter` | `chat-persistence.ts` for a Prisma app | | `ai-persistence/build-cloudflare-adapter` | `chat-persistence.ts` for a Worker on D1, plus Durable Object locks | -| `ai-persistence/build-custom-adapter` | `chat-persistence.ts` for anything else — raw `pg`, Kysely, SQLite, Mongo, Supabase | +| `ai-persistence/build-custom-adapter` | `chat-persistence.ts` for anything else: raw `pg`, Kysely, SQLite, Mongo, Supabase | -Browser-side persistence is not in this package — its skill ships with +Browser-side persistence is not in this package. Its skill ships with `@tanstack/ai` as `ai-core/client-persistence`, alongside the framework code it teaches. @@ -1261,338 +324,11 @@ They are plain Markdown at `node_modules/@tanstack/ai-persistence/skills//SKILL.md` if you prefer to read or follow them yourself. -## Store interface reference - -These are the public contracts from `@tanstack/ai-persistence`. Implement only -the stores you need. - -### MessageStore - -```ts -import type { ModelMessage } from '@tanstack/ai' - -interface MessageStore { - loadThread(threadId: string): Promise> - saveThread(threadId: string, messages: Array): Promise -} -``` - -`saveThread` receives the full authoritative model-message history, not a delta. -`loadThread` returns `[]` (never `null`) for a thread that was never saved. - -### RunStore - -```ts -import type { TokenUsage } from '@tanstack/ai' - -interface RunRecord { - runId: string - threadId: string - status: 'running' | 'completed' | 'failed' | 'interrupted' - startedAt: number // epoch ms - finishedAt?: number // epoch ms, set once the run reaches a terminal status - error?: string - usage?: TokenUsage // token counts, from @tanstack/ai -} - -interface RunStore { - createOrResume(input: { - runId: string - threadId: string - status?: RunRecord['status'] - startedAt: number - }): Promise - update( - runId: string, - patch: Partial< - Pick - >, - ): Promise - get(runId: string): Promise - // The most recent 'running' run for a thread (greatest `startedAt` wins), or - // null when the thread is idle. `reconstructChat` calls it to report - // `activeRun`, which is how a hydrating client tails a run that is still - // generating. - findActiveRun(threadId: string): Promise -} -``` - -Three contracts to hold: - -- `createOrResume` must be idempotent. A second call for an existing `runId` - returns the stored record unchanged, which is what makes resuming a run safe. - Retries may repeat the same run id. -- `update` against an unknown `runId` is a no-op. -- `findActiveRun` must do real work. Stub it to `null` and `reconstructChat` - always reports `activeRun: null`, so a client that reloads (or switches back - to) a still-generating thread restores the transcript but never resumes the - live reply. Nothing detects it either, because `null` is also the right answer - for an idle thread. - -Every method on a store you provide is required. A backend that genuinely has no -run lifecycle should declare `ChatTranscriptStores` and omit `runs` entirely -rather than supply a `RunStore` with a stubbed method: an absent store is caught -by the type system, an incomplete one fails silently at runtime. - -### InterruptStore - -```ts -interface InterruptRecord { - interruptId: string - runId: string - threadId: string - status: 'pending' | 'resolved' | 'cancelled' - requestedAt: number // epoch ms - resolvedAt?: number // epoch ms, set once resolved or cancelled - payload: Record - response?: unknown -} - -interface InterruptStore { - create(record: Omit): Promise - resolve(interruptId: string, response?: unknown): Promise - cancel(interruptId: string): Promise - get(interruptId: string): Promise - list(threadId: string): Promise> - listPending(threadId: string): Promise> - listByRun(runId: string): Promise> - listPendingByRun(runId: string): Promise> -} -``` - -`create` accepts a record without `status`/`resolvedAt` so every interrupt is -born `'pending'`; it is insert-if-absent, so a duplicate `create` never clobbers -an already-resolved interrupt. The `list*` methods return records ordered by -`requestedAt` ascending. An `interrupts` store requires a `runs` store when used -with chat persistence. - -### MetadataStore - -```ts -interface MetadataStore { - get(scope: string, key: string): Promise - set(scope: string, key: string, value: unknown): Promise - delete(scope: string, key: string): Promise -} -``` - -Namespaces and value schemas are application-owned, and `(scope, key)` is the -composite identity. A stored `null` is indistinguishable from absence at the type -level, so wrap a value you must persist as `null` (e.g. `{ value: null }`), or -reject nullish values outright the way the SQLite store above does. - -### GenerationRunStore - -The generation counterpart to `RunStore`. Keyed by its own `runId`, with -`threadId` the slot `findLatestForThread` looks runs up by. -`withGenerationPersistence` requires this store, not `runs`. - -Its `status` uses the same vocabulary as a chat run's `RunStatus`, so one status -column and one set of checks cover both tables. - -```ts -import type { PersistedArtifactRef, TokenUsage } from '@tanstack/ai' - -// The same vocabulary as a chat run's `RunStatus`. -type GenerationRunStatus = 'running' | 'completed' | 'failed' | 'interrupted' - -interface GenerationRunRecord { - runId: string - threadId: string // the slot this run fills, hydrated by findLatestForThread - activity: string // 'image' | 'audio' | 'tts' | 'video' | 'transcription' - provider: string - model: string - status: GenerationRunStatus - startedAt: number // epoch ms - finishedAt?: number // epoch ms, set once the run reaches a terminal status - error?: { message: string; code?: string } - result?: unknown // terminal result metadata (ids, urls) — never media bytes - artifacts?: Array // present with an artifacts + blobs backend - usage?: TokenUsage -} - -interface GenerationRunStore { - createOrResume(input: { - runId: string - activity: string - provider: string - model: string - startedAt: number - threadId: string - status?: GenerationRunStatus - }): Promise - update( - runId: string, - patch: Partial< - Pick< - GenerationRunRecord, - 'status' | 'finishedAt' | 'error' | 'result' | 'artifacts' | 'usage' - > - >, - ): Promise - get(runId: string): Promise - // The most recent run filed under a thread (greatest `startedAt`), or null. - // Required: it is the only query that hydrates a generation, so an adapter - // without it would be indistinguishable from one whose thread has no runs — - // `persistence: true` would silently restore nothing, forever. - findLatestForThread(threadId: string): Promise -} -``` - -Implement `createOrResume` idempotently: a second call for an existing `runId` -returns the stored record unchanged (`startedAt` / `activity` / `provider` / -`model` / `threadId` are not mutated), which is what makes resuming a run safe. -`update` against an unknown `runId` is a no-op. - -### ArtifactStore - -Metadata rows for persisted media. The bytes live in a `BlobStore`; this record -holds the descriptive metadata and an optional `sourceUrl` for reference-only -backends. Provide it together with a `BlobStore` to keep generated bytes. - -```ts -interface ArtifactRecord { - artifactId: string - runId: string - threadId: string - blobKey?: string // where the bytes live; absent on pre-blobKey records - name: string - mimeType: string - size: number - sourceUrl?: string // where the bytes were fetched FROM (provenance) - createdAt: number // epoch ms -} - -interface ArtifactStore { - save(record: ArtifactRecord): Promise - get(artifactId: string): Promise - list(runId: string): Promise> // [] when the run has none - delete(artifactId: string): Promise - deleteForRun(runId: string): Promise -} -``` - -### BlobStore - -A durable object/blob store for the bytes. `withGenerationPersistence` writes -each generated file under the key `artifacts//`. - -```ts -type BlobBody = - | ReadableStream - | ArrayBuffer - | ArrayBufferView - | string - | Blob - -interface BlobRecord { - key: string - size?: number - etag?: string - contentType?: string - customMetadata?: Record - createdAt?: number // epoch ms first written - updatedAt?: number // epoch ms last overwritten -} - -interface BlobObject extends BlobRecord { - arrayBuffer(): Promise - text(): Promise - body?: ReadableStream - // The slice served, when a range was requested. Absent on a whole read. - range?: { offset: number; length: number } -} - -interface BlobListPage { - objects: Array - cursor?: string // present only when `truncated` - truncated?: boolean -} - -interface BlobPutOptions { - contentType?: string - customMetadata?: Record - // Exact byte length of `body`, when the producer knows it. Advisory: use it - // to pick an upload strategy (single-shot vs multipart), never as a - // substitute for counting the bytes you actually store. - expectedLength?: number -} - -interface BlobRange { - offset: number // from the start of the object; must be inside it - length?: number // defaults to "to the end"; clamped when it overshoots -} - -interface BlobGetOptions { - range?: BlobRange -} - -interface BlobListOptions { - prefix?: string - cursor?: string - limit?: number -} - -interface BlobStore { - put(key: string, body: BlobBody, options?: BlobPutOptions): Promise - get(key: string, options?: BlobGetOptions): Promise - head(key: string): Promise - delete(key: string): Promise - list(options?: BlobListOptions): Promise -} -``` - -Three contracts to hold for `list`: - -- `prefix` matches literally and case-sensitively. Escape SQL `LIKE` - metacharacters. -- When `limit` is given and more keys match, return `truncated: true` with a - `cursor`. Passing that cursor back returns the strictly-following keys, so - paging visits every key exactly once. -- `limit: 0` yields an empty, untruncated page. - -And one for `put`: the body can be a `ReadableStream` with **no declared -length**. That is how a URL-fetched artifact arrives whenever the origin does -not declare one it can be held to — a chunked reply, or a compressed one whose -`content-length` describes the compressed bytes. (When the origin *does* declare -a usable length, the body reaches you exactly as `fetch` produced it, length -intact, and `expectedLength` carries the same number.) Your store must drain a -length-less stream, not require a length up front. Backends that need a declared -length for a single-shot upload (Cloudflare R2 on workerd is one) can re-attach -`expectedLength` when it is present and stream through a multipart upload when -it is not — the `ai-persistence/build-cloudflare-artifact-store` skill ships that -recipe. The conformance testkit exercises the length-less case, so a store that -only handles byte bodies fails the suite. - -And one for `get`: honour `options.range` by returning **only that slice**. -`size` keeps reporting the whole object, and the returned `range` reports what -you actually served — together they are the `206` response a media player's -seeking depends on. `resolveBlobRange(size, range)` does the clamping (a -`length` past the end is legal and clamps; an `offset` past the end throws, -because a serve route should have answered `416` from `record.size` first): - -```ts ignore -import { resolveBlobRange } from '@tanstack/ai-persistence' - -async get(key: string, options?: BlobGetOptions) { - const row = await selectBlob(key) - if (!row) return null - if (!options?.range) return blobObject(row, row.body) - const served = resolveBlobRange(row.size, options.range) - // Slice at the storage layer, not after loading the whole object. - const bytes = await selectBlobSlice(key, served.offset, served.length) - return blobObject(row, bytes, served) -} -``` - -This is not optional for a store that holds bytes — the conformance testkit -asserts it. Ignoring `range` and returning the whole file is what makes -`