diff --git a/.changeset/persistence-layer.md b/.changeset/persistence-layer.md deleted file mode 100644 index faa370ec9..000000000 --- a/.changeset/persistence-layer.md +++ /dev/null @@ -1,78 +0,0 @@ ---- -'@tanstack/ai': minor -'@tanstack/ai-client': minor -'@tanstack/ai-angular': minor -'@tanstack/ai-preact': minor -'@tanstack/ai-react': minor -'@tanstack/ai-solid': minor -'@tanstack/ai-svelte': minor -'@tanstack/ai-vue': minor -'@tanstack/ai-persistence': minor -'@tanstack/ai-persistence-sql': minor -'@tanstack/ai-persistence-sqlite': minor -'@tanstack/ai-persistence-postgres': minor -'@tanstack/ai-persistence-cloudflare': minor -'@tanstack/ai-persistence-drizzle': minor -'@tanstack/ai-persistence-prisma': minor -'@tanstack/ai-sandbox': minor ---- - -Persistence + resumable runs as composable middleware. - -`withChatPersistence(...)` makes `chat()` runs durable: it loads/saves thread -message history (server-authoritative), creates/updates run records, persists -every AG-UI `StreamChunk` to an append-only public event log, and persists -usage. `withGenerationPersistence(...)` tracks generation run status and -optionally persists media artifacts/blobs (image, audio, TTS, video, -transcription). Both are fully **optional** — a run with no persistence -middleware is unchanged. The primary API is `AIPersistence` / -`defineAIPersistence`; `ChatPersistence` / `defineChatPersistence` remain -deprecated compatibility aliases. - -**Resume.** Each persisted chunk carries an in-band, opaque `cursor` (a -monotonic per-run sequence). A client that disconnects mid-run reconnects with -`{ threadId, runId, cursor }`; `chat({ cursor })` replays the persisted public -event tail after that cursor. The public chat hooks expose resume state and a -manual `resume()` helper, with automatic resume behavior enabled unless -`autoResume` is opted out. - -**Interrupts.** Actionable waits are represented by -`RUN_FINISHED.outcome.type === 'interrupt'` and resumed with AG-UI -`RunAgentInput.resume[]`. Pending user-actionable interrupts block normal new -input on the same thread by default. Approval custom events are legacy -compatibility/projection only. - -**Event model.** Public stream events are separate from internal CAS/checkpoint -events. `PublicEventStore` replays the AG-UI `StreamChunk` stream itself; -`InternalEventStore` stores package-owned checkpoints that must not leak into -public replay. Optional feature validation fails loudly when a requested feature -is missing its required stores. - -**Backends (shared SQL core + thin adapters).** One SQL implementation behind a -minimal `SqlDriver` (`@tanstack/ai-persistence-sql`), with dialect support for -SQLite, Postgres, and MySQL and backends for SQLite (`-sqlite`, -node:sqlite/better-sqlite3), Postgres (`-postgres`, pg), Cloudflare D1 -(`-cloudflare`, with R2-backed `BlobStore`, D1-indexed R2 artifacts, and -concrete Durable Object locks), and bring-your-own Drizzle (`-drizzle`) and -Prisma (`-prisma`). The shared SQL core emits MySQL-safe DDL and conflict -handling (binary key columns, `LONGTEXT` persisted payloads, and no-op -`ON DUPLICATE KEY UPDATE` idempotent inserts). Raw drivers auto-migrate the -small base schema (`runs`, `public_events`, `internal_events`, `messages`, -`interrupts`, `metadata`, `_tanstack_ai_migrations`); ORMs own their schema. -The Drizzle and Prisma packages now include `tanstack-ai-persistence-drizzle` -and `tanstack-ai-persistence-prisma` CLIs that generate migration SQL files for -manual `migrate: false` workflows. -Cloudflare artifact metadata is lazily indexed in D1 while artifact bytes and -generic blobs live in R2, with optional artifact/blob cleanup APIs for deletion -and garbage collection. -`memoryPersistence()` ships in core for tests/examples. - -**Sandboxes, MCP, and workflows.** `@tanstack/ai-sandbox` now consumes -configured persistence capabilities directly. Place `withChatPersistence(...)` -before `withSandbox(...)` and sandbox resume records are stored in persistence -metadata while sandbox ensure-locking uses the shared persistence lock -capability when present. `defineSandbox({ persistence: { workspace } })` can -checkpoint managed sandbox workspace files to persistence metadata/artifacts -and restore them before `hooks.onReady`. MCP persistence is app-owned metadata -plus raw stream replay only. Workflow extensions are deferred to optional -packages and should reuse these primitives without adding base schema cost. diff --git a/.changeset/persistence-v2.md b/.changeset/persistence-v2.md new file mode 100644 index 000000000..66ccbe706 --- /dev/null +++ b/.changeset/persistence-v2.md @@ -0,0 +1,43 @@ +--- +'@tanstack/ai': minor +'@tanstack/ai-client': minor +'@tanstack/ai-durable-stream': minor +'@tanstack/ai-persistence': minor +'@tanstack/ai-persistence-drizzle': minor +'@tanstack/ai-persistence-prisma': minor +'@tanstack/ai-angular': minor +'@tanstack/ai-preact': minor +'@tanstack/ai-react': minor +'@tanstack/ai-solid': minor +'@tanstack/ai-svelte': minor +'@tanstack/ai-vue': minor +--- + +Persistence v2: split state (middleware) vs delivery (transport) durability. + +State durability stays on the middleware layer as `withChatPersistence` / +`withGenerationPersistence`, backed by a store contract (messages, runs, +interrupts, metadata, locks, artifacts, blobs). The `@tanstack/ai-persistence` +package now ships a state-only middleware, an in-core memory backend, and a +shared conformance suite; `@tanstack/ai-persistence-drizzle` and +`@tanstack/ai-persistence-prisma` provide state backends. + +Delivery durability (replay a disconnected/reloaded stream) moves to the +transport layer via a pluggable `StreamDurability` sink — the SDK owns zero +delivery-event storage. The new `@tanstack/ai-durable-stream` package provides a +durable-streams-protocol `StreamDurability` adapter, and the client +(`@tanstack/ai-client`) plus every framework binding (`@tanstack/ai-react`, +`-solid`, `-vue`, `-svelte`, `-angular`, `-preact`) switch to resumable SSE with +interrupt-only resume. + +The home-grown delivery/event-log subsystem is removed: the in-band `cursor` on +`StreamChunk`, the `cursor` param on `chat()`, the public/internal event stores, +the `ResumeSource` seam, cursor-replay client machinery, and the deprecated +approval-store shim are all deleted (this feature is unreleased — no back-compat +shims). Interrupt resume (approvals / client-tool results via +`RunAgentInput.resume[]`) is preserved as state durability. + +The delivery-store / SQL-driver packages that only existed to back the removed +event log — `@tanstack/ai-persistence-sql`, `-sqlite`, `-postgres`, and +`-cloudflare` — are removed. They were never published (part of this same +unreleased persistence work), so no npm tombstone release is needed. diff --git a/docs/chat/connection-adapters.md b/docs/chat/connection-adapters.md index c3ec11a1c..3bd48d452 100644 --- a/docs/chat/connection-adapters.md +++ b/docs/chat/connection-adapters.md @@ -333,7 +333,6 @@ function websocketConnection(url: string): SubscribeConnectionAdapter { JSON.stringify({ threadId: runContext?.threadId, runId: runContext?.runId, - cursor: runContext?.cursor, resume: runContext?.resume, messages, data, @@ -402,7 +401,6 @@ const myAdapter: ConnectConnectionAdapter = { body: JSON.stringify({ threadId: runContext?.threadId, runId: runContext?.runId, - cursor: runContext?.cursor, resume: runContext?.resume, messages, ...data, diff --git a/docs/config.json b/docs/config.json index 8cf049e1b..a3560bbe7 100644 --- a/docs/config.json +++ b/docs/config.json @@ -409,6 +409,11 @@ "addedAt": "2026-06-18", "updatedAt": "2026-07-09" }, + { + "label": "Delivery Durability", + "to": "persistence/delivery-durability", + "addedAt": "2026-07-09" + }, { "label": "Persistence Controls", "to": "persistence/controls", @@ -419,7 +424,7 @@ "label": "Migrations", "to": "persistence/migrations", "addedAt": "2026-07-08", - "updatedAt": "2026-07-08" + "updatedAt": "2026-07-09" }, { "label": "Chat Persistence", @@ -449,7 +454,7 @@ "label": "Cloudflare", "to": "persistence/cloudflare", "addedAt": "2026-07-03", - "updatedAt": "2026-07-08" + "updatedAt": "2026-07-09" }, { "label": "Prisma", diff --git a/docs/media/generation-hooks.md b/docs/media/generation-hooks.md index fc1317984..d6e5e954c 100644 --- a/docs/media/generation-hooks.md +++ b/docs/media/generation-hooks.md @@ -33,7 +33,7 @@ Generation hooks share a consistent API across all media types: | `useGenerateVideo` | `VideoGenerateInput` | `VideoGenerateResult` | | `useGeneration` | Generic `TInput` | Generic `TResult` | -Every hook returns the same core shape: `generate`, `result`, `isLoading`, `error`, `status`, `stop`, and `reset`. Streaming hooks also expose lightweight resume state and persisted artifact refs. You provide either a `connection` (streaming transport) or a `fetcher` (direct async call). +Every hook returns the same core shape: `generate`, `result`, `isLoading`, `error`, `status`, `stop`, and `reset`. Streaming hooks also expose a read-only generation state snapshot (`resumeSnapshot`, `resumeState`, `pendingArtifacts`, `resultArtifacts`) for observability and rendering persisted artifact refs. You provide either a `connection` (streaming transport) or a `fetcher` (direct async call). ## Server Setup @@ -399,10 +399,8 @@ function EmbeddingGenerator() { | `fetcher` | `GenerationFetcher` | Direct async function (no streaming protocol needed) | | `id` | `string` | Unique identifier for this generation instance | | `body` | `Record` | Additional body parameters sent with connection requests | -| `persistence` | `{ server?: GenerationServerPersistence }` | Stores the lightweight generation resume snapshot. Generated media bytes are not stored in browser persistence. | -| `autoResume` | `boolean` | Whether the hook should resume a persisted run on mount. Defaults to `true`. | -| `initialResumeSnapshot` | `GenerationResumeSnapshot` | Initial lightweight snapshot restored by the app or a persistence adapter. | -| `resumeState` | `GenerationResumeState` | Explicit `{ threadId, runId, cursor }` to use for the next resume or generation request. | +| `persistence` | `{ server?: GenerationServerPersistence }` | Stores the lightweight generation state snapshot. Generated media bytes are not stored in browser persistence. | +| `initialResumeSnapshot` | `GenerationResumeSnapshot` | Initial lightweight snapshot restored by the app or a persistence adapter, surfaced as read-only state. | | `onResult` | `(result: TResult) => TOutput \| null \| void` | Transform or react to results | | `onError` | `(error: Error) => void` | Error callback | | `onProgress` | `(progress: number, message?: string) => void` | Progress updates (0-100) | @@ -421,9 +419,8 @@ function EmbeddingGenerator() { | `status` | `GenerationClientState` | `'idle'` \| `'generating'` \| `'success'` \| `'error'` | | `stop` | `() => void` | Abort the current generation | | `reset` | `() => void` | Clear result, error, and return to idle | -| `resume` | `(state?: GenerationResumeState) => Promise` | Reconnect to the current, initial, or explicit resumable generation run | -| `resumeSnapshot` | `GenerationResumeSnapshot \| undefined` | Lightweight snapshot containing resume cursor, status, errors, and artifact refs | -| `resumeState` | `GenerationResumeState \| null` | Current `{ threadId, runId, cursor }`, or `null` when nothing is resumable | +| `resumeSnapshot` | `GenerationResumeSnapshot \| undefined` | Read-only lightweight snapshot containing run/cursor metadata, status, errors, and artifact refs | +| `resumeState` | `GenerationResumeState \| null` | Observed `{ threadId, runId }` metadata from the snapshot, or `null` (read-only) | | `pendingArtifacts` | `Array` | Persisted artifact refs observed during generation or replay before completion | | `resultArtifacts` | `Array` | Persisted artifact refs attached to the final replayed result | @@ -452,12 +449,19 @@ const { result } = useGenerateImage({ // result is now string[] instead of ImageGenerationResult ``` -## Resumable Generation +## Generation State Snapshot -All generation hooks can resume streamed generation endpoints when the server -uses `withGenerationPersistence(...)` and returns SSE events. Add `persistence.server` to -store the latest snapshot, rely on the default `autoResume: true`, or opt out -and call `resume()` from your UI. +Generation hooks do **not** auto-resume or expose a `resume()` action. A +generation run is only started by an explicit `generate(...)` call — hooks never +re-launch a run on mount. What the hooks _do_ surface, when the server uses +`withGenerationPersistence(...)` and returns SSE events, is a **read-only** +snapshot of the latest observed run for observability and for rendering +persisted artifact refs. + +Add `persistence.server` to store the latest snapshot under a stable `id`, or +pass `initialResumeSnapshot` to hydrate it from your own store. The hook then +reflects `resumeSnapshot`, `resumeState`, `pendingArtifacts`, and +`resultArtifacts` as read-only values. ```tsx import { localStorageAIPersistence } from '@tanstack/ai-client' @@ -469,22 +473,32 @@ export function TrailerVideoGenerator() { connection: fetchServerSentEvents('/api/generate/video'), persistence: { server: localStorageAIPersistence({ - keyPrefix: 'tanstack-ai:generation-resume:', + keyPrefix: 'tanstack-ai:generation-state:', }), }, }) return ( - +
+ + {video.resumeState &&

Last run: {video.resumeState.runId}

} + {video.pendingArtifacts.map((artifact) => ( + + {artifact.name} + + ))} +
) } ``` -If the server aborts provider work on disconnect, resume can only replay events -that were already persisted. If a durable producer keeps running on the server, -the hook can reconnect and receive the later artifact/result events. +The snapshot lets you render persisted artifact refs and observe run status +after a reload, but it does not reconnect to or replay a server run. Generated +bytes live in server-side artifact/blob stores. See +[Generation Persistence](../persistence/generation-persistence) for serving +those durable bytes. ## Framework Variants diff --git a/docs/persistence/chat-persistence.md b/docs/persistence/chat-persistence.md index eedccd2d1..97879e564 100644 --- a/docs/persistence/chat-persistence.md +++ b/docs/persistence/chat-persistence.md @@ -4,20 +4,23 @@ id: chat-persistence --- Use chat persistence when the server should be authoritative for a thread. The -client may keep local UI state, but the durable transcript, run status, -replayable event log, and pending user decisions live behind -`withChatPersistence(...)`. +client may keep local UI state, but the durable transcript, run status, and +pending user decisions live behind `withChatPersistence(...)`. This is **state +durability** — messages, runs, and interrupts. Delivery durability (replaying an +in-flight stream after a disconnect) is a separate transport concern; see +[Delivery Durability](./delivery-durability). -By the end, your endpoint accepts `{ threadId, runId, cursor, resume }`, writes -streamed chunks to durable storage, and lets the client resume after an -in-session disconnect or full page reload. +By the end, your endpoint accepts `{ threadId, runId, resume }`, persists chat +state at boundaries, and — paired with a delivery-durability sink — lets the +client reconnect to an in-progress response after a disconnect or reload. ## Install a backend -SQLite is the simplest durable backend for a Node server: +SQLite is the simplest durable backend for a Node server. The batteries-included +`sqlPersistence` ships Drizzle-generated migrations: ```sh -pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-sqlite +pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-drizzle ``` ## Create the server endpoint @@ -25,38 +28,47 @@ pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-sqlite Build the persistence instance once and reuse it across requests. ```ts -import { chat, toServerSentEventsResponse } from '@tanstack/ai' +import { + chat, + memoryStream, + toServerSentEventsResponse, +} from '@tanstack/ai' import { anthropicText } from '@tanstack/ai-anthropic' import { withChatPersistence } from '@tanstack/ai-persistence' -import { sqlitePersistence } from '@tanstack/ai-persistence-sqlite' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -const persistence = sqlitePersistence({ - path: '.tanstack-ai/state.sqlite', +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/state.sqlite', migrate: true, }) export async function POST(request: Request) { - const { messages, threadId, runId, cursor, resume } = await request.json() + const { messages, threadId, runId, resume } = await request.json() const stream = chat({ threadId, runId, - cursor, resume, adapter: anthropicText('claude-sonnet-4-6'), messages, middleware: [withChatPersistence(persistence)], }) - return toServerSentEventsResponse(stream) + // State persists at boundaries; the durability sink makes the delivered + // stream resumable (native Last-Event-ID reconnect / second-tab join). + return toServerSentEventsResponse(stream, { + durability: memoryStream(request), + }) } ``` `withChatPersistence(...)` loads stored thread history, saves the resulting -transcript, records run status, and appends every public AG-UI event with an -opaque cursor. When `cursor` is present, the run replays persisted events after -that cursor instead of re-running the adapter. For the exact event log and -cursor validation rules, see [Persistence Internals](./internals). +transcript, and records run status and interrupts at run boundaries. `resume` +carries interrupt/approval decisions back into a paused run. Delivery resume is +handled entirely by the transport's durability sink — see +[Delivery Durability](./delivery-durability). For the state store contract, see +[Persistence Internals](./internals). If the same app also uses `withGenerationPersistence`, keep **run IDs unique across activities** — they may share a store and `threadId`, but not a @@ -117,15 +129,17 @@ export function Chat() { } ``` -Auto-resume is enabled by default. On mount, reconnect, or when the tab comes -back online, the client can continue an interrupted run by forwarding the last -known `{ threadId, runId, cursor }`. Opt out with `autoResume: false`, or call -`chat.resume()` when you want a manual retry button. +Delivery resume is transparent: the resumable SSE connection reattaches to an +in-flight run via the browser's native `Last-Event-ID` on reconnect, with no +client cursor state. There is no `resume()`/`autoResume` on `useChat` — see +[Delivery Durability](./delivery-durability). -`chat.resumeState` contains the active resume identity, or `null` when there is -nothing to continue. `chat.pendingInterrupts` contains the client-side -descriptors needed to answer pending user decisions. `persistence.server` -stores them together and hydrates them on the next client construction. +`chat.resumeState` contains the active interrupt-resume identity +(`{ threadId, runId }`), or `null` when there is nothing to continue. +`chat.pendingInterrupts` contains the client-side descriptors needed to answer +pending user decisions, resolved with `chat.resumeInterrupts(...)`. +`persistence.server` stores them together and hydrates them on the next client +construction. ## Choose the controls you need @@ -136,8 +150,8 @@ chat, these are the common combinations: | --- | --- | | Browser-only drafts | `persistence.client` on the chat client. | | Server-owned transcript | `stores.messages` and `features: ['messages']`. | -| Reconnect without re-running the model | `stores.runs`, `stores.publicEvents`, and `durable-replay`. | -| Pending approvals or human input | `interrupts`, which also requires run and public-event stores. | +| Reconnect without re-running the model | A delivery-durability sink on the transport — see [Delivery Durability](./delivery-durability). | +| Pending approvals or human input | `interrupts`, which also requires the `stores.runs` and `stores.interrupts` stores. | | Multi-worker resume safety | Add `stores.locks` when a backend supports it. | ## Resume pending decisions diff --git a/docs/persistence/cloudflare.md b/docs/persistence/cloudflare.md index 506448957..c6d726fca 100644 --- a/docs/persistence/cloudflare.md +++ b/docs/persistence/cloudflare.md @@ -3,138 +3,57 @@ title: Cloudflare Persistence id: cloudflare --- -Use the Cloudflare backend when `chat()` runs in Workers and persistence should -stay on Cloudflare primitives. D1 stores the core SQL state, R2 can hold blobs -and artifact bytes, and Durable Objects can provide cross-isolate locks. +Run `chat()` in Workers and keep **state durability** on Cloudflare primitives. +D1 is SQLite-compatible, so the Drizzle D1 adapter satisfies the state store +contract directly — there is no dedicated first-party Cloudflare persistence +package. Point [`drizzlePersistence`](./drizzle) at a Drizzle D1 database and you +get messages, runs, interrupts, metadata, artifacts, and blobs backed by D1. -## Bind D1, R2, and Durable Objects +Delivery durability (replaying an in-flight stream after a disconnect) is a +separate transport concern — see [Delivery Durability](./delivery-durability). -```ts -import { cloudflarePersistence } from '@tanstack/ai-persistence-cloudflare' +## Bind D1 with drizzlePersistence + +```ts ignore +import { drizzlePersistence } from '@tanstack/ai-persistence-drizzle' +import { drizzle } from 'drizzle-orm/d1' interface Env { AI_D1: D1Database - AI_BLOBS: R2Bucket - AI_LOCKS: DurableObjectNamespace } export function persistence(env: Env) { - return cloudflarePersistence({ - d1: env.AI_D1, - r2: env.AI_BLOBS, - durableObjects: env.AI_LOCKS, - r2ArtifactPrefix: 'tanstack-ai/artifacts/', - r2BlobPrefix: 'tanstack-ai/blobs/', - migrate: true, - }) + // D1 is SQLite-compatible; the Drizzle D1 adapter is a BaseSQLiteDatabase and + // is assignable to drizzlePersistence's `db`. Generate/apply migrations from + // the exported `schema` with drizzle-kit (see the Drizzle guide). + return drizzlePersistence(drizzle(env.AI_D1)) } ``` -`d1` is required. `r2` and `durableObjects` are optional; include them only when -your app needs blob/artifact storage or distributed locks. - -You can also use R2 as the blob side of a hybrid persistence design while your -app keeps runs, messages, events, metadata, and artifact indexes in a separate -user-owned database. Implement that shape with [Custom Stores](./custom-stores): -the Cloudflare backend is convenient when D1 owns the SQL state, but the R2 -artifact/blob pattern is not limited to D1. +D1 backs every state store, including `stores.artifacts` (metadata rows) and +`stores.blobs` (byte payloads in a SQLite blob column). If you need generated +bytes in R2 instead of D1, keep run/message/artifact-index state in D1 and +implement an R2-backed blob store via [Custom Stores](./custom-stores). -For generated media hooks backed by R2 artifact refs, see +For generated media hooks backed by artifact refs, see [Generation Persistence](./generation-persistence). For sandbox workspace -checkpoints backed by D1/R2, see [Sandbox Persistence](./sandbox-persistence). - -For image, audio, speech, transcription, and video hooks that reconnect after a -refresh, see [Generation Persistence](./generation-persistence). - -## Core state lives in D1 - -D1 backs the shared SQL stores: - -- runs, -- public replay events, -- internal events, -- messages, -- interrupts, -- metadata, -- migration bookkeeping. - -That means reconnect and resume behavior does not depend on R2. R2 only stores -byte payloads for the optional blob and artifact stores. +checkpoints, see [Sandbox Persistence](./sandbox-persistence). -## Store blobs and artifacts in R2 +## Migrations -When you pass `r2`, the backend adds: +`drizzlePersistence` is bring-your-own: generate migrations from the exported +`schema` with drizzle-kit and apply them to D1 (`wrangler d1 migrations`). Unlike +the batteries-included `sqlPersistence` (Node SQLite only), the D1 path does not +bundle migrations — you own the D1 schema lifecycle. See the +[Drizzle guide](./drizzle) for the schema export and the drizzle-kit flow. -- `stores.blobs` backed by R2 objects, -- `stores.artifacts` with D1 metadata/index rows and R2-backed bytes. +## Locks -Artifact `list(runId)` reads the D1 index without downloading byte bodies. -Artifact `get(artifactId)` hydrates bytes from R2 when the artifact record -points at a blob. Optional artifact cleanup deletes both the D1 row and the R2 -object. - -Use artifacts for files tied to a chat or generation run, such as generated -reports, media outputs, downloads, screenshots, or sandbox workspace file -checkpoints. Use blobs for lower-level object storage when your app owns the -index shape. [Generation Persistence](./generation-persistence) shows generated -media artifacts, and [Sandbox Persistence](./sandbox-persistence) shows the -Cloudflare layout for project-builder apps. - -For Lovable-style builders, persistence stores generated artifact metadata and -sandbox workspace manifests in D1 while R2 stores the bytes. That gives -resumable media generation and workspace checkpointing, but it is not a turnkey -full project sync product: history, branching, merge policy, garbage -collection, and source-control semantics remain app responsibilities. - -## Use Durable Objects for locks - -Pass a Durable Object namespace when sandbox resume, workflow coordination, or -another feature needs cross-isolate mutual exclusion. - -```ts -import { - LockDurableObject, - cloudflarePersistence, -} from '@tanstack/ai-persistence-cloudflare' - -interface Env { - AI_D1: D1Database - AI_LOCKS: DurableObjectNamespace -} - -export { LockDurableObject } - -export function persistence(env: Env) { - return cloudflarePersistence({ - d1: env.AI_D1, - durableObjects: env.AI_LOCKS, - migrate: true, - durableObjectLocks: { - leaseMs: 30_000, - pollMs: 50, - }, - }) -} -``` - -Bind the exported `LockDurableObject` class in your Worker config and pass that -namespace as `durableObjects`. - -## Self-managed schema - -Cloudflare migrations are disabled by default. If you want the backend to create -the D1 tables lazily, pass `migrate: true`. If you deploy schema separately, -leave `migrate` unset or set it to `false`, and apply both the shared SQL DDL -and the Cloudflare artifact index DDL. - -```ts -import { cloudflareArtifactDdl } from '@tanstack/ai-persistence-cloudflare' -import { ddl } from '@tanstack/ai-persistence-sql' - -export const statements = [...ddl('sqlite'), ...cloudflareArtifactDdl()] -``` +The state backends ship an in-memory lock as a dev default. For cross-isolate +mutual exclusion in Workers (sandbox resume, workflow coordination), provide a +Durable Object–backed lock store that implements the `LockStore` contract and +pass it as `stores.locks` via [Custom Stores](./custom-stores). Bind the Durable +Object class in your Worker config as usual. -D1 is SQLite-compatible, so use the SQLite dialect for the shared core DDL. The -artifact DDL creates the Cloudflare-specific artifact index table used when R2 -is attached. For the D1/R2 table map, prefix defaults, and artifact deletion -caveats, see [Persistence Internals](./internals). +For the state store method contracts, see +[Persistence Internals](./internals). diff --git a/docs/persistence/controls.md b/docs/persistence/controls.md index a383bef94..a07009c95 100644 --- a/docs/persistence/controls.md +++ b/docs/persistence/controls.md @@ -25,9 +25,9 @@ chat({ }) ``` -There is no resume cursor, no server transcript, and no durable run record. This -is enough for one-shot server work, tests, or prototypes where reload recovery -does not matter. +There is no server transcript and no durable run record. This is enough for +one-shot server work, tests, or prototypes where reload recovery does not +matter. ## Lever 2: client snapshots @@ -54,10 +54,10 @@ const chat = useChat({ ``` `persistence.client` stores rendered UI messages. `persistence.server` stores -the lightweight server resume snapshot: `{ threadId, runId, cursor }`, pending -interrupt descriptors, status, and errors. The snapshot is not the source of -truth for a server-authoritative transcript; it only lets the client reconnect -to durable server state. +the lightweight server resume snapshot: `{ threadId, runId }`, pending interrupt +descriptors, status, and errors. The snapshot is not the source of truth for a +server-authoritative transcript; it only lets the client restore pending +interrupts after a full reload. ## Lever 3: server messages @@ -92,44 +92,23 @@ const middleware = withChatPersistence(persistence, { }) ``` -`stores.messages` is enough for server-owned history. It does not provide replay -after a disconnected stream; add lever 4 when reconnects must continue the same -run without re-calling the model. +`stores.messages` is enough for server-owned history. It does not make the +delivered stream resumable after a disconnect — that is **delivery durability**, +a transport concern handled by a durability sink on `toServerSentEvents(...)`, +not a persistence store. See [Delivery Durability](./delivery-durability). -## Lever 4: durable replay - -Use durable replay when reconnects should replay persisted public AG-UI events -after the last cursor. - -```ts -import { withChatPersistence } from '@tanstack/ai-persistence' -import { postgresPersistence } from '@tanstack/ai-persistence-postgres' - -const persistence = postgresPersistence({ - connectionString: process.env.DATABASE_URL ?? '', -}) - -export const middleware = withChatPersistence(persistence, { - features: ['messages', 'durable-replay'], -}) -``` - -Durable replay requires `stores.runs` and `stores.publicEvents`. `runs` tracks -run status, usage, and errors. `publicEvents` appends the user-visible stream -events and returns opaque cursors. Store the cursor and pass it back; never -derive meaning from its format. - -## Lever 5: interrupts and approvals +## Lever 4: interrupts and approvals Use interrupt persistence when a run can pause for a user decision and resume later. ```ts import { withChatPersistence } from '@tanstack/ai-persistence' -import { sqlitePersistence } from '@tanstack/ai-persistence-sqlite' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -const persistence = sqlitePersistence({ - path: '.tanstack-ai/state.sqlite', +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/state.sqlite', migrate: true, }) @@ -138,9 +117,9 @@ export const middleware = withChatPersistence(persistence, { }) ``` -The `interrupts` feature requires `stores.runs`, `stores.publicEvents`, and -`stores.interrupts`. A pending wait is surfaced through the public stream, then -the client resumes the same run with AG-UI `resume[]` entries. +The `interrupts` feature requires `stores.runs` and `stores.interrupts`. A +pending wait is surfaced on the stream, then the client resumes the same run +with AG-UI `resume[]` entries. ```ts import type { UseChatReturn } from '@tanstack/ai-react' @@ -159,14 +138,13 @@ await chat.resumeInterrupts([ Approval UIs are one common presentation of interrupts. New durable flows should treat the interrupt outcome and `resume[]` payload as the source of truth. -## Lever 6: extension stores +## Lever 5: extension stores Use extension stores when persistence must support integrations beyond the -basic chat replay path. +basic message/interrupt path. | Store | Use it for | | --- | --- | -| `stores.internalEvents` | Private checkpoints that must not replay to users. | | `stores.metadata` | App-owned session ids, manifests, correlation state, and pointers. | | `stores.locks` | Cross-worker coordination for the same thread, sandbox, or workflow. | | `stores.artifacts` | Named outputs tied to a run and thread. | @@ -178,29 +156,16 @@ media and file storage. If you pass a manual feature list, include both ```ts import { withChatPersistence } from '@tanstack/ai-persistence' -import { cloudflarePersistence } from '@tanstack/ai-persistence-cloudflare' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -declare const env: { - AI_D1: D1Database - AI_BLOBS: R2Bucket - AI_LOCKS: DurableObjectNamespace -} - -const persistence = cloudflarePersistence({ - d1: env.AI_D1, - r2: env.AI_BLOBS, - durableObjects: env.AI_LOCKS, +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/state.sqlite', + migrate: true, }) export const middleware = withChatPersistence(persistence, { - features: [ - 'messages', - 'durable-replay', - 'metadata', - 'locks', - 'artifacts', - 'blobs', - ], + features: ['messages', 'metadata', 'locks', 'artifacts', 'blobs'], }) ``` diff --git a/docs/persistence/custom-stores.md b/docs/persistence/custom-stores.md index 2ca56cb31..ba30a3a82 100644 --- a/docs/persistence/custom-stores.md +++ b/docs/persistence/custom-stores.md @@ -68,7 +68,7 @@ const persistence = defineAIPersistence({ }) const middleware = withChatPersistence(persistence, { - features: ['messages', 'durable-replay', 'interrupts', 'metadata'], + features: ['messages', 'interrupts', 'metadata'], }) ``` @@ -131,18 +131,17 @@ export function persistenceMiddleware(db: AppDb) { } ``` -Add `publicEvents` when reconnect replay matters, `interrupts` when runs pause -for user action, `internalEvents` for package or workflow checkpoints, -`metadata` for app-owned correlation, `locks` for cross-process coordination, -and `artifacts` plus `blobs` when runs produce durable files or media. Every -persistence feature is supported by implementing the corresponding stores. +Add `interrupts` when runs pause for user action, `metadata` for app-owned +correlation, `locks` for cross-process coordination, and `artifacts` plus +`blobs` when runs produce durable files or media. Every persistence feature is +supported by implementing the corresponding stores. (Delivery durability — +replaying an in-flight stream after a disconnect — is a transport concern, not a +store; see [Delivery Durability](./delivery-durability).) | Feature | Required stores | | --- | --- | | `messages` | `stores.messages` | -| `durable-replay` | `stores.runs`, `stores.publicEvents` | -| `interrupts` | `stores.runs`, `stores.publicEvents`, `stores.interrupts` | -| `internal-events` | `stores.internalEvents` | +| `interrupts` | `stores.runs`, `stores.interrupts` | | `metadata` | `stores.metadata` | | `locks` | `stores.locks` | | `artifacts` | `stores.artifacts` | diff --git a/docs/persistence/delivery-durability.md b/docs/persistence/delivery-durability.md new file mode 100644 index 000000000..0e41e5179 --- /dev/null +++ b/docs/persistence/delivery-durability.md @@ -0,0 +1,155 @@ +--- +title: Delivery Durability +id: delivery-durability +--- + +**Delivery durability** is the transport-layer concern of letting a client +disconnect, reload, or open a second tab and still receive the full, ordered +run stream **exactly once**. It is distinct from **state durability** (thread +messages, run status, interrupts, artifacts — the middleware layer covered by +[Chat Persistence](./chat-persistence.md) and +[Generation Persistence](./generation-persistence.md)). + +You get it by handing the transport helper a `durability` sink. The sink owns +**zero** application storage of its own — it appends the produced chunks to a +log and replays them on resume. Two backends ship: + +- `memoryStream(request)` — a process-local log. Zero infrastructure; the right + default for local dev and tests. +- `durableStream(request, { server })` — the + [durable-streams](https://durablestreams.com) protocol, for production. The + durable-streams server owns the bytes (its own WAL / group-commit); we store + nothing. + +## How resume works + +`chat()` is lazy — calling it runs no provider request until the first +`for await`. The transport helper uses that: + +- **Fresh request** — the helper iterates the stream, batches the chunks, + `append`s each batch to the durability log, and forwards them as SSE, tagging + every event with `id: `. +- **Reconnect / second tab** — the helper reads the resume offset off the + request (the browser's native `Last-Event-ID`, or a `?offset` query param), + replays the log strictly after that offset, and **never iterates the input + stream** — so `chat()` never calls the provider again. The untouched iterator + is simply garbage-collected. + +Because each SSE event carries an `id:`, native `EventSource` resends +`Last-Event-ID` on reconnect with zero application code. Joining an +already-running run passes `?offset=-1`. + +## Server: wire a durability sink + +Pass `durability` to any transport helper. In development, `memoryStream` needs +no setup: + +```ts +import { chat, memoryStream, toServerSentEventsResponse } from '@tanstack/ai' +import { openaiText } from '@tanstack/ai-openai' + +export async function POST(request: Request) { + const stream = chat({ + adapter: openaiText('gpt-5.5'), + messages: [{ role: 'user', content: 'Tell me about guitars.' }], + stream: true, + }) + + return toServerSentEventsResponse(stream, { + durability: memoryStream(request), + }) +} +``` + +For production, swap in `durableStream` — the only change is the sink: + +```ts +import { chat, toServerSentEventsResponse } from '@tanstack/ai' +import { durableStream } from '@tanstack/ai-durable-stream' +import { openaiText } from '@tanstack/ai-openai' + +export async function POST(request: Request) { + const stream = chat({ + adapter: openaiText('gpt-5.5'), + messages: [{ role: 'user', content: 'Tell me about guitars.' }], + stream: true, + }) + + return toServerSentEventsResponse(stream, { + durability: durableStream(request, { server: process.env.DS_URL ?? '' }), + // Optional: how many chunks to buffer per append (default 32). + batch: 32, + }) +} +``` + +The same `durability` option is available on `toHttpResponse` for +newline-delimited JSON transports (they resume via `?offset` rather than native +`Last-Event-ID`). + +## Client: plain, resumable SSE + +The client needs **no cursor machinery**. A plain `sse` connection is +resumable: when the server tags events with `id:` offsets, a dropped connection +reconnects with `Last-Event-ID` automatically and de-dupes any replayed prefix. + +```ts +import { useChat } from '@tanstack/ai-react' +import { fetchServerSentEvents } from '@tanstack/ai-client' + +function Chat() { + const chat = useChat({ + connection: fetchServerSentEvents('/api/chat'), + }) + // ...render chat.messages +} +``` + +### Joining an in-flight or finished run + +To attach a second tab (or re-attach after a full reload) to a run that is +already streaming, `joinRun(runId)` opens the stream from the start via +`?offset=-1`: + +```ts ignore +import { fetchServerSentEvents } from '@tanstack/ai-client' + +const connection = fetchServerSentEvents('/api/chat') + +for await (const chunk of connection.joinRun(runId)) { + // Same ordered event stream the original tab is receiving. + if (chunk.type === 'TEXT_MESSAGE_CONTENT' && 'delta' in chunk) { + appendDelta(chunk.delta) + } +} +``` + +## Ceiling: producer lifetime (`waitUntil`) + +Delivery durability replays what was **produced**. It cannot resume an LLM +completion that never finished producing: if the **producer process** dies +mid-run (e.g. a serverless function torn down the instant the client socket +closes), the log holds only the partial output and cannot continue it. + +The fix is deployment, not code — run the producer in something that outlives +the client socket: + +- a platform `waitUntil(...)` (Cloudflare Workers, Vercel) so the function keeps + running after the response is returned, +- a durable object, or +- a background queue / worker. + +With the producer kept alive, the log fills to completion and any reconnect or +second tab replays the whole run exactly once. + +## Escape hatch: CDN fan-out + +The shipped default keeps your app server in the connection path — it tails the +durability log and re-emits SSE. That is fine for hundreds to low-thousands of +concurrent readers. + +For kernel-speed CDN fan-out, point the browser **directly** at the +durable-streams server (the app server only writes; the CDN collapses many +readers onto one origin read). That direct-to-DS wiring is an advanced, +documented pattern, not shipped API — reach for it only when the app-server +fan-out ceiling is the actual bottleneck. diff --git a/docs/persistence/drizzle.md b/docs/persistence/drizzle.md index 61d642d50..17657215f 100644 --- a/docs/persistence/drizzle.md +++ b/docs/persistence/drizzle.md @@ -3,65 +3,103 @@ title: Persistence with Drizzle id: drizzle --- -Use Drizzle persistence when Drizzle already owns your SQLite or Postgres -database connection and migration workflow. The adapter unwraps Drizzle's -underlying client and exposes the shared SQL-backed `AIPersistence` stores to -`withChatPersistence(...)`. +`@tanstack/ai-persistence-drizzle` is the batteries-included SQL backend for +TanStack AI **state** persistence. It ships a Drizzle schema, drizzle-kit +migrations, and two entry points that both return the same `AIPersistence` +contract consumed by `withChatPersistence(...)` and `withGenerationPersistence(...)`. -By the end, your Drizzle-backed server can persist chat messages, replay -events, interrupts, metadata, and other SQL-backed stores while keeping schema -deployment in your existing Drizzle workflow. - -## Generate the migration - -Create a Drizzle migration file for your dialect. +- `sqlPersistence({ dialect, url, migrate })` — **batteries-included.** Give it a + dialect and a URL; it builds the database and applies the migrations bundled + in the package. +- `drizzlePersistence(db)` — **bring your own.** Pass a Drizzle database you + already constructed and migrated against the exported `schema`. ```sh -pnpm exec tanstack-ai-persistence-drizzle --dialect postgres +pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-drizzle drizzle-orm ``` -The default output path is `drizzle/_tanstack_ai_persistence.sql`. -You can also pass `--out`, `--stdout`, `--timestamp`, `--name`, and `--force`. - -Run the generated SQL through your Drizzle migration workflow before deploying. -Lazy migrations are opt-in; use `migrate: true` only for local or development -databases. +## Batteries-included: `sqlPersistence` -## Create the persistence object - -Pass your Drizzle database and dialect to `drizzlePersistence(...)`. +For local development and single-node deployments, `sqlPersistence` is the +fastest path. The `sqlite` dialect is bundled with pre-generated migrations, so +`migrate: true` creates the schema on first use. ```ts -import { drizzlePersistence } from '@tanstack/ai-persistence-drizzle' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' import { withChatPersistence } from '@tanstack/ai-persistence' -import { db } from './db' -export const persistence = drizzlePersistence({ - db, - dialect: 'postgres', +export const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:./.tanstack-ai/state.sqlite', + migrate: true, }) -export const middleware = withChatPersistence(persistence, { - features: ['messages', 'durable-replay', 'interrupts', 'metadata'], -}) +export const middleware = withChatPersistence(persistence) ``` -The adapter supports SQLite and Postgres workflows. It uses the shared SQL -stores, so the feature behavior matches the raw SQL and Prisma backends. +Use `url: ':memory:'` for tests. For production, generate and review the +migrations ahead of time (see below) and leave `migrate` unset so the schema is +deployed by your own pipeline rather than lazily at runtime. + +## Bring your own: `drizzlePersistence` + +When Drizzle already owns your database connection and migration workflow, pass +your `db` directly. This works with any Drizzle sqlite driver +(`better-sqlite3`, `libsql`/Turso, D1, `node:sqlite`). + +```ts ignore +import { drizzle } from 'drizzle-orm/better-sqlite3' +import Database from 'better-sqlite3' +import { drizzlePersistence, schema } from '@tanstack/ai-persistence-drizzle' +import { withChatPersistence } from '@tanstack/ai-persistence' + +const db = drizzle(new Database('state.sqlite'), { schema }) + +export const persistence = drizzlePersistence(db) +export const middleware = withChatPersistence(persistence) +``` + +The package re-exports the `schema` (and each table) so you can compose it into +your own Drizzle schema and drive migrations from your existing setup. + +## Generate migrations with drizzle-kit + +The schema is the single source of truth for migrations. The package ships a +`drizzle.config.ts`; regenerate the SQL under `drizzle/` after any schema change: + +```sh +pnpm --filter @tanstack/ai-persistence-drizzle db:generate +``` + +For a bring-your-own database, point drizzle-kit at the exported schema in your +own `drizzle.config.ts` and run `drizzle-kit generate` / `drizzle-kit migrate` +through your normal workflow. See [Migrations](./migrations) for the full flow. + +## What is persisted + +The schema mirrors the `AIPersistence` state records column-for-column: + +- `messages` — thread message history +- `runs` — run lifecycle (status, usage, timing) +- `interrupts` — interrupts / approvals +- `metadata` — scoped key/value metadata +- `artifacts` — generation artifact references +- `blobs` — generic blob objects + +Delivery durability (resuming an interrupted stream) is a **transport** concern +and is not stored here. Locks are not part of the SQL schema; an in-memory lock +is provided as a dev default. Swap in a distributed lock for multi-process +deployments via [Custom Stores](./custom-stores). -## Use it for different persistence goals +## Use it across the guides -Use the same Drizzle-backed `AIPersistence` object across the topic guides: +The same Drizzle-backed `AIPersistence` object works across the topic guides: -- [Chat Persistence](./chat-persistence) for server-owned transcripts and - replay cursors. +- [Chat Persistence](./chat-persistence) for server-owned transcripts. - [Persistence Controls](./controls) when you need to choose a feature list. -- [MCP Persistence](./mcp-persistence) when MCP session ids or tool-call - correlation should live in metadata and internal events. - [Custom Stores](./custom-stores) when you want Drizzle for SQL state but a separate object store for blobs. -If generated media or file artifacts must be durable, Drizzle can own the SQL -state, but you still need `stores.artifacts` and `stores.blobs`. Implement that -hybrid shape with [Custom Stores](./custom-stores), or use [Cloudflare](./cloudflare) -when D1 plus R2 fits your deployment. +Other dialects (`postgres`, `mysql`) are bring-your-own today: construct a +Drizzle db with your own driver and use `drizzlePersistence(db)` with migrations +generated from the exported `schema`. diff --git a/docs/persistence/generation-persistence.md b/docs/persistence/generation-persistence.md index b4cd8521e..1de8a751d 100644 --- a/docs/persistence/generation-persistence.md +++ b/docs/persistence/generation-persistence.md @@ -3,23 +3,25 @@ title: Generation Persistence id: generation-persistence --- -Use generation persistence when an image, audio, speech, transcription, or video -endpoint should survive a refresh, tab reconnect, or transient network drop. -The server owns the durable run through `withGenerationPersistence(...)`; the client -stores only the lightweight resume snapshot it needs to reconnect. +Use generation persistence when the generated media from an image, audio, +speech, transcription, or video endpoint should survive a refresh. The server +owns the durable run through `withGenerationPersistence(...)`, which records run +status plus artifact/blob records; the client keeps only a lightweight, +read-only state snapshot for observability. -By the end, your generation hook can resume a streaming endpoint and display -generated media from persisted artifact refs instead of browser-stored bytes. +By the end, your generation hook can display generated media from persisted +artifact refs instead of browser-stored bytes, and observe the last run's status +after a reload. ## Install a backend with artifacts and blobs Generated media/file persistence requires both `stores.artifacts` and -`stores.blobs`. Cloudflare D1 plus R2 is the durable media path shown below: -D1 stores replay state and artifact metadata, while R2 stores generated media -bytes. +`stores.blobs`. The batteries-included SQLite backend provides both: artifact +metadata rows and blob bytes. For a cloud object store (R2/S3) behind the blob +side, bring your own blob store via [Custom Stores](./custom-stores). ```sh -pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-cloudflare +pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-drizzle ``` ## Create the generation endpoint @@ -36,43 +38,36 @@ import { toServerSentEventsResponse, } from '@tanstack/ai' import { openaiImage } from '@tanstack/ai-openai' -import { cloudflarePersistence } from '@tanstack/ai-persistence-cloudflare' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' import { withGenerationPersistence } from '@tanstack/ai-persistence' -interface Env { - AI_D1: D1Database - AI_BLOBS: R2Bucket - AI_LOCKS: DurableObjectNamespace -} - -function persistence(env: Env) { - return cloudflarePersistence({ - d1: env.AI_D1, - r2: env.AI_BLOBS, - durableObjects: env.AI_LOCKS, - migrate: true, - }) -} +// State durability: run status + artifact metadata + blob bytes. The +// batteries-included SQLite backend stores artifact bytes in a blob column; +// for a cloud object store (R2/S3), bring your own blob store via custom stores. +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/generation.sqlite', + migrate: true, +}) -export async function POST(request: Request, env: Env) { - const { input, threadId, runId, cursor } = +export async function POST(request: Request) { + const { input, threadId, runId } = await generationParamsFromRequest('image', request) if (typeof input.prompt !== 'string') { throw new Error('This endpoint accepts text image prompts only.') } - const identity: { threadId?: string; runId?: string; cursor?: string } = {} + const identity: { threadId?: string; runId?: string } = {} if (threadId !== undefined) identity.threadId = threadId if (runId !== undefined) identity.runId = runId - if (cursor !== undefined) identity.cursor = cursor const stream = generateImage({ ...identity, prompt: input.prompt, adapter: openaiImage('gpt-image-2'), stream: true as const, - middleware: [withGenerationPersistence(persistence(env))], + middleware: [withGenerationPersistence(persistence)], }) return toServerSentEventsResponse(stream) @@ -97,10 +92,10 @@ and [Persistence Internals](./internals#shared-backends-unique-runid-across-chat ## Wire the client hook -Use `persistence.server` to store the latest generation resume snapshot under a -stable `id`. The snapshot contains `{ threadId, runId, cursor }`, status, -errors, and lightweight artifact refs. It does not contain generated image, -audio, or video bytes. +Use `persistence.server` to store the latest generation state snapshot under a +stable `id`. The snapshot contains `{ threadId, runId }`, status, errors, and +lightweight artifact refs. It does not contain generated image, audio, or video +bytes. ```tsx group=generation-persistence-client import { fetchServerSentEvents, useGenerateImage } from '@tanstack/ai-react' @@ -128,11 +123,7 @@ export function HeroImageGenerator() { Generate - {image.resumeState && ( - - )} + {image.resumeState &&

Last run: {image.resumeState.runId}

} {image.pendingArtifacts.map((artifact) => ( @@ -144,10 +135,13 @@ export function HeroImageGenerator() { } ``` -Auto-resume is enabled by default. Set `autoResume: false` when the UI should -wait for an explicit user action, then call `resume()`. `resume(state)` can also -accept an explicit `{ threadId, runId, cursor }` when your app stores the -identity outside the hook persistence adapter. +The snapshot is **read-only**. Generation hooks never auto-start a run on mount +and expose no `resume()` action — a run begins only when you call +`generate(...)`. `resumeState`, `pendingArtifacts`, and `resultArtifacts` let you +render persisted artifact refs and observe the last run's status after a reload; +they do not reconnect to a server run. Pass `initialResumeSnapshot` to hydrate +this state from your own store when the identity lives outside the hook +persistence adapter. ## Serve persisted artifacts @@ -155,31 +149,21 @@ Generation hooks store lightweight artifact refs. Serve the durable bytes from your app by looking up the artifact by `artifactId`. ```ts -import { cloudflarePersistence } from '@tanstack/ai-persistence-cloudflare' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -interface Env { - AI_D1: D1Database - AI_BLOBS: R2Bucket - AI_LOCKS: DurableObjectNamespace -} - -function persistence(env: Env) { - return cloudflarePersistence({ - d1: env.AI_D1, - r2: env.AI_BLOBS, - durableObjects: env.AI_LOCKS, - migrate: true, - }) -} +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/generation.sqlite', + migrate: true, +}) export async function GET( request: Request, context: { params: Promise<{ artifactId: string }> }, - env: Env, ) { void request const { artifactId } = await context.params - const { stores } = persistence(env) + const { stores } = persistence if (!stores.artifacts || !stores.blobs) { throw new Error('Artifact and blob stores are required.') @@ -223,19 +207,13 @@ include every input and output artifact you want persisted. ```ts import { generateImage } from '@tanstack/ai' import { openaiImage } from '@tanstack/ai-openai' -import { cloudflarePersistence } from '@tanstack/ai-persistence-cloudflare' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' import { withGenerationPersistence } from '@tanstack/ai-persistence' -declare const env: { - AI_D1: D1Database - AI_BLOBS: R2Bucket - AI_LOCKS: DurableObjectNamespace -} - -const persistence = cloudflarePersistence({ - d1: env.AI_D1, - r2: env.AI_BLOBS, - durableObjects: env.AI_LOCKS, +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/generation.sqlite', + migrate: true, }) const result = await generateImage({ @@ -271,11 +249,12 @@ snapshot stores `resumeState`, status, pending artifact refs, completed result artifact refs, and lightweight error metadata. Generated bytes are persisted by the server through artifact/blob stores, and -`withGenerationPersistence(...)` records generation run status plus artifact/blob records. -It does not append generation stream chunks to `publicEvents` by itself. If -your endpoint aborts the provider request when the browser disconnects, resume -cannot produce a future result for work the server canceled. Resumable -generation needs a durable producer, explicit replay events, or a durable result -that your endpoint can read when the client reconnects. For the artifact write -path and current generation replay caveats, see +`withGenerationPersistence(...)` records generation run status plus artifact/blob +records. State durability does not itself make the delivered stream resumable — +that is a transport concern; pair the endpoint with a delivery-durability sink +(see [Delivery Durability](./delivery-durability)). If your endpoint aborts the +provider request when the browser disconnects, resume cannot produce a future +result for work the server canceled. Resumable generation needs a durable +producer that outlives the client socket, or a durable result that your endpoint +can read when the client reconnects. For the artifact write path, see [Persistence Internals](./internals). diff --git a/docs/persistence/mcp-persistence.md b/docs/persistence/mcp-persistence.md index db4730455..ef6bc83b4 100644 --- a/docs/persistence/mcp-persistence.md +++ b/docs/persistence/mcp-persistence.md @@ -17,14 +17,20 @@ Managed MCP tools run inside `chat()`, so start with the same run durability you would use for a normal server-owned chat thread. ```ts -import { chat, toServerSentEventsResponse } from '@tanstack/ai' +import { + chat, + memoryStream, + toServerSentEventsResponse, +} from '@tanstack/ai' import { anthropicText } from '@tanstack/ai-anthropic' import { createMCPClient } from '@tanstack/ai-mcp' import { withChatPersistence } from '@tanstack/ai-persistence' -import { postgresPersistence } from '@tanstack/ai-persistence-postgres' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -const persistence = postgresPersistence({ - connectionString: process.env.DATABASE_URL ?? '', +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/state.sqlite', + migrate: true, }) const weather = await createMCPClient({ @@ -32,28 +38,30 @@ const weather = await createMCPClient({ }) export async function POST(request: Request) { - const { messages, threadId, runId, cursor } = await request.json() + const { messages, threadId, runId } = await request.json() const stream = chat({ threadId, runId, - cursor, adapter: anthropicText('claude-sonnet-4-6'), messages, mcp: { clients: [weather] }, middleware: [ withChatPersistence(persistence, { - features: ['messages', 'durable-replay', 'metadata', 'internal-events'], + features: ['messages', 'metadata'], }), ], }) - return toServerSentEventsResponse(stream) + return toServerSentEventsResponse(stream, { + durability: memoryStream(request), + }) } ``` -Public replay events keep the UI reconnectable. MCP-specific state belongs in -metadata and internal events when your app needs it. +MCP-specific state belongs in metadata when your app needs it. Making the +delivered stream reconnectable is a transport concern — see +[Delivery Durability](./delivery-durability). ## Store MCP session metadata @@ -86,8 +94,9 @@ store methods; it does not prescribe an MCP session table. ## Store private MCP checkpoints -Use `stores.internalEvents` for private tool-call or session checkpoints that -should not replay to the UI as public AG-UI events. +Use `stores.metadata` for private tool-call or session checkpoints that should +not surface to the UI. Metadata is app-owned key/value state scoped to a thread +or run. ```ts import type { AIPersistence } from '@tanstack/ai-persistence' @@ -98,29 +107,20 @@ async function recordMcpCheckpoint(input: { toolCallId: string serverId: string }) { - const internalEvents = input.persistence.stores.internalEvents + const metadata = input.persistence.stores.metadata - if (!internalEvents) { - throw new Error('MCP checkpoints require stores.internalEvents.') + if (!metadata) { + throw new Error('MCP checkpoints require stores.metadata.') } - const latestSeq = await internalEvents.latestSeq(input.runId, 'mcp') - - await internalEvents.append({ - runId: input.runId, - expectedSeq: latestSeq, - namespace: 'mcp', + await metadata.set(input.runId, `mcp:${input.toolCallId}`, { type: 'tool-call-observed', - payload: { - toolCallId: input.toolCallId, - serverId: input.serverId, - }, + serverId: input.serverId, }) } ``` -Use public events only for user-visible stream replay. Use internal events for -integration checkpoints, and use metadata for current lookup state. +Use metadata for integration checkpoints and current lookup state. ## Coordinate MCP work across workers diff --git a/docs/persistence/migrations.md b/docs/persistence/migrations.md index 3053fd780..fa55470b9 100644 --- a/docs/persistence/migrations.md +++ b/docs/persistence/migrations.md @@ -3,116 +3,67 @@ title: Persistence Migrations id: migrations --- -Run persistence migrations when you deploy a packaged SQL, ORM, or Cloudflare -backend. The default is production-safe: packaged backends do not create tables -unless you opt in with `migrate: true` or run generated DDL through your own -migration system. +The Drizzle backend owns its schema and migrations through drizzle-kit — there +is no hand-authored DDL. The default is production-safe: the packaged backend +does not create tables unless you opt in with `migrate: true` or apply the +generated migrations through your own workflow. -By the end, you can generate the right migration file for each adapter and know -when lazy migration is appropriate. +By the end, you can regenerate migrations after a schema change and know when +lazy migration is appropriate. ## Use lazy migration only for local development -Pass `migrate: true` when a local or development database should create the -TanStack AI tables on first use. +Pass `migrate: true` when a local or development database should apply the +bundled migrations on first use. ```ts -import { sqlitePersistence } from '@tanstack/ai-persistence-sqlite' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -export const persistence = sqlitePersistence({ - path: '.tanstack-ai/state.sqlite', +export const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:./.tanstack-ai/state.sqlite', migrate: true, }) ``` -For production, leave `migrate` unset or set it to `false`, generate a migration -file, review it, and apply it before traffic uses the persistence stores. +For production, leave `migrate` unset, apply the migrations with your own +pipeline, and deploy them before traffic uses the persistence stores. -## Generic SQL migrations +## Regenerate the bundled migrations -Use the generic SQL CLI for SQLite, Postgres, or MySQL DDL. +The schema in `src/schema.ts` is the single source of truth. After changing it, +regenerate the SQL under `drizzle/` with drizzle-kit: ```sh -pnpm exec tanstack-ai-persistence-sql --dialect sqlite --out migrations/001_tanstack_ai.sql -pnpm exec tanstack-ai-persistence-sql --dialect postgres --stdout -pnpm exec tanstack-ai-persistence-sql --dialect mysql --out migrations/001_tanstack_ai.sql +pnpm --filter @tanstack/ai-persistence-drizzle db:generate ``` -Options: +This runs `drizzle-kit generate` against the package's `drizzle.config.ts`, +diffing the schema and emitting a new migration file. Commit the generated +`drizzle/` output alongside the schema change. -| Option | Purpose | -| --- | --- | -| `--dialect sqlite\|postgres\|mysql` | Choose the SQL dialect. | -| `--out ` | Write SQL to a migration file. | -| `--stdout` | Print SQL instead of writing a file. | -| `--timestamp ` | Use a deterministic timestamp in generated names. | -| `--name ` | Use a deterministic migration name. | -| `--force` | Overwrite an existing output file. | +## Bring-your-own migrations -The generic CLI writes the shared persistence schema. It does not create -Cloudflare R2 artifact index tables, app-owned tables, or ORM-specific folder -layouts. - -## SQLite and Postgres backend migrations - -The raw SQLite and Postgres packages use the shared SQL schema. +When you construct your own Drizzle database and use `drizzlePersistence(db)`, +drive migrations from the exported `schema` with your own drizzle-kit config: ```ts -import { ddl } from '@tanstack/ai-persistence-sql' - -export const sqliteStatements = ddl('sqlite') -export const postgresStatements = ddl('postgres') -``` - -Use the generic CLI when you want migration files, or call `ddl(...)` from your -own migration generator. - -## Cloudflare migrations - -Cloudflare D1 uses the SQLite dialect for the shared SQL tables. If you also -attach R2 for artifacts and blobs, add the Cloudflare artifact index DDL. - -```ts -import { cloudflareArtifactDdl } from '@tanstack/ai-persistence-cloudflare' -import { ddl } from '@tanstack/ai-persistence-sql' - -export const statements = [...ddl('sqlite'), ...cloudflareArtifactDdl()] -``` +// drizzle.config.ts +import { defineConfig } from 'drizzle-kit' -Apply those statements with Wrangler, your D1 migration workflow, or your normal -deployment pipeline. Use `migrate: true` only when you intentionally want the -Worker backend to apply schema lazily. - -For the shared SQL table map and the difference between plain DDL and lazy -migration bookkeeping, see [Persistence Internals](./internals). - -## Prisma migrations - -Use the Prisma CLI when you want the default Prisma migration folder layout. - -```sh -pnpm exec tanstack-ai-persistence-prisma --dialect sqlite -pnpm exec tanstack-ai-persistence-prisma --dialect postgres --name tanstack_ai_persistence +export default defineConfig({ + dialect: 'sqlite', + schema: './node_modules/@tanstack/ai-persistence-drizzle/dist/esm/schema.js', + out: './drizzle', +}) ``` -The default output path is -`prisma/migrations/_tanstack_ai_persistence/migration.sql`. - -Prisma supports SQLite and Postgres workflows. Run the generated SQL through -your Prisma migration process before using `prismaPersistence(...)` in -production. See [Prisma](./prisma) for adapter wiring. - -## Drizzle migrations - -Use the Drizzle CLI when you want a Drizzle-style SQL migration file. - ```sh -pnpm exec tanstack-ai-persistence-drizzle --dialect sqlite -pnpm exec tanstack-ai-persistence-drizzle --dialect postgres --out drizzle/001_tanstack_ai.sql +pnpm drizzle-kit generate +pnpm drizzle-kit migrate ``` -The default output path is `drizzle/_tanstack_ai_persistence.sql`. - -Drizzle supports SQLite and Postgres workflows. Run the generated SQL through -your Drizzle migration workflow before using `drizzlePersistence(...)` in -production. See [Drizzle](./drizzle) for adapter wiring. +Compose the exported tables into your own schema file when you want TanStack AI +state to live in the same migration history as the rest of your app. See +[Drizzle](./drizzle) for adapter wiring and [Prisma](./prisma) for the peer +Prisma backend. diff --git a/docs/persistence/overview.md b/docs/persistence/overview.md index f7e68882f..cbd9a80af 100644 --- a/docs/persistence/overview.md +++ b/docs/persistence/overview.md @@ -3,41 +3,53 @@ title: Persistence Overview id: overview --- -Persistence makes server-side runs durable. For chat, add -`withChatPersistence(...)` to a `chat()` call; for media generation, add -`withGenerationPersistence(...)` to `generateImage` / `generateAudio` / TTS / -video / transcription. Pass either middleware an `AIPersistence` store -aggregate, and TanStack AI uses those stores to load thread history, record run -state, append replayable public events, and provide optional capabilities such -as interrupts, metadata, locks, artifacts, and blobs. - -By the end of this section, you can choose the smallest persistence layer your -app needs, wire it to your server route, and move from local development -migrations to a production-owned schema. - -## How persistence works +Persistence makes server-side runs durable. TanStack AI splits durability into +two independent concerns: + +- **State durability (this section)** — thread messages, run records, + interrupts/approvals, metadata, locks, and generation artifacts. This is a + store concern and lives on the **middleware** layer: add + `withChatPersistence(...)` to a `chat()` call, or `withGenerationPersistence(...)` + to `generateImage` / `generateAudio` / TTS / video / transcription. +- **Delivery durability (transport)** — a client disconnects, reloads, or opens + a second tab and still receives the full ordered stream. This is a *transport* + concern, handled by a pluggable delivery sink on the transport helpers + (`toServerSentEvents*`), **not** by the persistence middleware. That half is + documented separately once the transport delivery layer lands. + +The persistence middleware persists **state only**. It never rewrites the chunk +stream and stamps no cursor — a persisted run produces byte-identical chunks to +a non-persisted one. + +By the end of this section, you can choose the smallest state store your app +needs, wire it to your server route, and move from local development migrations +to a production-owned schema. + +## How state persistence works Both middlewares are opt-in. A run without them stays in-memory. A run with -either middleware calls the stores exposed by your `AIPersistence` object. +either middleware calls the stores exposed by your `AIPersistence` object at +boundaries (load thread on config, save thread + run status on finish, record +interrupts on pause) — low volume, not per-token. ```ts import { chat, toServerSentEventsResponse } from '@tanstack/ai' import { anthropicText } from '@tanstack/ai-anthropic' import { withChatPersistence } from '@tanstack/ai-persistence' -import { sqlitePersistence } from '@tanstack/ai-persistence-sqlite' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -const persistence = sqlitePersistence({ - path: '.tanstack-ai/state.sqlite', +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:./chat.db', migrate: true, }) export async function POST(request: Request) { - const { messages, threadId, runId, cursor, resume } = await request.json() + const { messages, threadId, runId, resume } = await request.json() const stream = chat({ threadId, runId, - cursor, resume, adapter: anthropicText('claude-sonnet-4-6'), messages, @@ -48,23 +60,23 @@ export async function POST(request: Request) { } ``` -When the client later sends `{ threadId, runId, cursor }`, persistence replays -the stored public event tail after that cursor. Treat `cursor` as opaque: store -and forward it, but do not parse it. +The `resume` array carries AG-UI interrupt responses (approvals, client-tool +results) for a thread that has pending interrupts — that is state, resolved by +`withChatPersistence`. Making the *stream itself* replayable across a disconnect +is a separate transport-layer delivery concern. ### Run IDs must be unique across activities `withChatPersistence` and `withGenerationPersistence` may share one `AIPersistence` backend. The `runs` store is keyed only by `runId` (there is no activity discriminator), so **every chat run and every generation run needs a -distinct `runId`**. Reusing the same id across activities overwrites run status -and confuses chat cursor / event-log validation. +distinct `runId`**. Reusing the same id across activities overwrites run status. How ids are created: | Path | Default `runId` | | --- | --- | -| Client chat (`useChat` / `ChatClient`) | New `run-${Date.now()}-…` per send; resume reuses the stored id | +| Client chat (`useChat` / `ChatClient`) | New `run-${Date.now()}-…` per send; interrupt resume reuses the stored id | | Client generation hooks | New `run-${Date.now()}-…` per generate via the generation client | | Server `chat({ runId })` | Caller-supplied, or falls back to a per-request `chat-…` id from the engine | | Server generation (`generateImage`, …) | Caller-supplied `runId`, else `requestId` for run tracking | @@ -76,27 +88,24 @@ generation for the same conversation is fine. ## What `AIPersistence` contains -`AIPersistence` is an object with optional stores. Implement only the stores -your scenario needs, or use a packaged backend that implements them for you. +`AIPersistence` is an object with optional state stores. Implement only the +stores your scenario needs, or use a packaged backend that implements them for +you. ```ts import { defineAIPersistence } from '@tanstack/ai-persistence' import type { ArtifactStore, BlobStore, - InternalEventStore, InterruptStore, MessageStore, MetadataStore, - PublicEventStore, RunStore, } from '@tanstack/ai-persistence' import type { LockStore } from '@tanstack/ai' declare const messages: MessageStore declare const runs: RunStore -declare const publicEvents: PublicEventStore -declare const internalEvents: InternalEventStore declare const interrupts: InterruptStore declare const metadata: MetadataStore declare const locks: LockStore @@ -107,8 +116,6 @@ export const persistence = defineAIPersistence({ stores: { messages, runs, - publicEvents, - internalEvents, interrupts, metadata, locks, @@ -121,38 +128,31 @@ export const persistence = defineAIPersistence({ There is no separate high-level callback builder. Production apps usually keep their own database, object storage, queues, and retention policies, then expose those systems through `AIPersistence` store methods such as -`messages.loadThread`, `runs.createOrResume`, `publicEvents.append`, or -`metadata.set`. +`messages.loadThread`, `runs.createOrResume`, or `metadata.set`. ## Choose your path Start with [Persistence Controls](./controls) when you need to decide which -stores and client options to enable. It maps the six control levers from -in-memory runs through extension stores. +stores and client options to enable. -Read [Migrations](./migrations) before deploying a packaged backend. Migrations -are opt-in by default; pass `migrate: true` only when you intentionally want -local or development lazy migration. +Read [Migrations](./migrations) before deploying a packaged backend. Each ORM +owns its own migrations (drizzle-kit / prisma migrate). Use the topic pages when you know the feature you are building: -- [Chat Persistence](./chat-persistence) for server-authoritative chat threads, - reconnect replay, and pending human decisions. +- [Chat Persistence](./chat-persistence) for server-authoritative chat threads + and pending human decisions. - [Generation Persistence](./generation-persistence) for image, audio, speech, transcription, and video hooks with durable artifact refs. - [Sandbox Persistence](./sandbox-persistence) for coding-agent sandboxes, workspace checkpoints, records, and locks. -- [MCP Persistence](./mcp-persistence) for durable MCP session metadata, - tool-call correlation, and event/checkpoint storage. +- [MCP Persistence](./mcp-persistence) for durable MCP session metadata and + tool-call correlation. Use the adapter pages when you know the infrastructure you are deploying on: -- [Cloudflare](./cloudflare) for Workers, D1, R2, and Durable Object locks. -- [SQL Backends](./sql-backends) for SQLite, Postgres, and MySQL-compatible - custom SQL drivers. -- [Prisma](./prisma) when Prisma owns your SQLite or Postgres database access. -- [Drizzle](./drizzle) when Drizzle owns your SQLite or Postgres database access. +- [Prisma](./prisma) when Prisma owns your database access. +- [Drizzle](./drizzle) when Drizzle owns your database access. - [Custom Stores](./custom-stores) when your app owns the persistence boundary or you are writing an integration. -- [Persistence Internals](./internals) when you need the exact store contracts, - event-log semantics, resume requirements, and SQL/Cloudflare storage maps. +- [Persistence Internals](./internals) when you need the exact store contracts. diff --git a/docs/persistence/prisma.md b/docs/persistence/prisma.md index da6dda2a3..0068d92a9 100644 --- a/docs/persistence/prisma.md +++ b/docs/persistence/prisma.md @@ -3,67 +3,190 @@ title: Persistence with Prisma id: prisma --- -Use Prisma persistence when Prisma already owns your SQLite or Postgres -database connection and migration workflow. The adapter wraps Prisma's raw SQL -methods and exposes the same `AIPersistence` stores used by -`withChatPersistence(...)`. +`@tanstack/ai-persistence-prisma` is the Prisma backend for TanStack AI +**state** persistence. It is **bring-your-own-client**: you add the shipped +model fragment to your own `schema.prisma`, run `prisma migrate` through your +normal workflow, then hand the generated `PrismaClient` to +`prismaPersistence(...)`. It returns the same `AIPersistence` contract consumed +by `withChatPersistence(...)` and `withGenerationPersistence(...)`. -By the end, your Prisma-backed server can persist chat messages, replay events, -interrupts, metadata, and other SQL-backed stores without adding a second -database client. +Prefer [Drizzle](./drizzle) if you want a batteries-included `sqlPersistence` +with bundled migrations. Reach for Prisma when Prisma already owns your database +connection and migration workflow. -## Generate the migration +```sh +pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-prisma @prisma/client +pnpm add -D prisma +``` + +## Add the models to your schema + +Copy the models from this package's `prisma/schema.prisma` into your own Prisma +schema (or import them if you split your schema across files). The fragment +defines six models — `Message`, `Run`, `Interrupt`, `Metadata`, `Artifact`, and +`Blob` — that back the `AIPersistence` state stores: + +```prisma +model Message { + threadId String @id @map("thread_id") + messagesJson String @map("messages_json") + + @@map("messages") +} + +model Run { + runId String @id @map("run_id") + threadId String @map("thread_id") + status String + startedAt BigInt @map("started_at") + finishedAt BigInt? @map("finished_at") + error String? + usageJson String? @map("usage_json") + + @@map("runs") +} + +model Interrupt { + interruptId String @id @map("interrupt_id") + runId String @map("run_id") + threadId String @map("thread_id") + status String + requestedAt BigInt @map("requested_at") + resolvedAt BigInt? @map("resolved_at") + payloadJson String @map("payload_json") + responseJson String? @map("response_json") + + @@map("interrupts") +} + +model Metadata { + scope String + key String + valueJson String @map("value_json") + + @@id([scope, key]) + @@map("metadata") +} + +model Artifact { + artifactId String @id @map("artifact_id") + runId String @map("run_id") + threadId String @map("thread_id") + name String + mimeType String @map("mime_type") + size BigInt + externalUrl String? @map("external_url") + createdAt BigInt @map("created_at") + + @@map("artifacts") +} + +model Blob { + key String @id + contentType String? @map("content_type") + size BigInt? + etag String? + customMetadataJson String? @map("custom_metadata_json") + createdAt BigInt? @map("created_at") + updatedAt BigInt? @map("updated_at") + body Bytes? + + @@map("blobs") +} +``` + +JSON-valued fields are stored in `*_json` `String` (TEXT) columns because +Prisma's `Json` type is unavailable on SQLite; the adapter serializes and parses +them for you. Integer columns use `BigInt` so epoch-millisecond timestamps fit +the full 64-bit range — the adapter converts `bigint` back to `number` at the +boundary, so the records you read are plain numbers. -Create a Prisma migration file for your dialect. +## Run the migration + +Generate and apply a migration with your normal Prisma workflow: ```sh -pnpm exec tanstack-ai-persistence-prisma --dialect postgres +pnpm exec prisma migrate dev --name tanstack_ai_persistence ``` -The default output path is -`prisma/migrations/_tanstack_ai_persistence/migration.sql`. You can -also pass `--out`, `--stdout`, `--timestamp`, `--name`, and `--force`. - -Run the generated migration with your normal Prisma workflow before deploying. -Lazy migrations are opt-in; use `migrate: true` only for local or development -databases. +For production, generate the migration ahead of time and apply it with +`prisma migrate deploy` from your deployment pipeline. ## Create the persistence object -Pass your Prisma client and dialect to `prismaPersistence(...)`. +Pass your generated `PrismaClient` to `prismaPersistence(...)`: ```ts +import { PrismaClient } from '@prisma/client' import { prismaPersistence } from '@tanstack/ai-persistence-prisma' import { withChatPersistence } from '@tanstack/ai-persistence' -import { prisma } from './prisma' -export const persistence = prismaPersistence({ - prisma, - dialect: 'postgres', -}) +const prisma = new PrismaClient() -export const middleware = withChatPersistence(persistence, { - features: ['messages', 'durable-replay', 'interrupts', 'metadata'], -}) +export const persistence = prismaPersistence(prisma) +export const middleware = withChatPersistence(persistence) ``` -The adapter uses Prisma's `$queryRawUnsafe`, `$executeRawUnsafe`, and -`$transaction` methods behind the `SqlDriver` contract. It supports SQLite and -Postgres. +The middleware then flows into your chat handler like any other. On the server: + +```ts +import { chat, toServerSentEventsResponse } from '@tanstack/ai' +import { openaiText } from '@tanstack/ai-openai' +import { middleware } from './persistence' + +export async function POST(request: Request) { + const body: unknown = await request.json() + const messages = Array.isArray(body) ? body : [] + + const stream = chat({ + adapter: openaiText('gpt-5.5'), + messages, + middleware: [middleware], + }) + + return toServerSentEventsResponse(stream) +} +``` + +## What is persisted + +The models mirror the `AIPersistence` state records: + +- `Message` — thread message history +- `Run` — run lifecycle (status, usage, timing) +- `Interrupt` — interrupts / approvals +- `Metadata` — scoped key/value metadata +- `Artifact` — generation artifact references +- `Blob` — generic blob objects + +Delivery durability (resuming an interrupted stream) is a **transport** concern +and is not stored here. Locks are not part of the SQL schema; an in-memory lock +is provided as a dev default. Swap in a distributed lock for multi-process +deployments via [Custom Stores](./custom-stores). + +## Keep the Drizzle and Prisma schemas in sync + +> **Coupling: `persistence-schema-dual-source`.** The Prisma models above and the +> Drizzle schema in `@tanstack/ai-persistence-drizzle` describe the **same** +> state tables and have **no** auto-converter between them. Changing one without +> the other silently diverges the two backends. + +Any change to either schema requires all three of: + +1. the sibling schema updated to match, column-for-column; +2. regenerated migrations for **both** ORMs (`drizzle-kit generate` and + `prisma migrate dev`); +3. the shared conformance suite re-run against memory, Drizzle, and Prisma. + +The two schemas stay column-for-column identical: Drizzle's `integer()` columns +and Prisma's `BigInt` columns are both 64-bit INTEGER-affinity columns in +SQLite, so the on-disk shape matches. -## Use it for different persistence goals +## Use it across the guides -Use the same Prisma-backed `AIPersistence` object across the topic guides: +The same Prisma-backed `AIPersistence` object works across the topic guides: -- [Chat Persistence](./chat-persistence) for server-owned transcripts and - replay cursors. +- [Chat Persistence](./chat-persistence) for server-owned transcripts. - [Persistence Controls](./controls) when you need to choose a feature list. -- [MCP Persistence](./mcp-persistence) when MCP session ids or tool-call - correlation should live in metadata and internal events. -- [Custom Stores](./custom-stores) when you want to keep Prisma for SQL state - but provide a separate object store for blobs. - -If generated media or file artifacts must be durable, Prisma can own the SQL -state, but you still need `stores.artifacts` and `stores.blobs`. Implement that -hybrid shape with [Custom Stores](./custom-stores), or use a backend such as -[Cloudflare](./cloudflare) when D1 plus R2 fits your deployment. +- [Custom Stores](./custom-stores) when you want Prisma for SQL state but a + separate object store for blobs. diff --git a/docs/persistence/sandbox-persistence.md b/docs/persistence/sandbox-persistence.md index 114923913..5eab9eceb 100644 --- a/docs/persistence/sandbox-persistence.md +++ b/docs/persistence/sandbox-persistence.md @@ -17,7 +17,7 @@ Install the sandbox package, persistence package, and the persistence backend you already use for durable chat runs: ```sh -pnpm add @tanstack/ai-sandbox @tanstack/ai-persistence @tanstack/ai-persistence-sqlite +pnpm add @tanstack/ai-sandbox @tanstack/ai-persistence @tanstack/ai-persistence-drizzle ``` ## Persist sandbox identity @@ -30,13 +30,14 @@ tenant. import { chat } from '@tanstack/ai' import { claudeCodeText } from '@tanstack/ai-claude-code' import { withChatPersistence } from '@tanstack/ai-persistence' -import { sqlitePersistence } from '@tanstack/ai-persistence-sqlite' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' import { defineSandbox, withSandbox } from '@tanstack/ai-sandbox' import { dockerSandbox } from '@tanstack/ai-sandbox-docker' import type { ModelMessage } from '@tanstack/ai' -const persistence = sqlitePersistence({ - path: '.tanstack-ai/state.sqlite', +const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:.tanstack-ai/state.sqlite', migrate: true, }) @@ -76,9 +77,11 @@ Without a durable lock, two workers may try to ensure the same sandbox at the same time. Without persistence metadata, resume only works inside the process that still has the in-memory sandbox record. -On Cloudflare, use Durable Object locks through -[Cloudflare persistence](./cloudflare). On Node, use a backend-specific lock -store when more than one process can resume the same sandbox key. +The batteries-included backends ship an in-memory lock (dev default). For +multi-worker deployments, provide a distributed lock store — a Durable Object +lock on Cloudflare (see [Cloudflare persistence](./cloudflare)), or another +backend-specific lock on Node — when more than one process can resume the same +sandbox key. ## Persist workspace files @@ -87,11 +90,12 @@ filesystem is gone?" It uses `stores.metadata` for the workspace manifest, `stores.artifacts` for file records, `stores.blobs` for file bytes, and usually `stores.locks` for multi-worker updates. -```ts +```ts ignore import { chat } from '@tanstack/ai' import { claudeCodeText } from '@tanstack/ai-claude-code' -import { cloudflarePersistence } from '@tanstack/ai-persistence-cloudflare' +import { drizzlePersistence } from '@tanstack/ai-persistence-drizzle' import { withChatPersistence } from '@tanstack/ai-persistence' +import { drizzle } from 'drizzle-orm/d1' import { defineSandbox, defineWorkspace, @@ -101,18 +105,13 @@ import { cloudflareSandbox, type Sandbox } from '@tanstack/ai-sandbox-cloudflare interface Env { AI_D1: D1Database - AI_BLOBS: R2Bucket - AI_LOCKS: DurableObjectNamespace Sandbox: DurableObjectNamespace } export function runProjectBuilder(env: Env) { - const persistence = cloudflarePersistence({ - d1: env.AI_D1, - r2: env.AI_BLOBS, - durableObjects: env.AI_LOCKS, - migrate: true, - }) + // D1 is SQLite-compatible, so the Drizzle D1 adapter satisfies the state + // store contract. Run migrations from the exported schema via drizzle-kit. + const persistence = drizzlePersistence(drizzle(env.AI_D1)) const projectSandbox = defineSandbox({ id: 'project-builder', @@ -143,14 +142,7 @@ export function runProjectBuilder(env: Env) { messages: [{ role: 'user', content: 'Build the app.' }], middleware: [ withChatPersistence(persistence, { - features: [ - 'messages', - 'durable-replay', - 'metadata', - 'artifacts', - 'blobs', - 'locks', - ], + features: ['messages', 'metadata', 'artifacts', 'blobs', 'locks'], }), withSandbox(projectSandbox), ], diff --git a/docs/persistence/sql-backends.md b/docs/persistence/sql-backends.md index 89ef63d3f..320444fad 100644 --- a/docs/persistence/sql-backends.md +++ b/docs/persistence/sql-backends.md @@ -4,139 +4,84 @@ id: sql-backends --- Use a SQL backend when your app runs in Node or another server runtime with a -database connection. The raw SQLite and Postgres packages expose the same -`AIPersistence` stores to `withChatPersistence(...)`. The shared SQL core also -supports MySQL DDL for custom drivers, but there is no standalone MySQL backend -package. +database connection. TanStack AI ships a single, Drizzle-backed SQL backend — +`@tanstack/ai-persistence-drizzle` — that exposes the `AIPersistence` state +stores to `withChatPersistence(...)` and `withGenerationPersistence(...)`. -For production, treat the SQL schema as part of your app-owned persistence -boundary. Generate or copy the TanStack AI DDL into your normal migration -system, review it like any other schema change, and deploy it before traffic -uses the persistence stores. Use lazy migration only when you intentionally want -the backend to create tables at runtime. See [Migrations](./migrations) for the -full CLI reference. +There are two entry points, both returning the same contract: -## Raw SQL backends +- `sqlPersistence({ dialect, url, migrate })` — batteries-included: builds the + database and applies the bundled migrations. +- `drizzlePersistence(db)` — bring your own Drizzle database and migration + workflow, driven by the exported `schema`. -SQLite is file-backed and works well for local apps, prototypes, and single-node -deployments. Fresh development examples can set `migrate: true` explicitly so a -new local database creates its tables on first use. - -```ts -import { sqlitePersistence } from '@tanstack/ai-persistence-sqlite' - -export const persistence = sqlitePersistence({ - path: '.tanstack-ai/state.sqlite', - migrate: true, -}) +```sh +pnpm add @tanstack/ai-persistence @tanstack/ai-persistence-drizzle drizzle-orm ``` -Postgres is the usual durable choice for replicated server deployments. +## Batteries-included + +The `sqlite` dialect ships pre-generated migrations, so a fresh database can +create its tables on first use with `migrate: true`. ```ts -import { postgresPersistence } from '@tanstack/ai-persistence-postgres' +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' -export const persistence = postgresPersistence({ - connectionString: process.env.DATABASE_URL ?? '', +export const persistence = sqlPersistence({ + dialect: 'sqlite', + url: 'file:./.tanstack-ai/state.sqlite', migrate: true, }) ``` -Both raw backends use the shared SQL core. They do not create persistence -tables by default. `migrate` is `false` unless you pass `migrate: true`, which -opts into lazy table creation on first use. - -## Shared schema - -The base SQL schema is intentionally small: - -- `runs` -- `public_events` -- `internal_events` -- `messages` -- `interrupts` -- `metadata` -- `_tanstack_ai_migrations` - -Public events are the replayable AG-UI stream. Internal events are separate -package or app checkpoints and are not returned to reconnecting chat clients. - -## Own migrations - -If your deployment applies migrations separately, leave `migrate` unset or set -it to `false`, and run the exported DDL yourself. +Use `url: ':memory:'` for tests. For production, generate and review the +migrations ahead of time and deploy them before traffic reaches the persistence +stores; leave `migrate` unset so tables are not created lazily at runtime. -```ts -import { ddl } from '@tanstack/ai-persistence-sql' -import { postgresPersistence } from '@tanstack/ai-persistence-postgres' - -export const persistence = postgresPersistence({ - connectionString: process.env.DATABASE_URL ?? '', -}) - -export const migrationStatements = ddl('postgres') -``` +## Bring your own Drizzle database -Apply those statements with your normal migration system before traffic reaches -the app. +When Drizzle already owns your database access, pass your `db`. This works with +any Drizzle sqlite driver (`better-sqlite3`, `libsql`/Turso, D1, `node:sqlite`). -## Generate SQL migrations +```ts ignore +import { drizzle } from 'drizzle-orm/better-sqlite3' +import Database from 'better-sqlite3' +import { drizzlePersistence, schema } from '@tanstack/ai-persistence-drizzle' -Use the generic SQL CLI when you want a migration file without importing DDL in -application code. It emits the shared SQL persistence schema for every SQL -dialect supported by the core generator: SQLite, Postgres, and MySQL. +const db = drizzle(new Database('state.sqlite'), { schema }) -```sh -pnpm exec tanstack-ai-persistence-sql --dialect sqlite --out migrations/001_tanstack_ai.sql -pnpm exec tanstack-ai-persistence-sql --dialect postgres --stdout -pnpm exec tanstack-ai-persistence-sql --dialect mysql --out migrations/001_tanstack_ai.sql +export const persistence = drizzlePersistence(db) ``` -Options: - -| Option | Purpose | -| --- | --- | -| `--dialect sqlite\|postgres\|mysql` | Choose the SQL dialect to generate. | -| `--out ` | Write SQL to a migration file. | -| `--stdout` | Print SQL instead of writing a file. | -| `--timestamp ` | Use a deterministic timestamp in generated default names. | -| `--name ` | Use a deterministic migration name. | -| `--force` | Overwrite an existing output file. | - -The generic CLI writes only the shared SQL persistence schema. Cloudflare R2 -artifact indexes, ORM-specific migration folder layouts, and app-owned tables -remain separate. - -## ORM-backed SQL +See [Drizzle](./drizzle) for the full adapter guide and [Migrations](./migrations) +for the drizzle-kit workflow. -Use [Prisma](./prisma) or [Drizzle](./drizzle) when those clients already own -your SQLite or Postgres database access and migration workflow. They use the -same shared SQL stores and feature behavior as the raw SQL backends. +## Schema -## MySQL-compatible deployments +The schema mirrors the `AIPersistence` state records column-for-column: -There is no `@tanstack/ai-persistence-mysql` package today. If your app needs a -MySQL-compatible database, use the shared SQL core from -`@tanstack/ai-persistence-sql` through an adapter that provides a `SqlDriver` -with `dialect: 'mysql'`. The Prisma and Drizzle persistence adapters currently -support SQLite and Postgres. +- `messages` +- `runs` +- `interrupts` +- `metadata` +- `artifacts` +- `blobs` -```ts -import { ddl } from '@tanstack/ai-persistence-sql' -export const migrationStatements = ddl('mysql') -``` +Delivery durability (replaying an interrupted stream) is a transport concern and +is not stored in these tables. Locks are not part of the SQL schema; an +in-memory lock is provided as a dev default — see [Custom Stores](./custom-stores) +to plug in a distributed lock. -You can also generate the same MySQL DDL with the generic CLI: +## Other dialects -```sh -pnpm exec tanstack-ai-persistence-sql --dialect mysql --out migrations/tanstack_ai.sql -``` +`sqlPersistence` bundles the `sqlite` dialect. For `postgres` or `mysql`, +construct a Drizzle database with your own driver and use `drizzlePersistence(db)`, +generating migrations from the exported `schema` with drizzle-kit. Prisma users +can use [Prisma](./prisma) as a peer backend. ## Choosing a backend -Use SQLite when one process owns the database file. Use Postgres when multiple -app instances need the same durable run state. Use Prisma or Drizzle when those -clients are already how your application accesses a SQLite or Postgres -database. Use the shared SQL core when you need a MySQL-compatible custom -driver. Use [Cloudflare](./cloudflare) instead of the Node SQL backends when -your runtime is Workers and your database is D1. +Use `sqlPersistence` with sqlite when one process owns the database file or for +local development. Use `drizzlePersistence(db)` when Drizzle already owns your +database connection — including Postgres, MySQL, libsql/Turso, or D1 — and you +manage migrations yourself. diff --git a/examples/ts-react-chat/package.json b/examples/ts-react-chat/package.json index 332fd682c..e13e03e6c 100644 --- a/examples/ts-react-chat/package.json +++ b/examples/ts-react-chat/package.json @@ -35,7 +35,7 @@ "@tanstack/ai-opencode": "workspace:*", "@tanstack/ai-openrouter": "workspace:*", "@tanstack/ai-persistence": "workspace:*", - "@tanstack/ai-persistence-sql": "workspace:*", + "@tanstack/ai-persistence-drizzle": "workspace:*", "@tanstack/ai-react": "workspace:*", "@tanstack/ai-react-ui": "workspace:*", "@tanstack/ai-sandbox": "workspace:*", @@ -54,7 +54,6 @@ "@tanstack/store": "^0.8.0", "highlight.js": "^11.11.1", "lucide-react": "^0.561.0", - "mysql2": "^3.22.5", "react": "^19.2.3", "react-dom": "^19.2.3", "react-markdown": "^10.1.0", diff --git a/examples/ts-react-chat/src/lib/mysql-persistence.test.ts b/examples/ts-react-chat/src/lib/mysql-persistence.test.ts deleted file mode 100644 index 4b2ec5cad..000000000 --- a/examples/ts-react-chat/src/lib/mysql-persistence.test.ts +++ /dev/null @@ -1,94 +0,0 @@ -import { describe, expect, it } from 'vitest' -import { EventType } from '@tanstack/ai' -import { createMysqlPersistence } from './mysql-persistence' -import type { Pool, PoolConnection } from 'mysql2/promise' - -function createFakePool() { - const calls: Array = [] - let inserted = false - const connection = { - async execute(sql: string) { - calls.push(sql) - if (sql.includes('SELECT seq, event FROM public_events')) { - return [ - inserted - ? [ - { - seq: 1, - event: JSON.stringify({ - type: EventType.TEXT_MESSAGE_CONTENT, - messageId: 'm1', - delta: 'hi', - timestamp: 1, - }), - }, - ] - : [], - [], - ] - } - if (sql.includes('SELECT MAX(seq) AS max_seq')) { - return [[{ max_seq: 0 }], []] - } - if (sql.includes('INSERT INTO public_events')) { - inserted = true - } - return [[], []] - }, - async beginTransaction() { - calls.push('BEGIN') - }, - async commit() { - calls.push('COMMIT') - }, - async rollback() { - calls.push('ROLLBACK') - }, - release() { - calls.push('RELEASE') - }, - } as unknown as PoolConnection - const pool = { - async getConnection() { - return connection - }, - async execute(sql: string) { - calls.push(sql) - return [[], []] - }, - } as unknown as Pool - return { calls, pool } -} - -describe('createMysqlPersistence', () => { - it('sets READ COMMITTED before beginning store transactions', async () => { - const { calls, pool } = createFakePool() - - const persistence = createMysqlPersistence(pool) - await persistence.stores.publicEvents!.append({ - runId: 'run-1', - expectedSeq: 0, - event: { - type: EventType.TEXT_MESSAGE_CONTENT, - messageId: 'm1', - delta: 'hi', - timestamp: 1, - }, - }) - - const beginIndex = calls.indexOf('BEGIN') - expect(beginIndex).toBeGreaterThan(0) - expect(calls[beginIndex - 1]).toBe( - 'SET TRANSACTION ISOLATION LEVEL READ COMMITTED', - ) - }) - - it('rejects undefined bind parameter values before mysql2 execution', async () => { - const { pool } = createFakePool() - const persistence = createMysqlPersistence(pool) - - await expect( - persistence.stores.metadata!.set('scope', 'key', undefined), - ).rejects.toThrow('Unsupported MySQL bind parameter value') - }) -}) diff --git a/examples/ts-react-chat/src/lib/mysql-persistence.ts b/examples/ts-react-chat/src/lib/mysql-persistence.ts index 71bfb7b04..7c4c8e0be 100644 --- a/examples/ts-react-chat/src/lib/mysql-persistence.ts +++ b/examples/ts-react-chat/src/lib/mysql-persistence.ts @@ -1,132 +1,23 @@ -import mysql from 'mysql2/promise' -import { createSqlPersistence } from '@tanstack/ai-persistence-sql' -import type { - FieldPacket, - Pool, - PoolConnection, - PoolOptions, - QueryResult, - RowDataPacket, -} from 'mysql2/promise' -import type { ExecuteValues } from 'mysql2' -import type { SqlDriver } from '@tanstack/ai-persistence-sql' - -let pool: Pool | undefined - -interface MysqlExecutable { - execute( - sql: string, - values?: Array, - ): Promise<[T, Array]> -} - -function readPort(value: string | undefined): number | undefined { - if (!value) return undefined - const port = Number(value) - return Number.isInteger(port) && port > 0 ? port : undefined -} - -function createPoolOptions(): string | PoolOptions { - if (process.env.MYSQL_URL) return process.env.MYSQL_URL - return { - host: process.env.MYSQL_HOST || '127.0.0.1', - port: readPort(process.env.MYSQL_PORT) ?? 3306, - user: process.env.MYSQL_USER || 'root', - password: process.env.MYSQL_PASSWORD || '', - database: process.env.MYSQL_DATABASE || 'tanstack_ai_chat', - waitForConnections: true, - connectionLimit: 10, - } -} - -function getPool(): Pool { - const options = createPoolOptions() - pool ??= - typeof options === 'string' - ? mysql.createPool(options) - : mysql.createPool(options) - return pool -} - -function isExecuteValue(value: unknown): value is ExecuteValues { - if (value === null) return true - if (value === undefined) return false - const type = typeof value - if ( - type === 'string' || - type === 'number' || - type === 'bigint' || - type === 'boolean' - ) { - return true - } - if ( - value instanceof Date || - value instanceof Buffer || - value instanceof Uint8Array - ) { - return true - } - if (Array.isArray(value)) return value.every(isExecuteValue) - if (type === 'object') { - return Object.values(value).every(isExecuteValue) - } - return false -} - -function toExecuteValues(params: ReadonlyArray): Array { - return params.map((param) => { - if (!isExecuteValue(param)) { - throw new TypeError('Unsupported MySQL bind parameter value') - } - return param +import { sqlPersistence } from '@tanstack/ai-persistence-drizzle' +import type { AIPersistence } from '@tanstack/ai-persistence' + +/** + * Durable **state** persistence for the persistent-chat demo. + * + * Persistence v2 ships a batteries-included SQLite backend (Drizzle + bundled + * migrations) via {@link sqlPersistence}. The former hand-rolled MySQL + * `SqlDriver` was removed — for Postgres/MySQL, construct your own Drizzle `db` + * and use `drizzlePersistence(db)` with migrations generated from the exported + * schema. Delivery durability (disconnect → reconnect → ordered resume) is a + * separate transport concern handled by `toServerSentEvents(stream, { durability })`. + */ +let persistence: AIPersistence | undefined + +export function createChatPersistence(): AIPersistence { + persistence ??= sqlPersistence({ + dialect: 'sqlite', + url: process.env.CHAT_DB_URL ?? 'file:./chat-persistence-demo.db', + migrate: true, }) -} - -function createDriver( - execute: MysqlExecutable, - getConnection: () => Promise, -): SqlDriver { - const driver: SqlDriver = { - dialect: 'mysql' as SqlDriver['dialect'], - async exec(sql, params = []) { - await execute.execute(sql, toExecuteValues(params)) - }, - async query = Record>( - sql: string, - params: ReadonlyArray = [], - ): Promise> { - const [rows] = await execute.execute>( - sql, - toExecuteValues(params), - ) - return rows.map((row) => ({ ...row })) as Array - }, - async transaction(fn) { - const connection = await getConnection() - const tx = createDriver(connection, getConnection) - try { - await connection.execute( - 'SET TRANSACTION ISOLATION LEVEL READ COMMITTED', - ) - await connection.beginTransaction() - const result = await fn(tx) - await connection.commit() - return result - } catch (error) { - await connection.rollback() - throw error - } finally { - connection.release() - } - }, - } - return driver -} - -export function createMysqlPersistence(mysqlPool: Pool = getPool()) { - return createSqlPersistence( - createDriver(mysqlPool, () => mysqlPool.getConnection()), - { migrate: true }, - ) + return persistence } diff --git a/examples/ts-react-chat/src/routes/api.mysql-persistent-chat.ts b/examples/ts-react-chat/src/routes/api.mysql-persistent-chat.ts index b09f78432..860fb6497 100644 --- a/examples/ts-react-chat/src/routes/api.mysql-persistent-chat.ts +++ b/examples/ts-react-chat/src/routes/api.mysql-persistent-chat.ts @@ -2,230 +2,22 @@ import { createFileRoute } from '@tanstack/react-router' import { chat, chatParamsFromRequestBody, - EventType, maxIterations, + memoryStream, toServerSentEventsResponse, } from '@tanstack/ai' import { openaiText } from '@tanstack/ai-openai' -import { decodeCursor, withChatPersistence } from '@tanstack/ai-persistence' -import { createMysqlPersistence } from '@/lib/mysql-persistence' -import type { RunErrorEvent, StreamChunk } from '@tanstack/ai' -import type { RunRecord } from '@tanstack/ai-persistence' +import { withChatPersistence } from '@tanstack/ai-persistence' +import { createChatPersistence } from '@/lib/mysql-persistence' const SYSTEM_PROMPT = `You are a concise assistant in a durable chat demo. -Streams are persisted to MySQL so the browser can refresh and resume an -in-progress response. Keep answers short enough that the streaming behavior is -easy to inspect.` +Chat state (messages, run status, interrupts) is persisted via the state +middleware. Delivery durability — so the browser can refresh and reconnect to an +in-progress response — is handled by the transport helper's durability sink. +Keep answers short enough that the streaming behavior is easy to inspect.` -const mysqlPersistence = createMysqlPersistence() -const activeRuns = new Map>() -const TERMINAL_STATUSES = new Set(['completed', 'failed', 'interrupted']) -const POLL_DELAYS_MS = [100, 150, 250, 500, 750, 1000] as const - -type ChatParams = Awaited> - -function jsonError(message: string, status: number): Response { - return new Response(JSON.stringify({ error: message }), { - status, - headers: { 'Content-Type': 'application/json' }, - }) -} - -function runError(message: string, run?: RunRecord): RunErrorEvent { - return { - type: EventType.RUN_ERROR, - timestamp: Date.now(), - message, - ...(run ? { runId: run.runId, threadId: run.threadId } : {}), - } -} - -function isTerminal(run: RunRecord): boolean { - return TERMINAL_STATUSES.has(run.status) -} - -function delay(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)) -} - -async function ensureProducer(params: ChatParams): Promise { - const runId = params.runId - const existing = activeRuns.get(runId) - if (existing) { - if (!params.resume) return - await existing.catch(() => undefined) - const current = activeRuns.get(runId) - if (current && current !== existing) return - if (current === existing) activeRuns.delete(runId) - } - - const promise = (async () => { - const stream = chat({ - adapter: openaiText('gpt-5.5'), - middleware: [ - withChatPersistence(mysqlPersistence, { - features: ['messages', 'durable-replay', 'interrupts'], - }), - ], - systemPrompts: [SYSTEM_PROMPT], - agentLoopStrategy: maxIterations(4), - messages: params.messages, - threadId: params.threadId, - runId: params.runId, - cursor: params.cursor, - resume: params.resume, - }) - - for await (const _chunk of stream) { - // Drain the model stream so withChatPersistence writes every public event to - // MySQL. Browser disconnects only cancel response tailing, not this run. - } - })() - - activeRuns.set(runId, promise) - promise - .catch((error) => { - console.error('[mysql-persistent-chat] Producer failed:', error) - }) - .finally(() => { - if (activeRuns.get(runId) === promise) activeRuns.delete(runId) - }) -} - -async function validateCursorRequest( - params: ChatParams, - decodedCursor: { runId: string; seq: number }, -): Promise { - if (decodedCursor.runId !== params.runId) { - return jsonError('Cursor runId does not match request runId.', 409) - } - if (decodedCursor.seq < 1) { - return jsonError('Cursor sequence must reference a persisted event.', 400) - } - - const run = await mysqlPersistence.stores.runs!.get(params.runId) - if (!run) { - return jsonError('Run for cursor was not found.', 404) - } - if (run.threadId !== params.threadId) { - return jsonError( - 'Cursor run threadId does not match request threadId.', - 409, - ) - } - - const latestSeq = await mysqlPersistence.stores.publicEvents!.latestSeq( - params.runId, - ) - if (latestSeq === 0) { - return jsonError('Run has no persisted public events.', 404) - } - if (latestSeq < decodedCursor.seq) { - return jsonError('Cursor is ahead of persisted public events.', 409) - } - - let foundCursorEvent = false - for await (const event of mysqlPersistence.stores.publicEvents!.read( - params.runId, - { afterSeq: decodedCursor.seq - 1 }, - )) { - foundCursorEvent = event.seq === decodedCursor.seq - break - } - if (!foundCursorEvent) { - return jsonError('Cursor event was not found.', 404) - } - - if ( - run.status === 'running' && - !params.resume && - !activeRuns.has(params.runId) - ) { - return jsonError( - 'Run is still marked running, but no in-process producer is active. Restarting producers after a server process restart is out of scope for this example.', - 409, - ) - } - - if (run.status === 'failed' && latestSeq <= decodedCursor.seq) { - return jsonError( - 'Run failed and no newer persisted terminal event is available.', - 409, - ) - } - - return null -} - -async function* tailPersistedEvents( - runId: string, - afterSeq: number, - signal: AbortSignal, -): AsyncIterable { - let seq = afterSeq - let idlePolls = 0 - - while (!signal.aborted) { - let yielded = false - for await (const persisted of mysqlPersistence.stores.publicEvents!.read( - runId, - seq > 0 ? { afterSeq: seq } : undefined, - )) { - if (signal.aborted) return - seq = persisted.seq - yielded = true - idlePolls = 0 - yield persisted.event - } - - const run = await mysqlPersistence.stores.runs!.get(runId) - const activeProducer = activeRuns.has(runId) - if (!run) { - if (activeProducer) { - if (!yielded) { - const delayMs = - POLL_DELAYS_MS[Math.min(idlePolls, POLL_DELAYS_MS.length - 1)] - idlePolls += 1 - await delay(delayMs) - } - continue - } - yield runError('Run disappeared while tailing persisted events.') - return - } - - if (run.status === 'running' && !activeProducer) { - yield runError( - 'Run is still marked running, but no in-process producer is active. Restarting producers after a server process restart is out of scope for this example.', - run, - ) - return - } - - if (isTerminal(run) && !activeProducer) { - const latestSeq = - await mysqlPersistence.stores.publicEvents!.latestSeq(runId) - if (latestSeq <= seq) { - if (run.status === 'failed' && !yielded) { - yield runError( - run.error || - 'Run failed before any newer terminal event was available.', - run, - ) - } - return - } - } - - if (!yielded) { - const delayMs = - POLL_DELAYS_MS[Math.min(idlePolls, POLL_DELAYS_MS.length - 1)] - idlePolls += 1 - await delay(delayMs) - } - } -} +const persistence = createChatPersistence() export const Route = createFileRoute('/api/mysql-persistent-chat')({ server: { @@ -241,58 +33,32 @@ export const Route = createFileRoute('/api/mysql-persistent-chat')({ ) } - try { - let decodedCursor: { runId: string; seq: number } | undefined - if (params.cursor) { - try { - decodedCursor = decodeCursor(params.cursor) - } catch { - return jsonError('Invalid cursor.', 400) - } - const invalidCursorResponse = await validateCursorRequest( - params, - decodedCursor, - ) - if (invalidCursorResponse) return invalidCursorResponse - } - const runId = decodedCursor?.runId ?? params.runId - const afterSeq = decodedCursor?.seq ?? 0 - - if (!params.cursor || params.resume) { - await ensureProducer(params) - } else { - const run = await mysqlPersistence.stores.runs!.get(runId) - if (!run) return jsonError('Run was not found.', 404) - } - - const responseAbortController = new AbortController() - request.signal.addEventListener( - 'abort', - () => responseAbortController.abort(), - { once: true }, - ) - - return toServerSentEventsResponse( - tailPersistedEvents( - runId, - afterSeq, - responseAbortController.signal, - ), - { abortController: responseAbortController }, - ) - } catch (error) { - console.error('[mysql-persistent-chat] Chat request failed:', error) - return new Response( - JSON.stringify({ - error: - error instanceof Error ? error.message : 'An error occurred', + // State durability: persist messages / run status / interrupts at + // boundaries. This is lazy — the provider call only fires on the first + // `for await` inside the transport helper. + const stream = chat({ + adapter: openaiText('gpt-5.5'), + middleware: [ + withChatPersistence(persistence, { + features: ['messages', 'interrupts'], }), - { - status: 500, - headers: { 'Content-Type': 'application/json' }, - }, - ) - } + ], + systemPrompts: [SYSTEM_PROMPT], + agentLoopStrategy: maxIterations(4), + messages: params.messages, + threadId: params.threadId, + runId: params.runId, + ...(params.resume ? { resume: params.resume } : {}), + }) + + // Delivery durability: the durability sink appends each batch and tags + // every SSE event with an `id: runId@seq` offset, so a reconnect + // (native Last-Event-ID) or second-tab join replays the ordered stream + // exactly once without re-invoking the provider. `memoryStream` is the + // zero-infra dev default; swap in `durableStream(request, { server })` + // for horizontally-scaled deployments. + const durability = memoryStream(request) + return toServerSentEventsResponse(stream, { durability, batch: 8 }) }, }, }, diff --git a/examples/ts-react-chat/src/routes/mysql-persistence.tsx b/examples/ts-react-chat/src/routes/mysql-persistence.tsx index 88dce0b4a..bc8b52183 100644 --- a/examples/ts-react-chat/src/routes/mysql-persistence.tsx +++ b/examples/ts-react-chat/src/routes/mysql-persistence.tsx @@ -74,7 +74,7 @@ function MysqlPersistenceRoute() {
-

MySQL persistent chat

+

Persistent chat

Thread {threadId.slice(0, 8)} {resumeState ? ` · run ${resumeState.runId.slice(0, 8)}` : ''} @@ -95,8 +95,9 @@ function MysqlPersistenceRoute() { {messages.length === 0 && (

Send a prompt, then refresh while the response is streaming. The - same server process continues tailing persisted MySQL events from - the saved run cursor. + transport's delivery-durability sink replays the ordered stream on + reconnect (native Last-Event-ID), while chat state persists to the + SQLite state backend.
)} {messages.map((message) => ( diff --git a/knip.json b/knip.json index 029d47d0d..f7fbcb313 100644 --- a/knip.json +++ b/knip.json @@ -58,8 +58,8 @@ "ignore": ["src/use-chat-context.ts"] }, "packages/ai-bedrock": {}, - "packages/ai-persistence-postgres": { - "ignoreDependencies": ["pg"] + "packages/ai-persistence": { + "ignoreDependencies": ["vitest"] } } } diff --git a/packages/ai-angular/src/inject-chat.ts b/packages/ai-angular/src/inject-chat.ts index a61666a72..70c961681 100644 --- a/packages/ai-angular/src/inject-chat.ts +++ b/packages/ai-angular/src/inject-chat.ts @@ -80,14 +80,6 @@ export function injectChat< ? { connection: options.connection } : { fetcher: options.fetcher } - let lifecycleOwned = false - const resumeIfLifecycleOwned = () => { - if (!lifecycleOwned) { - return - } - void client.maybeAutoResume() - } - const client = new ChatClient({ devtoolsBridgeFactory: createChatDevtoolsBridge, ...transport, @@ -119,7 +111,6 @@ export function injectChat< onError: (err) => options.onError?.(err), onResumeStateChange: (resumeState, pendingInterrupts) => { options.onResumeStateChange?.(resumeState, pendingInterrupts) - resumeIfLifecycleOwned() }, tools: options.tools, onCustomEvent: (eventType, data, context) => @@ -173,25 +164,15 @@ export function injectChat< afterNextRender( () => { - lifecycleOwned = true client.mountDevtools() - void client.maybeAutoResume() - if (typeof window !== 'undefined') { - window.addEventListener('online', handleOnline) - } + // Delivery-durability resume is transparent: the resumable SSE + // connection adapter reattaches via the browser's native Last-Event-ID + // on reconnect. No client-side auto-resume wiring is needed. }, { injector }, ) - const handleOnline = () => { - void client.maybeAutoResume() - } - destroyRef.onDestroy(() => { - lifecycleOwned = false - if (typeof window !== 'undefined') { - window.removeEventListener('online', handleOnline) - } if (liveSource?.()) { client.unsubscribe() } else { diff --git a/packages/ai-angular/src/inject-generate-audio.ts b/packages/ai-angular/src/inject-generate-audio.ts index 302fa3424..fbb57607c 100644 --- a/packages/ai-angular/src/inject-generate-audio.ts +++ b/packages/ai-angular/src/inject-generate-audio.ts @@ -24,7 +24,7 @@ export interface InjectGenerateAudioOptions< TOutput = AudioGenerationResult, > extends Pick< InjectGenerationOptions, - 'persistence' | 'autoResume' | 'initialResumeSnapshot' | 'resumeState' + 'persistence' | 'initialResumeSnapshot' > { /** Connect-based adapter for streaming transport (SSE, HTTP stream, custom) */ connection?: ConnectConnectionAdapter diff --git a/packages/ai-angular/src/inject-generate-video.ts b/packages/ai-angular/src/inject-generate-video.ts index 0e7801d05..4e75df991 100644 --- a/packages/ai-angular/src/inject-generate-video.ts +++ b/packages/ai-angular/src/inject-generate-video.ts @@ -38,9 +38,7 @@ export interface InjectGenerateVideoOptions { body?: ReactiveOption> devtools?: AIDevtoolsDisplayOptions persistence?: GenerationPersistenceOptions - autoResume?: boolean initialResumeSnapshot?: GenerationResumeSnapshot - resumeState?: ReactiveOption onResult?: (result: VideoGenerateResult) => TOutput | null | void onError?: (error: Error) => void onProgress?: (progress: number, message?: string) => void @@ -63,7 +61,6 @@ export interface InjectGenerateVideoResult { resumeState: Signal pendingArtifacts: Signal> resultArtifacts: Signal> - resume: (state?: GenerationResumeState) => Promise } // `TTransformed` infers from the `onResult` return position so the callback @@ -119,10 +116,6 @@ export function injectGenerateVideo( const bodySource = options.body !== undefined ? toReactive(options.body) : undefined - const resumeStateSource = - options.resumeState !== undefined - ? toReactive(options.resumeState) - : undefined const baseOptions = { id: clientId, @@ -130,13 +123,9 @@ export function injectGenerateVideo( ...(options.persistence !== undefined && { persistence: options.persistence, }), - ...(options.autoResume !== undefined && { autoResume: options.autoResume }), ...(options.initialResumeSnapshot !== undefined && { initialResumeSnapshot: options.initialResumeSnapshot, }), - ...(resumeStateSource?.() !== undefined && { - resumeState: resumeStateSource(), - }), devtoolsBridgeFactory: createVideoDevtoolsBridge, devtools: { ...options.devtools, @@ -208,45 +197,17 @@ export function injectGenerateVideo( () => { client.updateOptions({ body: bodySource(), - ...(resumeStateSource?.() !== undefined && { - resumeState: resumeStateSource(), - }), - }) - }, - { injector }, - ) - } - - if (resumeStateSource && !bodySource) { - effect( - () => { - const nextResumeState = resumeStateSource() - client.updateOptions({ - ...(nextResumeState !== undefined && { - resumeState: nextResumeState, - }), }) }, { injector }, ) } + // Mount devtools only. Generation runs are never auto-started after render — + // persisted state is read-only for display. afterNextRender( () => { client.mountDevtools() - void client - .maybeAutoResume() - .catch((err: unknown) => { - if (disposed) return - const nextError = err instanceof Error ? err : new Error(String(err)) - options.onError?.(nextError) - error.set(nextError) - status.set('error') - }) - .finally(() => { - if (disposed) return - setResumeSnapshotState(client.getResumeSnapshot()) - }) }, { injector }, ) @@ -255,14 +216,6 @@ export function injectGenerateVideo( client.dispose() }) - const resume = async (state?: GenerationResumeState) => { - const didResume = await client.resume(state) - if (!disposed) { - setResumeSnapshotState(client.getResumeSnapshot()) - } - return didResume - } - return { generate: (input: VideoGenerateInput) => client.generate(input), result: result.asReadonly(), @@ -277,6 +230,5 @@ export function injectGenerateVideo( resumeState: resumeState.asReadonly(), pendingArtifacts: pendingArtifacts.asReadonly(), resultArtifacts: resultArtifacts.asReadonly(), - resume, } } diff --git a/packages/ai-angular/src/inject-generation.ts b/packages/ai-angular/src/inject-generation.ts index 48d4edbe7..542f2c6e6 100644 --- a/packages/ai-angular/src/inject-generation.ts +++ b/packages/ai-angular/src/inject-generation.ts @@ -40,14 +40,10 @@ export interface InjectGenerationOptions { body?: ReactiveOption> /** Display options for TanStack AI Devtools. */ devtools?: AIDevtoolsDisplayOptions - /** Server-side lightweight resume state persistence. */ + /** Server-side lightweight generation state persistence. */ persistence?: GenerationPersistenceOptions - /** Whether to resume a persisted run after render. Defaults to true. */ - autoResume?: boolean - /** Initial lightweight resume snapshot restored by the app. */ + /** Initial lightweight resume snapshot restored by the app (read-only state). */ initialResumeSnapshot?: GenerationResumeSnapshot - /** Explicit run/cursor state to use for the next resume/generation request. */ - resumeState?: ReactiveOption /** * Callback when a result is received. Can optionally return a transformed value. * @@ -81,14 +77,12 @@ export interface InjectGenerationResult { reset: () => void /** Lightweight generation resume snapshot, if one is available */ resumeSnapshot: Signal - /** Current resumable run/cursor state, if one is available */ + /** Observed run/cursor metadata from the snapshot (read-only state) */ resumeState: Signal /** Pending persisted artifact references observed during generation/replay */ pendingArtifacts: Signal> /** Final persisted artifact references observed from a replayed result */ resultArtifacts: Signal> - /** Resume the current/initial generation run, if resumable */ - resume: (state?: GenerationResumeState) => Promise } // `TTransformed` infers from the `onResult` return position (a covariant @@ -146,10 +140,6 @@ export function injectGeneration< const bodySource = options.body !== undefined ? toReactive(options.body) : undefined - const resumeStateSource = - options.resumeState !== undefined - ? toReactive(options.resumeState) - : undefined const clientOptions: GenerationClientOptions = { id: clientId, @@ -157,13 +147,9 @@ export function injectGeneration< ...(options.persistence !== undefined && { persistence: options.persistence, }), - ...(options.autoResume !== undefined && { autoResume: options.autoResume }), ...(options.initialResumeSnapshot !== undefined && { initialResumeSnapshot: options.initialResumeSnapshot, }), - ...(resumeStateSource?.() !== undefined && { - resumeState: resumeStateSource(), - }), devtoolsBridgeFactory: createGenerationDevtoolsBridge, devtools: { ...options.devtools, @@ -222,45 +208,17 @@ export function injectGeneration< () => { client.updateOptions({ body: bodySource(), - ...(resumeStateSource?.() !== undefined && { - resumeState: resumeStateSource(), - }), - }) - }, - { injector }, - ) - } - - if (resumeStateSource && !bodySource) { - effect( - () => { - const nextResumeState = resumeStateSource() - client.updateOptions({ - ...(nextResumeState !== undefined && { - resumeState: nextResumeState, - }), }) }, { injector }, ) } + // Mount devtools only. Generation runs are never auto-started after render — + // persisted state is read-only for display. afterNextRender( () => { client.mountDevtools() - void client - .maybeAutoResume() - .catch((err: unknown) => { - if (disposed) return - const nextError = err instanceof Error ? err : new Error(String(err)) - options.onError?.(nextError) - error.set(nextError) - status.set('error') - }) - .finally(() => { - if (disposed) return - setResumeSnapshotState(client.getResumeSnapshot()) - }) }, { injector }, ) @@ -269,14 +227,6 @@ export function injectGeneration< client.dispose() }) - const resume = async (state?: GenerationResumeState) => { - const didResume = await client.resume(state) - if (!disposed) { - setResumeSnapshotState(client.getResumeSnapshot()) - } - return didResume - } - return { generate: ((input: TInput) => client.generate(input)) as ( input: Record, @@ -291,6 +241,5 @@ export function injectGeneration< resumeState: resumeState.asReadonly(), pendingArtifacts: pendingArtifacts.asReadonly(), resultArtifacts: resultArtifacts.asReadonly(), - resume, } } diff --git a/packages/ai-angular/tests/inject-chat.test.ts b/packages/ai-angular/tests/inject-chat.test.ts index 8f51a3729..6855074af 100644 --- a/packages/ai-angular/tests/inject-chat.test.ts +++ b/packages/ai-angular/tests/inject-chat.test.ts @@ -3,7 +3,7 @@ import { z } from 'zod' import { EventType } from '@tanstack/ai/client' import { Component, signal } from '@angular/core' import { TestBed } from '@angular/core/testing' -import { ChatClient, type ConnectionAdapter } from '@tanstack/ai-client' +import { ChatClient } from '@tanstack/ai-client' import { injectChat } from '../src/inject-chat' import { createMockConnectionAdapter, @@ -137,14 +137,16 @@ describe('injectChat — resume', () => { messageId: 'msg-1', timestamp: Date.now(), delta: 'Hi', - cursor: 'cursor-1', }, { type: EventType.RUN_FINISHED, runId: 'run-1', threadId: 'thread-1', timestamp: Date.now(), - outcome: { type: 'success' }, + outcome: { + type: 'interrupt', + interrupts: [{ id: 'interrupt-1', reason: 'client_tool_input' }], + }, }, ], }) @@ -156,54 +158,16 @@ describe('injectChat — resume', () => { await result.sendMessage('Hi') + // A run that pauses on an interrupt forwards interrupt (state) resume — + // the thread/run ids to target on a follow-up. No delivery cursor. expect(onResumeStateChange).toHaveBeenCalledWith( expect.objectContaining({ threadId: 'thread-1', runId: expect.any(String), - cursor: 'cursor-1', }), - [], + expect.arrayContaining([expect.objectContaining({ id: 'interrupt-1' })]), ) }) - - it('forwards initialResumeSnapshot to ChatClient auto-resume', async () => { - type RunContext = Parameters< - Extract< - ConnectionAdapter, - { - connect: unknown - } - >['connect'] - >[3] - const contexts: Array = [] - const adapter: Extract = { - async *connect(_messages, _data, _signal, runContext) { - contexts.push(runContext) - }, - } - const { flush } = renderInjectChat({ - connection: adapter, - initialResumeSnapshot: { - resumeState: { - threadId: 'thread-1', - runId: 'run-1', - cursor: 'cursor-1', - }, - pendingInterrupts: [], - }, - }) - - flush() - await tick() - - expect(contexts).toEqual([ - expect.objectContaining({ - threadId: 'thread-1', - runId: 'run-1', - cursor: 'cursor-1', - }), - ]) - }) }) describe('injectChat — structured output', () => { diff --git a/packages/ai-angular/tests/inject-generation.test.ts b/packages/ai-angular/tests/inject-generation.test.ts index 8d1433a6e..2ee11685c 100644 --- a/packages/ai-angular/tests/inject-generation.test.ts +++ b/packages/ai-angular/tests/inject-generation.test.ts @@ -65,51 +65,10 @@ const videoResumeSnapshot: GenerationResumeSnapshot = { resumeState: { threadId: 'thread-resume', runId: 'run-resume', - cursor: 'cursor-resume', }, status: 'running', } -function createDeferred() { - let resolve!: (value: T) => void - let reject!: (reason?: unknown) => void - const promise = new Promise((res, rej) => { - resolve = res - reject = rej - }) - return { promise, resolve, reject } -} - -function createReplayVideoChunks(): Array { - return [ - { - type: 'RUN_STARTED', - runId: 'run-resume', - threadId: 'thread-resume', - cursor: 'cursor-start', - timestamp: Date.now(), - }, - { - type: 'CUSTOM', - name: 'generation:result', - value: { - jobId: 'job-replay', - status: 'completed', - url: 'https://example.com/video.mp4', - }, - cursor: 'cursor-result', - timestamp: Date.now(), - }, - { - type: 'RUN_FINISHED', - runId: 'run-resume', - threadId: 'thread-resume', - cursor: 'cursor-finished', - timestamp: Date.now(), - }, - ] as unknown as Array -} - function createRunContextCaptureAdapter(chunks: Array): { adapter: ConnectConnectionAdapter connect: ReturnType @@ -161,92 +120,56 @@ describe('injectGeneration', () => { expect(result.status()).toBe('success') }) - it('ignores auto-resume rejection after destroy', async () => { - const deferred = createDeferred() - const getItem = vi.fn(() => deferred.promise) - const onError = vi.fn() - const { result, destroy } = renderInjectGeneration({ - fetcher: async () => ({ value: 1 }), + it('does not auto-fire a generation after render from a persisted running snapshot', async () => { + // Regression guard for the removed generation resume surface. + const snapshot: GenerationResumeSnapshot = { + resumeState: { threadId: 'thread-resume', runId: 'run-resume' }, + status: 'running', + } + const { adapter, connect } = createRunContextCaptureAdapter([]) + const getItem = vi.fn(() => snapshot) + const { result } = renderInjectGeneration({ + id: 'no-auto-fire', + connection: adapter, persistence: { - server: { - getItem, - setItem: vi.fn(), - removeItem: vi.fn(), - }, + server: { getItem, setItem: vi.fn(), removeItem: vi.fn() }, }, - onError, + initialResumeSnapshot: snapshot, }) - await vi.waitFor(() => expect(getItem).toHaveBeenCalled()) - destroy() - deferred.reject(new Error('resume failed')) - await deferred.promise.catch(() => {}) await Promise.resolve() - expect(onError).not.toHaveBeenCalled() - expect(result.error()).toBeUndefined() + expect(connect).not.toHaveBeenCalled() + expect(getItem).not.toHaveBeenCalled() + expect(result.isLoading()).toBe(false) expect(result.status()).toBe('idle') + // The persisted snapshot remains exposed as read-only state. + expect(result.resumeState()).toEqual(snapshot.resumeState) }) }) describe('injectGenerateVideo', () => { - it('explicitly resumes from the current snapshot', async () => { - const { adapter, connect, runContexts } = createRunContextCaptureAdapter( - createReplayVideoChunks(), - ) - const { result, flush } = renderInjectGenerateVideo({ + it('does not auto-fire a video generation after render from a persisted running snapshot', async () => { + // Regression guard for the removed generation resume surface (video). + const { adapter, connect } = createRunContextCaptureAdapter([]) + const getItem = vi.fn(() => videoResumeSnapshot) + const { result } = renderInjectGenerateVideo({ + id: 'video-no-auto-fire', connection: adapter, - initialResumeSnapshot: videoResumeSnapshot, - autoResume: false, - }) - - const didResume = await result.resume() - flush() - - expect(didResume).toBe(true) - expect(connect).toHaveBeenCalledTimes(1) - expect(runContexts[0]).toEqual(videoResumeSnapshot.resumeState) - expect(result.resumeSnapshot()).toEqual( - expect.objectContaining({ - status: 'complete', - resumeState: null, - }), - ) - expect(result.result()).toEqual( - expect.objectContaining({ - jobId: 'job-replay', - }), - ) - }) - - it('ignores video auto-resume rejection after destroy', async () => { - const deferred = createDeferred() - const getItem = vi.fn(() => deferred.promise) - const onError = vi.fn() - const { result, destroy } = renderInjectGenerateVideo({ - fetcher: async () => ({ - jobId: 'job-1', - status: 'completed', - url: 'https://example.com/video.mp4', - }), persistence: { - server: { - getItem, - setItem: vi.fn(), - removeItem: vi.fn(), - }, + server: { getItem, setItem: vi.fn(), removeItem: vi.fn() }, }, - onError, + initialResumeSnapshot: videoResumeSnapshot, }) - await vi.waitFor(() => expect(getItem).toHaveBeenCalled()) - destroy() - deferred.reject(new Error('video resume failed')) - await deferred.promise.catch(() => {}) await Promise.resolve() - expect(onError).not.toHaveBeenCalled() - expect(result.error()).toBeUndefined() + expect(connect).not.toHaveBeenCalled() + expect(getItem).not.toHaveBeenCalled() + expect(result.isLoading()).toBe(false) expect(result.status()).toBe('idle') + // The persisted snapshot remains exposed as read-only state. + expect(result.resumeSnapshot()).toEqual(videoResumeSnapshot) + expect(result.resumeState()).toEqual(videoResumeSnapshot.resumeState) }) }) diff --git a/packages/ai-client/src/chat-client.ts b/packages/ai-client/src/chat-client.ts index c088196c5..56585ce80 100644 --- a/packages/ai-client/src/chat-client.ts +++ b/packages/ai-client/src/chat-client.ts @@ -129,10 +129,16 @@ export class ChatClient< private readonly persistor?: ChatPersistor private readonly serverPersistence?: ChatServerPersistence private serverPersistenceGeneration = 0 + // Whether the consumer supplied an `onError` callback. Server-persistence is + // best-effort and must never break chat, but its failures should still be + // observable: they are routed to the consumer's `onError` when one exists, + // otherwise surfaced via `console.warn`. + private hasUserOnError = false private disposed = false private currentRunId: string | null = null - // Resume tracking: the latest in-band cursor seen for the active run, so a - // reconnect can replay events after it. Cleared when the run terminates. + // Interrupt-resume tracking: the run/thread of the most recent interrupted + // run, so approvals/client-tool results can be sent back. Cleared when the + // run terminates. This is STATE (interrupt) resume, not delivery/cursor. private lastResume: ChatResumeState | null = null private pendingInterrupts: Array = [] private pendingInterruptRunId: string | null = null @@ -140,12 +146,10 @@ export class ChatClient< string, RunAgentResumeItem >() - private readonly autoResume: boolean - // When set, the next streamResponse() resumes this run/cursor instead of - // starting a fresh run (consumed once). + // When set, the next streamResponse() continues this interrupted run instead + // of starting a fresh run (consumed once). private pendingResumeRunId: string | null = null private pendingResumeThreadId: string | null = null - private pendingResumeCursor: string | null = null private pendingResumeItems: Array | null = null private activeResumeThreadId: string | null = null private activeResumeRunId: string | null = null @@ -223,7 +227,6 @@ export class ChatClient< constructor(options: ChatClientOptions) { this.uniqueId = options.id || this.generateUniqueId('chat') this.threadId = options.threadId || this.generateUniqueId('thread') - this.autoResume = options.autoResume ?? true const persistence = normalizePersistence(options.persistence) if (persistence.client) { this.persistor = new ChatPersistor( @@ -233,6 +236,7 @@ export class ChatClient< ) } this.serverPersistence = persistence.server + this.hasUserOnError = typeof options.onError === 'function' // Both `body` (deprecated) and `forwardedProps` populate the AG-UI // `RunAgentInput.forwardedProps` wire field. They are stored // separately so `updateOptions` can replace one without touching the @@ -506,6 +510,25 @@ export class ChatClient< } } + /** + * Route a caught server-persistence error to an observable sink. Server + * persistence is best-effort and must never break chat, so callers still + * swallow the failure — but it is no longer silent: it goes to the + * consumer's `onError` callback when one was provided, otherwise to + * `console.warn`. + */ + private reportServerPersistenceError(error: unknown): void { + const normalized = error instanceof Error ? error : new Error(String(error)) + if (this.hasUserOnError) { + this.callbacksRef.current.onError(normalized) + } else { + console.warn( + '[TanStack AI] Server persistence adapter error (non-fatal):', + normalized, + ) + } + } + private readInitialResumeSnapshot(): | ChatResumeSnapshot | null @@ -513,7 +536,8 @@ export class ChatClient< | Promise { try { return this.serverPersistence?.getItem(this.threadId) - } catch { + } catch (error) { + this.reportServerPersistenceError(error) return undefined } } @@ -534,8 +558,10 @@ export class ChatClient< this.notifyResumeStateChange() } }) - .catch(() => { - // Persistence adapters are best-effort and must not break chat setup. + .catch((error: unknown) => { + // Persistence adapters are best-effort and must not break chat setup, + // but the failure is surfaced to an observable sink. + this.reportServerPersistenceError(error) }) } @@ -629,109 +655,69 @@ export class ChatClient< } /** - * Observe the in-band resume cursor on each chunk so a reconnect can replay - * after the last seen event. Cleared when the run reaches a terminal event. + * Track interrupt state off the stream's terminal events. A RUN_FINISHED with + * an interrupt outcome records the pending interrupts + the run/thread to + * resume; any other terminal event for the tracked/current run clears that + * state. This is interrupt (state) resume — there is no delivery cursor. */ - private observeResumeCursor(chunk: StreamChunk): void { - if (chunk.type === 'RUN_FINISHED' || chunk.type === 'RUN_ERROR') { - // A server-signaled terminal event completes the run — drop its resume - // state. (A stream that merely ends without a terminal is an interruption - // and keeps its resume state so it can be continued.) - const runId = getChunkRunId(chunk) - const threadId = - 'threadId' in chunk && typeof chunk.threadId === 'string' - ? chunk.threadId - : this.activeResumeThreadId - const cursor = - 'cursor' in chunk && typeof chunk.cursor === 'string' - ? chunk.cursor - : undefined - if ( - chunk.type === 'RUN_FINISHED' && - chunk.outcome?.type === 'interrupt' - ) { - const interruptedRunId = - this.currentRunId && - (cursor || this.lastResume?.runId === this.currentRunId) - ? this.currentRunId - : (runId ?? this.activeResumeRunId ?? this.currentRunId ?? '') - const resumeTarget = - this.lastResume?.runId === interruptedRunId ? this.lastResume : null - if (cursor) { - this.lastResume = { - threadId: threadId ?? this.threadId, - runId: interruptedRunId, - cursor, - } - } else if (!resumeTarget) { - this.pendingInterrupts = [] - this.pendingInterruptRunId = null - this.pendingInterruptResumeItems.clear() - this.notifyResumeStateChange() - return - } - this.pendingInterruptRunId = interruptedRunId - this.pendingInterrupts = [...chunk.outcome.interrupts] - this.pendingInterruptResumeItems.clear() - this.notifyResumeStateChange() - return - } - - const isRunlessSessionError = chunk.type === 'RUN_ERROR' && !runId - const isTrackedRunTerminal = Boolean( - runId && this.lastResume?.runId === runId, - ) - const isPendingInterruptRunTerminal = Boolean( - runId && this.pendingInterruptRunId === runId, - ) - const isCurrentRunTerminal = Boolean( - (runId && this.currentRunId === runId) || - (this.currentRunId && this.lastResume?.runId === this.currentRunId), - ) - const isCurrentStreamTerminal = - this.isLoading && chunk.type === 'RUN_FINISHED' && !runId - if ( - isRunlessSessionError || - isTrackedRunTerminal || - isPendingInterruptRunTerminal || - isCurrentRunTerminal || - isCurrentStreamTerminal - ) { - this.lastResume = null - } - if ( - isRunlessSessionError || - isTrackedRunTerminal || - isPendingInterruptRunTerminal || - isCurrentRunTerminal || - isCurrentStreamTerminal - ) { - this.pendingInterrupts = [] - this.pendingInterruptRunId = null - this.pendingInterruptResumeItems.clear() - } - this.notifyResumeStateChange() + private observeInterruptState(chunk: StreamChunk): void { + if (chunk.type !== 'RUN_FINISHED' && chunk.type !== 'RUN_ERROR') { return } - const cursor = - 'cursor' in chunk && typeof chunk.cursor === 'string' - ? chunk.cursor - : undefined - if (cursor && this.currentRunId) { + const runId = getChunkRunId(chunk) + const threadId = + 'threadId' in chunk && typeof chunk.threadId === 'string' + ? chunk.threadId + : this.activeResumeThreadId + + if (chunk.type === 'RUN_FINISHED' && chunk.outcome?.type === 'interrupt') { + // Track the REQUEST run id (what the client sent) so a resume targets the + // same run even when provider events carry their own run id. + const interruptedRunId = + this.currentRunId ?? runId ?? this.activeResumeRunId ?? '' this.lastResume = { - threadId: this.activeResumeThreadId ?? this.threadId, - runId: this.currentRunId, - cursor, + threadId: threadId ?? this.threadId, + runId: interruptedRunId, } + this.pendingInterruptRunId = interruptedRunId + this.pendingInterrupts = [...chunk.outcome.interrupts] + this.pendingInterruptResumeItems.clear() this.notifyResumeStateChange() + return } + + const isRunlessSessionError = chunk.type === 'RUN_ERROR' && !runId + const isTrackedRunTerminal = Boolean( + runId && this.lastResume?.runId === runId, + ) + const isPendingInterruptRunTerminal = Boolean( + runId && this.pendingInterruptRunId === runId, + ) + const isCurrentRunTerminal = Boolean( + (runId && this.currentRunId === runId) || + (this.currentRunId && this.lastResume?.runId === this.currentRunId), + ) + const isCurrentStreamTerminal = + this.isLoading && chunk.type === 'RUN_FINISHED' && !runId + if ( + isRunlessSessionError || + isTrackedRunTerminal || + isPendingInterruptRunTerminal || + isCurrentRunTerminal || + isCurrentStreamTerminal + ) { + this.lastResume = null + this.pendingInterrupts = [] + this.pendingInterruptRunId = null + this.pendingInterruptResumeItems.clear() + } + this.notifyResumeStateChange() } /** - * The resume state for the active/interrupted run (the run id plus the last - * cursor seen), or null when there is nothing to resume. Apps can persist this - * to resume across a full reload; in-session reconnects use it automatically - * via {@link maybeAutoResume}. + * The interrupt-resume state for the active/interrupted run (its run/thread + * ids), or null when there is nothing to resume. Apps can persist this to + * resume interrupts across a full reload. */ getResumeState(): ChatResumeState | null { return this.lastResume ? { ...this.lastResume } : null @@ -741,21 +727,6 @@ export class ChatClient< return [...this.pendingInterrupts] } - /** - * Resume a run by replaying its persisted events after the last cursor, then - * continuing live — without re-sending messages. Uses the supplied state, or - * the tracked in-session state. No-op (returns false) when there is nothing to - * resume or a stream is already in flight. - */ - resume(state?: ChatResumeState): Promise { - const target = state ?? this.lastResume - if (!target || this.isLoading) return Promise.resolve(false) - this.pendingResumeThreadId = target.threadId - this.pendingResumeRunId = target.runId - this.pendingResumeCursor = target.cursor - return this.streamResponse() - } - resumeInterrupts( resume: Array, state?: ChatResumeState, @@ -764,28 +735,10 @@ export class ChatClient< if (!target || this.isLoading) return Promise.resolve(false) this.pendingResumeThreadId = target.threadId this.pendingResumeRunId = target.runId - this.pendingResumeCursor = target.cursor this.pendingResumeItems = [...resume] return this.streamResponse() } - /** - * Auto-resume hook for framework integrations to call on mount / when the tab - * comes back online. Honors the `autoResume` option (default true) and only - * fires when an interrupted run is tracked and no stream is in flight. - */ - maybeAutoResume(): Promise { - if ( - !this.autoResume || - this.isLoading || - !this.lastResume || - this.pendingInterrupts.length > 0 - ) { - return Promise.resolve(false) - } - return this.resume() - } - private generateUniqueId(prefix: string): string { return `${prefix}-${Date.now()}-${Math.random().toString(36).substring(7)}` } @@ -849,8 +802,9 @@ export class ChatClient< : this.serverPersistence.removeItem(this.threadId) if (result instanceof Promise) { result - .catch(() => { - // Persistence adapters are best-effort and must not break chat. + .catch((error: unknown) => { + // Best-effort: must not break chat, but is surfaced to a sink. + this.reportServerPersistenceError(error) }) .finally(() => { if (generation !== this.serverPersistenceGeneration) { @@ -858,8 +812,9 @@ export class ChatClient< } }) } - } catch { - // Persistence adapters are best-effort and must not break chat. + } catch (error) { + // Best-effort: must not break chat, but is surfaced to a sink. + this.reportServerPersistenceError(error) } } @@ -879,8 +834,9 @@ export class ChatClient< : this.serverPersistence.removeItem(this.threadId) if (result instanceof Promise) { result - .catch(() => { - // Persistence adapters are best-effort and must not break chat. + .catch((error: unknown) => { + // Best-effort: must not break chat, but is surfaced to a sink. + this.reportServerPersistenceError(error) }) .finally(() => { if (generation !== this.serverPersistenceGeneration) { @@ -888,8 +844,9 @@ export class ChatClient< } }) } - } catch { - // Persistence adapters are best-effort and must not break chat. + } catch (error) { + // Best-effort: must not break chat, but is surfaced to a sink. + this.reportServerPersistenceError(error) } } @@ -1066,7 +1023,7 @@ export class ChatClient< // per-run error only clears that run, while a runId-less RUN_ERROR is // treated as a session-level error that clears every active run. this.updateRunLifecycle(chunk) - this.observeResumeCursor(chunk) + this.observeInterruptState(chunk) // Yield control back to event loop for UI updates await new Promise((resolve) => setTimeout(resolve, 0)) } @@ -1235,17 +1192,16 @@ export class ChatClient< // Track generation so a superseded stream's cleanup doesn't clobber the new one const generation = ++this.streamGeneration - // Resuming reuses the original runId so the server replays that run's events. + // Resuming an interrupt reuses the original runId so the server continues + // that run with the resume decisions. const resumeThreadId = this.pendingResumeThreadId const resumeRunId = this.pendingResumeRunId - const resumeCursor = this.pendingResumeCursor const resumeItems = this.pendingResumeItems const isResumeRequest = Boolean( - resumeThreadId || resumeRunId || resumeCursor || resumeItems, + resumeThreadId || resumeRunId || resumeItems, ) this.pendingResumeThreadId = null this.pendingResumeRunId = null - this.pendingResumeCursor = null this.pendingResumeItems = null const runId = resumeRunId ?? @@ -1342,7 +1298,6 @@ export class ChatClient< : { type: 'object' }, })), forwardedProps: { ...mergedBody }, - ...(resumeCursor ? { cursor: resumeCursor } : {}), ...(resumeItems ? { resume: resumeItems } : {}), } this.devtoolsBridge.beginRun(runContext.runId, runContext.threadId) @@ -1600,7 +1555,6 @@ export class ChatClient< this.pendingInterruptResumeItems.clear() this.pendingResumeThreadId = null this.pendingResumeRunId = null - this.pendingResumeCursor = null this.pendingResumeItems = null this.notifyResumeStateChange() this.setError(undefined) @@ -1988,6 +1942,7 @@ export class ChatClient< } if (options.onError !== undefined) { this.callbacksRef.current.onError = options.onError + this.hasUserOnError = true } if (options.onSubscriptionChange !== undefined) { this.callbacksRef.current.onSubscriptionChange = diff --git a/packages/ai-client/src/connection-adapters.ts b/packages/ai-client/src/connection-adapters.ts index e427e7a1b..5b1787fab 100644 --- a/packages/ai-client/src/connection-adapters.ts +++ b/packages/ai-client/src/connection-adapters.ts @@ -48,6 +48,20 @@ export class StreamTruncatedError extends Error { } } +/** + * Thrown when a durable (id-tagged) run's stream ends with no terminal event + * and a reconnect makes no forward progress — the run cannot complete, so the + * consumer must not be left silently hanging on a stream that just stops. + */ +export class DurableStreamIncompleteError extends Error { + constructor() { + super( + 'Durable run ended without a terminal event and could not resume — the run did not complete.', + ) + this.name = 'DurableStreamIncompleteError' + } +} + function generateRunId(prefix: string): string { return `${prefix}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}` } @@ -142,10 +156,16 @@ async function* readStreamLines( * * A JSON parse failure throws — the consumer surfaces it as an error. */ -async function* responseToSSEChunks( +/** + * Yield StreamChunks parsed from an SSE Response body, each paired with the + * `id:` offset of the SSE event it arrived on (if any). Delivery durability + * tags every event with `id: `; carrying that id up lets the + * resumable adapter track the last offset (for reconnect) and de-dupe replays. + */ +async function* responseToSSEEvents( response: Response, abortSignal?: AbortSignal, -): AsyncGenerator { +): AsyncGenerator<{ chunk: StreamChunk; id?: string }> { if (!response.ok) { throw new Error( `HTTP error! status: ${response.status} ${response.statusText}`, @@ -155,11 +175,15 @@ async function* responseToSSEChunks( let lastThreadId: string | undefined let lastRunId: string | undefined let lastModel: string | undefined + let pendingId: string | undefined for await (const line of readStreamLines(reader, abortSignal)) { + if (line.startsWith('id:')) { + pendingId = line.slice(3).trim() + continue + } if ( line.startsWith(':') || line.startsWith('event:') || - line.startsWith('id:') || line.startsWith('retry:') ) { continue @@ -174,7 +198,7 @@ async function* responseToSSEChunks( timestamp: Date.now(), finishReason: 'stop', } - yield synthetic + yield { chunk: synthetic } return } const chunk = JSON.parse(data) as StreamChunk @@ -187,10 +211,110 @@ async function* responseToSSEChunks( if ('model' in chunk && typeof chunk.model === 'string') { lastModel = chunk.model } + const id = pendingId + pendingId = undefined + yield { chunk, ...(id !== undefined ? { id } : {}) } + } +} + +async function* responseToSSEChunks( + response: Response, + abortSignal?: AbortSignal, +): AsyncGenerator { + for await (const { chunk } of responseToSSEEvents(response, abortSignal)) { yield chunk } } +/** + * Iterate an SSE endpoint with native-style resumability. Each SSE event's + * `id:` (a `runId@seq` delivery-durability offset) is remembered; if the + * connection drops or ends before a terminal event, the request is re-issued + * with a `Last-Event-ID` header so the server replays strictly after the last + * offset. Already-seen offsets are de-duped, so an overlapping replay is safe. + * + * When the server does NOT tag events (no durability), no offset is ever seen, + * so no reconnect happens — behaviour is identical to a plain single fetch. + */ +async function* resumableServerSentEvents( + fetchClient: typeof globalThis.fetch, + url: string, + requestInit: RequestInit, + abortSignal?: AbortSignal, +): AsyncGenerator { + const seen = new Set() + let lastEventId: string | undefined + + for (;;) { + if (abortSignal?.aborted) return + const headers: Record = { + ...(requestInit.headers as Record | undefined), + } + if (lastEventId !== undefined) { + headers['Last-Event-ID'] = lastEventId + } + const response = await fetchClient(url, { + ...requestInit, + headers, + ...(abortSignal ? { signal: abortSignal } : {}), + }) + + let sawTerminal = false + let progressed = false + try { + for await (const { chunk, id } of responseToSSEEvents( + response, + abortSignal, + )) { + if (id !== undefined) { + if (seen.has(id)) continue + seen.add(id) + lastEventId = id + } + progressed = true + if (chunk.type === 'RUN_FINISHED' || chunk.type === 'RUN_ERROR') { + sawTerminal = true + } + yield chunk + if (sawTerminal) return + } + } catch (error) { + if (abortSignal?.aborted) return + // A truncated connection is resumable only if we have an offset and made + // forward progress; otherwise surface the failure. + if ( + error instanceof StreamTruncatedError && + lastEventId !== undefined && + progressed + ) { + continue + } + throw error + } + + if (abortSignal?.aborted) return + + if (lastEventId !== undefined) { + // A durable (id-tagged) run. + if (progressed) { + // Clean end WITHOUT a terminal event but we advanced — the producer is + // still going (or the socket rolled over). Reconnect from the last + // offset. + continue + } + // Ended without a terminal event AND made no forward progress on this + // pass: the run cannot complete. Surface an error rather than returning + // silently, which would leave the consumer with neither a terminal event + // nor a failure. + throw new DurableStreamIncompleteError() + } + + // A non-durable (untagged) stream that ended cleanly. Legitimate — the + // upper layer synthesizes a terminal event. Stop. + return + } +} + /** * Per-send context provided by the chat client to the connection adapter. * The adapter combines this with serialized messages to build a full @@ -200,12 +324,6 @@ export interface RunAgentInputContext { threadId: string runId: string parentRunId?: string - /** - * Resume cursor. When set, the request resumes `runId` — the server replays - * persisted events after this cursor (see `chat({ cursor })`). On a resume the - * client sends no new messages. - */ - cursor?: string /** AG-UI interrupt resume entries returned to the server on a follow-up run. */ resume?: Array /** Client-declared tools to advertise in the request payload. */ @@ -230,6 +348,22 @@ export interface ConnectConnectionAdapter { ) => AsyncIterable } +/** + * A {@link ConnectConnectionAdapter} that also supports joining an existing run + * (a second tab, or re-attaching after a full reload) via `joinRun`, replaying + * the ordered stream from the start off the server's delivery-durability sink. + */ +export interface ResumableConnectConnectionAdapter extends ConnectConnectionAdapter { + /** + * Join an in-flight or finished run by id, replaying from the start + * (`?offset=-1`). Read-only — sends no messages. + */ + joinRun: ( + runId: string, + abortSignal?: AbortSignal, + ) => AsyncIterable +} + export interface SubscribeConnectionAdapter { /** * Subscribe to stream chunks. @@ -452,7 +586,6 @@ function buildRunAgentInputBody( ...(runContext?.parentRunId !== undefined && { parentRunId: runContext.parentRunId, }), - ...(runContext?.cursor !== undefined && { cursor: runContext.cursor }), ...(runContext?.resume !== undefined && { resume: runContext.resume }), state: {}, messages: wireMessages, @@ -503,7 +636,7 @@ export function fetchServerSentEvents( options: | FetchConnectionOptions | (() => FetchConnectionOptions | Promise) = {}, -): ConnectConnectionAdapter { +): ResumableConnectConnectionAdapter { return { async *connect(messages, data, abortSignal, runContext) { // Resolve URL and options if they are functions @@ -535,15 +668,52 @@ export function fetchServerSentEvents( // under `exactOptionalPropertyTypes`), so spread it conditionally // rather than passing `undefined` explicitly. const signal = abortSignal || resolvedOptions.signal - const response = await fetchClient(resolvedUrl, { - method: 'POST', - headers: requestHeaders, - body: JSON.stringify(requestBody), - credentials: resolvedOptions.credentials || 'same-origin', - ...(signal ? { signal } : {}), - }) - yield* responseToSSEChunks(response, abortSignal) + // Resumable SSE: if the server tags events with `id:` offsets (delivery + // durability), a dropped/rolled-over connection auto-reconnects with a + // `Last-Event-ID` header and de-dupes the replayed prefix. With no tags, + // this is a single plain fetch. + yield* resumableServerSentEvents( + fetchClient, + resolvedUrl, + { + method: 'POST', + headers: requestHeaders, + body: JSON.stringify(requestBody), + credentials: resolvedOptions.credentials || 'same-origin', + }, + signal, + ) + }, + async *joinRun(runId, abortSignal) { + // Read an in-flight or finished run from the start. `?offset=-1` tells the + // server's delivery-durability sink to replay from the beginning; `runId` + // identifies which run. This is a read-only GET — no messages are sent. + const resolvedUrl = typeof url === 'function' ? url() : url + const resolvedOptions = + typeof options === 'function' ? await options() : options + + const separator = resolvedUrl.includes('?') ? '&' : '?' + const joinUrl = `${resolvedUrl}${separator}offset=-1&runId=${encodeURIComponent( + runId, + )}` + + const requestHeaders: Record = { + ...mergeHeaders(resolvedOptions.headers), + } + const fetchClient = resolvedOptions.fetchClient ?? fetch + const signal = abortSignal || resolvedOptions.signal + + yield* resumableServerSentEvents( + fetchClient, + joinUrl, + { + method: 'GET', + headers: requestHeaders, + credentials: resolvedOptions.credentials || 'same-origin', + }, + signal, + ) }, } } @@ -994,7 +1164,6 @@ export function fetcherToConnectionAdapter( data, threadId: runContext.threadId, runId: runContext.runId, - ...(runContext.cursor !== undefined && { cursor: runContext.cursor }), ...(runContext.resume !== undefined && { resume: runContext.resume }), }, { signal: abortSignal }, diff --git a/packages/ai-client/src/generation-client.ts b/packages/ai-client/src/generation-client.ts index acec24ea1..23c74675d 100644 --- a/packages/ai-client/src/generation-client.ts +++ b/packages/ai-client/src/generation-client.ts @@ -20,7 +20,6 @@ import type { GenerationClientState, GenerationFetcher, GenerationResumeSnapshot, - GenerationResumeState, GenerationServerPersistence, } from './generation-types' @@ -90,7 +89,6 @@ export class GenerationClient< private readonly devtoolsMetadata: AIDevtoolsClientMetadata private readonly devtoolsBridge: GenerationDevtoolsBridge private readonly threadId: string - private readonly autoResume: boolean private readonly serverPersistence: GenerationServerPersistence | undefined private body: Record private result: TOutput | null = null @@ -100,9 +98,7 @@ export class GenerationClient< private error: Error | undefined = undefined private status: GenerationClientState = 'idle' private resumeSnapshot: GenerationResumeSnapshot | undefined - private resumeState: GenerationResumeState | undefined private resumeSnapshotPersistenceQueue: Promise = Promise.resolve() - private resumeLifecycleToken = 0 private resumePersistenceError: Error | undefined = undefined private abortController: AbortController | null = null private readonly callbacksRef: GenerationCallbacks @@ -124,13 +120,8 @@ export class GenerationClient< this.connection = options.connection this.fetcher = options.fetcher this.body = options.body ?? {} - this.autoResume = options.autoResume ?? true this.serverPersistence = options.persistence?.server this.resumeSnapshot = options.initialResumeSnapshot - this.resumeState = - options.resumeState ?? - options.initialResumeSnapshot?.resumeState ?? - undefined this.callbacksRef = { onResult: options.onResult, @@ -201,10 +192,7 @@ export class GenerationClient< try { if (this.fetcher) { // Direct fetch path - const result = await this.fetcher(input, { - signal, - ...(this.resumeState ? { resumeState: this.resumeState } : {}), - }) + const result = await this.fetcher(input, { signal }) if (signal.aborted) return if (result instanceof Response) { // Server function returned SSE Response — parse stream @@ -266,41 +254,6 @@ export class GenerationClient< } } - async resume(state?: GenerationResumeState): Promise { - const resumeToken = this.resumeLifecycleToken - if (state) { - this.resumeState = state - this.resumeSnapshot = { - ...(this.resumeSnapshot ?? { status: 'running' }), - resumeState: state, - } - this.callbacksRef.onResumeSnapshotChange?.(this.resumeSnapshot) - } else { - await this.hydrateResumeSnapshot() - } - - if (this.disposed || resumeToken !== this.resumeLifecycleToken) { - return false - } - - if (!this.resumeState) { - return false - } - - // Resume requests are identified by resumeState/runContext. They do not - // need user media input and must not require callers to retain large input - // payloads across a reload. - await this.generate({} as TInput) - return true - } - - async maybeAutoResume(): Promise { - if (!this.autoResume || this.isLoading) { - return false - } - return this.resume() - } - /** * Process a stream of AG-UI events from the streaming connection adapter. */ @@ -368,7 +321,6 @@ export class GenerationClient< * Abort any in-flight generation request. */ stop(): void { - this.resumeLifecycleToken++ const runId = this.devtoolsBridge.getActiveRunId() if (this.abortController) { this.abortController.abort() @@ -404,12 +356,7 @@ export class GenerationClient< options: Partial< Pick< GenerationClientOptions, - | 'body' - | 'onResult' - | 'onError' - | 'onProgress' - | 'onChunk' - | 'resumeState' + 'body' | 'onResult' | 'onError' | 'onProgress' | 'onChunk' > >, ): void { @@ -428,9 +375,6 @@ export class GenerationClient< if (options.onChunk !== undefined) { this.callbacksRef.onChunk = options.onChunk } - if (options.resumeState !== undefined) { - this.resumeState = options.resumeState - } } dispose(): void { @@ -575,13 +519,6 @@ export class GenerationClient< } private createRunContext(runId: string): RunAgentInputContext { - if (this.resumeState) { - return { - threadId: this.resumeState.threadId, - runId: this.resumeState.runId, - cursor: this.resumeState.cursor, - } - } return { threadId: this.threadId, runId, @@ -593,18 +530,16 @@ export class GenerationClient< this.resumeSnapshot, chunk, ) - this.resumeState = this.resumeSnapshot.resumeState ?? undefined this.callbacksRef.onResumeSnapshotChange?.(this.resumeSnapshot) void this.persistResumeSnapshot(this.resumeSnapshot) } private completePlainFetcherResumeSnapshot(): void { - if (!this.resumeState && !this.resumeSnapshot) { + if (!this.resumeSnapshot) { return } - this.resumeState = undefined this.resumeSnapshot = { - ...(this.resumeSnapshot ?? {}), + ...this.resumeSnapshot, resumeState: null, status: 'complete', } @@ -612,19 +547,6 @@ export class GenerationClient< void this.persistResumeSnapshot(this.resumeSnapshot) } - private async hydrateResumeSnapshot(): Promise { - if (this.resumeSnapshot || !this.serverPersistence) { - return - } - const snapshot = await this.serverPersistence.getItem(this.threadId) - if (!snapshot) { - return - } - this.resumeSnapshot = snapshot - this.resumeState = snapshot.resumeState ?? undefined - this.callbacksRef.onResumeSnapshotChange?.(snapshot) - } - private async persistResumeSnapshot( snapshot: GenerationResumeSnapshot, ): Promise { diff --git a/packages/ai-client/src/generation-types.ts b/packages/ai-client/src/generation-types.ts index 6f792de02..53194e3a0 100644 --- a/packages/ai-client/src/generation-types.ts +++ b/packages/ai-client/src/generation-types.ts @@ -67,7 +67,6 @@ export type GenerationResumeStatus = 'idle' | 'running' | 'complete' | 'error' export interface GenerationResumeState { threadId: string runId: string - cursor: string } export type GenerationPendingArtifact = PersistedArtifactRef @@ -150,8 +149,6 @@ export const GENERATION_EVENTS = { export interface GenerationFetcherOptions { /** AbortSignal that is triggered when the user calls `stop()` */ signal: AbortSignal - /** Explicit persisted run/cursor metadata for direct server-function resume */ - resumeState?: GenerationResumeState } /** @@ -199,33 +196,20 @@ export interface GenerationClientOptions<_TInput, TResult, TOutput = TResult> { /** Metadata used to register this generation hook with TanStack AI Devtools */ devtools?: Partial - /** - * Whether framework integrations should attempt to resume a persisted run on - * mount. Low-level clients expose the flag for hooks; they do not auto-resume - * by themselves. Defaults to `true` in framework hooks. - */ - autoResume?: boolean - /** * Initial lightweight resume snapshot restored by framework hooks. Contains - * only cursor metadata, errors, and persisted artifact refs. + * only observed run metadata, errors, and persisted artifact refs. This is + * read-only state for display; it does not trigger any generation. */ initialResumeSnapshot?: GenerationResumeSnapshot /** - * Optional persistence adapters for lightweight generation resume state. + * Optional persistence adapters for lightweight generation state. * Generation hooks only support `server` persistence; generated media bytes * are never written into browser storage by this client. */ persistence?: GenerationPersistenceOptions - /** - * Explicit run/cursor state to send on the next generation request. This is - * only a resume input; explicit `stop()` still aborts the local connection and - * does not model a durable server-side cancel. - */ - resumeState?: GenerationResumeState - /** * Factory that constructs the devtools bridge. Default is a no-op * factory; the real implementation lives in `@tanstack/ai-client/devtools`. @@ -267,7 +251,6 @@ export function updateGenerationResumeSnapshot( ): GenerationResumeSnapshot { const threadId = stringField(chunk, 'threadId') const runId = stringField(chunk, 'runId') - const cursor = stringField(chunk, 'cursor') const previousArtifacts = previous?.pendingArtifacts ?? [] const next: GenerationResumeSnapshot = { resumeState: previous?.resumeState ?? null, @@ -281,8 +264,8 @@ export function updateGenerationResumeSnapshot( lastEvent: createGenerationEventSnapshot(chunk), } - if (threadId && runId && cursor) { - next.resumeState = { threadId, runId, cursor } + if (threadId && runId) { + next.resumeState = { threadId, runId } next.status = 'running' } else if (chunk.type === 'RUN_STARTED') { next.status = 'running' diff --git a/packages/ai-client/src/index.ts b/packages/ai-client/src/index.ts index 3dfb4f428..53fefb69e 100644 --- a/packages/ai-client/src/index.ts +++ b/packages/ai-client/src/index.ts @@ -116,9 +116,11 @@ export { stream, rpcStream, StreamTruncatedError, + DurableStreamIncompleteError, type ConnectConnectionAdapter, type ConnectionAdapter, type FetchConnectionOptions, + type ResumableConnectConnectionAdapter, type RunAgentInputContext, type SubscribeConnectionAdapter, type XhrConnectionOptions, diff --git a/packages/ai-client/src/types.ts b/packages/ai-client/src/types.ts index c73ce1ed1..52ee48fbb 100644 --- a/packages/ai-client/src/types.ts +++ b/packages/ai-client/src/types.ts @@ -24,7 +24,6 @@ export type { StructuredOutputPart } export interface ChatResumeState { threadId: string runId: string - cursor: string } export type ChatPendingInterrupt = Interrupt @@ -46,7 +45,6 @@ export interface ChatFetcherInput { data?: Record threadId: string runId: string - cursor?: string resume?: Array } @@ -429,19 +427,10 @@ export interface ChatClientBaseOptions< */ threadId?: string - /** - * Whether to auto-resume an interrupted run when {@link maybeAutoResume} is - * called (e.g. by a framework integration on mount / when the tab comes back - * online). Requires server-side persistence so the run's events can be - * replayed by `runId + cursor`. Defaults to `true`; set `false` to opt out. - */ - autoResume?: boolean - /** * Initial resumable run state, useful when rehydrating a persisted client - * after a full page reload. The server still owns replay/resume decisions; - * this only restores the client-side interrupt descriptors needed to send - * AG-UI resume entries. + * after a full page reload. This restores the client-side interrupt + * descriptors needed to send AG-UI resume entries. */ initialResumeSnapshot?: ChatResumeSnapshot diff --git a/packages/ai-client/src/video-generation-client.ts b/packages/ai-client/src/video-generation-client.ts index 04c6eae6f..30624519f 100644 --- a/packages/ai-client/src/video-generation-client.ts +++ b/packages/ai-client/src/video-generation-client.ts @@ -19,7 +19,6 @@ import type { GenerationClientState, GenerationFetcher, GenerationResumeSnapshot, - GenerationResumeState, GenerationServerPersistence, VideoGenerateInput, VideoGenerateResult, @@ -97,7 +96,6 @@ export class VideoGenerationClient { private readonly devtoolsMetadata: AIDevtoolsClientMetadata private readonly devtoolsBridge: VideoDevtoolsBridge private readonly threadId: string - private readonly autoResume: boolean private readonly serverPersistence: GenerationServerPersistence | undefined private body: Record @@ -110,9 +108,7 @@ export class VideoGenerationClient { private error: Error | undefined = undefined private status: GenerationClientState = 'idle' private resumeSnapshot: GenerationResumeSnapshot | undefined - private resumeState: GenerationResumeState | undefined private resumeSnapshotPersistenceQueue: Promise = Promise.resolve() - private resumeLifecycleToken = 0 private resumePersistenceError: Error | undefined = undefined private abortController: AbortController | null = null private readonly callbacksRef: VideoCallbacks @@ -134,13 +130,8 @@ export class VideoGenerationClient { this.connection = options.connection this.fetcher = options.fetcher this.body = options.body ?? {} - this.autoResume = options.autoResume ?? true this.serverPersistence = options.persistence?.server this.resumeSnapshot = options.initialResumeSnapshot - this.resumeState = - options.resumeState ?? - options.initialResumeSnapshot?.resumeState ?? - undefined this.callbacksRef = { onResult: options.onResult, @@ -259,40 +250,6 @@ export class VideoGenerationClient { } } - async resume(state?: GenerationResumeState): Promise { - const resumeToken = this.resumeLifecycleToken - if (state) { - this.resumeState = state - this.resumeSnapshot = { - ...(this.resumeSnapshot ?? { status: 'running' }), - resumeState: state, - } - this.callbacksRef.onResumeSnapshotChange?.(this.resumeSnapshot) - } else { - await this.hydrateResumeSnapshot() - } - - if (this.disposed || resumeToken !== this.resumeLifecycleToken) { - return false - } - - if (!this.resumeState) { - return false - } - - // Resume is driven by run/cursor metadata. Do not require callers to keep - // large prompt/media input payloads in browser state across a refresh. - await this.generate({} as VideoGenerateInput) - return true - } - - async maybeAutoResume(): Promise { - if (!this.autoResume || this.isLoading) { - return false - } - return this.resume() - } - /** * Direct fetcher mode: call fetcher and set result. */ @@ -304,10 +261,7 @@ export class VideoGenerationClient { if (!this.fetcher) return // Fetcher returns a completed result directly, or a Response with SSE body - const result = await this.fetcher(input, { - signal, - ...(this.resumeState ? { resumeState: this.resumeState } : {}), - }) + const result = await this.fetcher(input, { signal }) if (signal.aborted) return if (result instanceof Response) { @@ -400,7 +354,6 @@ export class VideoGenerationClient { * Abort any in-flight generation or polling. */ stop(): void { - this.resumeLifecycleToken++ const runId = this.devtoolsBridge.getActiveRunId() if (this.abortController) { this.abortController.abort() @@ -445,7 +398,6 @@ export class VideoGenerationClient { | 'onChunk' | 'onJobCreated' | 'onStatusUpdate' - | 'resumeState' > >, ): void { @@ -470,9 +422,6 @@ export class VideoGenerationClient { if (options.onStatusUpdate !== undefined) { this.callbacksRef.onStatusUpdate = options.onStatusUpdate } - if (options.resumeState !== undefined) { - this.resumeState = options.resumeState - } } dispose(): void { @@ -656,13 +605,6 @@ export class VideoGenerationClient { } private createRunContext(runId: string): RunAgentInputContext { - if (this.resumeState) { - return { - threadId: this.resumeState.threadId, - runId: this.resumeState.runId, - cursor: this.resumeState.cursor, - } - } return { threadId: this.threadId, runId, @@ -674,18 +616,16 @@ export class VideoGenerationClient { this.resumeSnapshot, chunk, ) - this.resumeState = this.resumeSnapshot.resumeState ?? undefined this.callbacksRef.onResumeSnapshotChange?.(this.resumeSnapshot) void this.persistResumeSnapshot(this.resumeSnapshot) } private completePlainFetcherResumeSnapshot(): void { - if (!this.resumeState && !this.resumeSnapshot) { + if (!this.resumeSnapshot) { return } - this.resumeState = undefined this.resumeSnapshot = { - ...(this.resumeSnapshot ?? {}), + ...this.resumeSnapshot, resumeState: null, status: 'complete', } @@ -693,19 +633,6 @@ export class VideoGenerationClient { void this.persistResumeSnapshot(this.resumeSnapshot) } - private async hydrateResumeSnapshot(): Promise { - if (this.resumeSnapshot || !this.serverPersistence) { - return - } - const snapshot = await this.serverPersistence.getItem(this.threadId) - if (!snapshot) { - return - } - this.resumeSnapshot = snapshot - this.resumeState = snapshot.resumeState ?? undefined - this.callbacksRef.onResumeSnapshotChange?.(snapshot) - } - private async persistResumeSnapshot( snapshot: GenerationResumeSnapshot, ): Promise { diff --git a/packages/ai-client/tests/chat-client-resume.test.ts b/packages/ai-client/tests/chat-client-resume.test.ts index 4d51ebcf4..f2e7e4839 100644 --- a/packages/ai-client/tests/chat-client-resume.test.ts +++ b/packages/ai-client/tests/chat-client-resume.test.ts @@ -1,10 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import { EventType } from '@tanstack/ai/client' import { ChatClient } from '../src/chat-client' -import { - createApprovalToolCallChunks, - createToolCallChunks, -} from './test-utils' import type { ConnectConnectionAdapter, RunAgentInputContext, @@ -40,12 +36,11 @@ function recordingAdapter(scripts: Array