TanStack AI Persistence, Resumable Streams & Memory with Redis

TanStack AI defines the contracts for an agent’s production state — chat persistence, session history, stored credentials, resumable streaming, locks, memory — and ships in-memory implementations that only work inside one process. @upstash/agentkit-tanstack-ai implements them on Upstash Redis, so they hold across serverless instances, page reloads, and devices.

npm install @upstash/agentkit-tanstack-ai @tanstack/ai

How to persist TanStack AI chats in Redis#

Persistence plugs into TanStack AI’s withPersistence() middleware, which comes from its persistence package:

npm install @tanstack/ai-persistence

import { chat } from "@tanstack/ai";import { withPersistence } from "@tanstack/ai-persistence";import { upstashPersistence } from "@upstash/agentkit-tanstack-ai/persistence";const persistence = upstashPersistence();chat({ adapter, messages, threadId, middleware: [withPersistence(persistence)] });

This covers every TanStack AI persistence store: messages, runs, interrupts, and metadata for chats, generationRuns and artifacts for one-shot generation jobs such as images or speech, and the sessions and log stores described below. The blobs store for generated bytes is added when you pass an Upstash Blob bucket:

import { Bucket } from "@upstash/blob";const persistence = upstashPersistence({ bucket: Bucket.fromEnv() });

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment. Bucket.fromEnv() reads UPSTASH_BLOB_TOKEN from the environment, which is only needed when you store generated files.

Runs are indexed by thread, so reconnecting to a live run (findActiveRun) is an indexed lookup, newest run first. Each write is one command or one Lua script, so concurrent instances cannot interleave it.

upstashPersistence({  redis, // optional: defaults to Redis.fromEnv()  prefix: "agentkit:tanstack", // optional: base key prefix  messagesTtlSeconds: 60 * 60 * 24 * 30, // optional: expire idle transcripts (default: never)  bucket: Bucket.fromEnv(), // optional: Upstash Blob bucket for generated files  logPollIntervalMs: 250, // optional: how often the session log checks for writes from other machines});

How to keep a session index in Redis#

upstashPersistence() also returns a sessions store, TanStack AI’s SessionIndexStore. It holds one entry per (tenantId, threadId) that a “your chats” list reads: members, title, harness, token usage, pinned and archived flags, and your own metadata. It holds no messages, which stay in the other stores.

const { sessions } = upstashPersistence().stores;await sessions.upsert({  tenantId,  threadId,  members: [{ userId, role: "owner" }],  title: "Refund question",  createdAt: Date.now(),  updatedAt: Date.now(),});// Newest first, one page at a time.const { entries, cursor, truncated } = await sessions.list({ tenantId, memberId: userId, limit: 20 });

• list is newest-first per tenant, and filters by memberId, search (title, case-insensitive), harness, parentThreadId, metadata, and withArchived. Pass the returned cursor back for the next page.
• upsert replaces the whole entry. To change one field safely from several machines, read the entry with getVersioned and write the changed copy with upsertIf, which only writes at the revision you read.
• delete removes only the entry, not the thread’s data.

To use the index on its own, import upstashSessionIndex(redis, prefix) from @upstash/agentkit-tanstack-ai/persistence.

How to keep the session log in Redis#

upstashPersistence() also returns a log store, TanStack AI’s LogStore: the append-only log a host keeps for a session tree (events, transcript, inputs, and tool steps), with what several machines need to share it safely.

const { log } = upstashPersistence().stores;// One owner at a time holds the log; every fenced write carries its epoch.const lease = await log.acquireOwnership({ logId, harness: "support", ownerId: hostId, ttlMs: 30_000 });if (lease) {  await log.append({ logId, seq: 1, epoch: lease.epoch, records: [{ type: "message", text: "hi" }] });  const entries = await log.read({ logId, after: 0 });}

• Append: append writes a batch at the next position, together with the pending inputs it consumes, in one atomic step. A second writer at the same position gets a LogConflictError and nothing is written. One call takes at most 10,000 records.
• Ownership: acquireOwnership, renewOwnership, and releaseOwnership give one host a lease with an epoch. A host that lost the log gets an OwnershipLostError on its next fenced write. Lease expiry uses the Redis server clock, so your machines need no clock sync.
• Input lane: offerInput lets any machine queue an input, and listLogsWithoutOwner finds logs whose owner expired, that have pending inputs, or whose wake time is due. markHostAlive and pinToWorkers keep a log on one group of workers.
• Snapshots: putSnapshot and truncate keep a log short. Reading before the truncation point throws LogTruncatedError.
• Subscriptions: subscribe calls your listener right after writes through the same store, and polls for writes from other machines every logPollIntervalMs (250 ms by default), because the REST API keeps no blocking reads. Treat a call as “something changed” and read to catch up.

The store passes all of TanStack AI’s log, ownership, input-lane, snapshot, and cross-instance conformance cases. upstashLogStore(redis, prefix, { pollIntervalMs }) is also exported from @upstash/agentkit-tanstack-ai/persistence for standalone use.

How to store provider API keys and OAuth tokens#

upstashCredentials() is a CredentialStore (from @tanstack/ai/byok) for API keys and OAuth tokens that users or organizations saved. TanStack AI’s storedProviderKeys() reads from it.

import { storedProviderKeys } from "@tanstack/ai/byok/server";import { upstashCredentials } from "@upstash/agentkit-tanstack-ai/byok";const credentials = upstashCredentials({ encryptionKey: process.env.CREDENTIALS_KEY });await credentials.set({  scope: { tenantId, userId },  id: "openai",  credential: { type: "api_key", value: apiKey },});// The user's own key, else the tenant's.const keys = storedProviderKeys(credentials, { tenantId, userId });

Each owner (tenantId plus userId) has its own hash, so list returns ids and types only and never a secret. getVersioned and setIf are compare-and-set on a revision that never comes back after a delete, so several instances refreshing one OAuth token cannot overwrite each other.

upstashCredentials({  redis, // optional: defaults to Redis.fromEnv()  prefix: "agentkit:tanstack", // optional: base key prefix  encryptionKey: process.env.CREDENTIALS_KEY, // optional but strongly recommended  requireEncryption: true, // optional: refuse plaintext rows once a key is set});

How to resume a TanStack AI stream after a reload#

import { chat, toServerSentEventsResponse } from "@tanstack/ai";import { upstashStream } from "@upstash/agentkit-tanstack-ai";export async function POST(request: Request) {  const stream = chat({ adapter, messages, threadId });  return toServerSentEventsResponse(stream, { durability: { adapter: upstashStream(request) } });}

Every chunk is written to a Redis Stream before it is sent. A client that reconnects with Last-Event-ID (or ?offset) replays what it missed and keeps following the live run, whichever instance serves the request. Without a Request, use upstashStream({ runId, offset }).

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

upstashStream(request, {  ttlSeconds: 86_400, // optional: how long a run stays resumable  pollIntervalMs: 150, // optional: how often a caught-up reader checks for new chunks  firstChunkDeadlineMs: 2_000, // optional: how long a join waits for a run that has not started});

How to use distributed locks with TanStack AI#

import { withLocks } from "@tanstack/ai/locks";import { upstashLocks } from "@upstash/agentkit-tanstack-ai";chat({ adapter, messages, middleware: [withLocks(upstashLocks()), withSandbox(sandbox)] });

withLocks doesn’t lock anything by itself. It gives the lock store to later middleware, which lock the one step they must not run twice: withSandbox uses it so two concurrent requests for a thread don’t both create a sandbox, and your own middleware can use it through getLocks(ctx). It does not serialize whole chat turns. Unlike TanStack’s InMemoryLockStore, which only works inside one process, upstashLocks() coordinates across instances.

Each lock is a lease that is renewed while the critical section runs. If the lease is lost, the section’s signal aborts.

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

upstashLocks({  leaseMs: 30_000, // optional: lease lifetime without renewal  acquireTimeoutMs: 30_000, // optional: how long to wait for a held key  retryDelayMs: 100, // optional: delay between attempts});

How to add long-term memory to TanStack AI#

Memory plugs into TanStack AI’s memoryMiddleware(), which comes from its memory package:

npm install @tanstack/ai-memory

import { memoryMiddleware } from "@tanstack/ai-memory";import { upstashMemory } from "@upstash/agentkit-tanstack-ai/memory";chat({  adapter,  messages,  middleware: [    memoryMiddleware({      adapter: upstashMemory(),      // derive these server-side from the session, never from the request body      scope: (ctx) => ({ threadId: ctx.threadId, userId: session.userId }),    }),  ],});

Before each turn, the most relevant memories for the user’s message are added to the system prompt, labelled by where they came from. The model gets a save_memory tool for durable facts, and each turn’s user message is captured too. Memory is per user across threads by default, or per thread when the scope has no userId.

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

upstashMemory({  topK: 5, // optional: memories injected per turn  scopeBy: "user", // optional: "user" (across threads) or "thread"  captureUserMessages: true, // optional: store each turn's user message  saveTool: true, // optional: offer the save_memory tool  waitForIndexing: true, // optional: make a save recallable on the very next turn});

How to cache tools and rate limit with TanStack AI#

import { rateLimit, Ratelimit, toolCache } from "@upstash/agentkit-tanstack-ai";chat({  adapter,  messages,  tools: [getWeather, sendEmail],  middleware: [    rateLimit({ limiter: Ratelimit.slidingWindow(10, "60 s"), identifier: userId }),    toolCache({ tools: ["get_weather"], userId, ttlSeconds: 600 }),  ],});

toolCache only caches the tools you list — list deterministic, side-effect-free tools only. rateLimit fails the run with RateLimitExceededError before the model is called. For an HTTP 429 instead, call createRateLimit({ limiter }).limit(userId) in your route before chat().

Both middlewares and createRateLimit read UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

How to add RAG with TanStack AI#

The schema builder s comes from @upstash/redis:

npm install @upstash/redis

import { s } from "@upstash/redis";import { createSearchTools } from "@upstash/agentkit-tanstack-ai";const tools = createSearchTools({  indexName: "products",  schema: s.object({ name: s.string(), price: s.number(), category: s.string().noTokenize() }),});chat({ adapter, messages, tools });

The tool descriptions are generated from the schema, and the index is created on first use.

Reads UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN from the environment.

Telemetry#

AgentKit adds its package name and version as a header on your Redis client’s requests. To turn it off, set UPSTASH_DISABLE_TELEMETRY, or pass enableTelemetry: false to a helper.