ai-persistence/server

Server Chat Persistence

Builds on ai-persistence. Package: @tanstack/ai-persistence.

withPersistence(persistence) is a ChatMiddleware that writes chat state to a backend: messages, runs, interrupts (optional metadata). It does not mutate the chunk stream and does not replace delivery durability.

Setup

import {
  chat,
  chatParamsFromRequest,
  toServerSentEventsResponse,
} from '@tanstack/ai'
import { openaiText } from '@tanstack/ai-openai'
import { withPersistence } from '@tanstack/ai-persistence'
// Your adapter — see ai-persistence/stores.
import { persistence } from './persistence'

export async function POST(request: Request) {
  const params = await chatParamsFromRequest(request)
  const stream = chat({
    adapter: openaiText('gpt-5.5'),
    messages: params.messages,
    threadId: params.threadId,
    runId: params.runId,
    ...(params.resume ? { resume: params.resume } : {}),
    middleware: [withPersistence(persistence)],
  })
  return toServerSentEventsResponse(stream)
}

Always pass threadId and runId from the client (via chatParamsFromRequest / body helpers). Forward resume when the client resolves pending interrupts.

For dev and tests, memoryPersistence() from @tanstack/ai-persistence is a drop-in backend that implements all four stores in process.

What each store does

Store	Role	Required?
messages	Full model-message transcript load/save	Yes for withPersistence
runs	Run status, timing, usage, errors	Optional; needed for interrupt durability
interrupts	Pending/resolved tool approvals & waits	Optional; requires runs
metadata	App-owned namespaced key/value	Optional

Named shapes: ChatTranscriptPersistence (floor), ChatPersistence (all four). Annotate your factory with one of these, not with bare AIPersistence — the unparameterized type is the all-optional bag, and withPersistence rejects it because stores.messages is possibly undefined.

Merge incoming messages by id

withPersistence merges incoming messages into the stored thread by id.

• Empty messages: load the stored thread and continue.
• Non-empty messages: merge by id. The last incoming id that already exists in stored is a cutoff. Stored messages after it are dropped. If no incoming id is in stored, every stored message stays. Same id: incoming wins. New ids and messages with no id are appended.
• saveThread replaces the thread with that merged list. Merge is middleware, not the store.

When state is written

Moment	Writes	Best-effort?
onStart	Pending turn snapshot (user + history)	Yes — failure does not abort
Interrupt boundary	New interrupts, run → interrupted, message snapshot	No
onFinish	Canonical transcript first, then run → completed, commit resumes	No
Stream (optional)	Throttled partial assistant text	Yes if snapshotStreaming: true
onError	Run → failed	Resumes stay pending
onAbort	Run → aborted — but only sometimes (see below)	Resumes stay pending

The canonical transcript already contains the completed terminal assistant messages. Native-combined output keeps the structured result on its terminal assistant message. Separate finalization and event-sourced harness output can preserve plain-text and structured-output assistant messages separately when those messages use different ids.

import { withPersistence } from '@tanstack/ai-persistence'
import { persistence } from './persistence'

withPersistence(persistence, {
  snapshotStreaming: true,
  snapshotIntervalMs: 1000, // default
})

onAbort writes conditionally, not always

A user pressing Stop and a user closing the tab produce the identical connection close, so onAbort can never infer intent from the abort alone. It writes:

• 'aborted' (terminal, with finishedAt) when the abort is an explicit cancel — info.cancelRequested === true, or a durable cancel request found via wasCancelRequested(runs, runId) (both from @tanstack/ai; paired with requestRunCancel/RUN_CANCEL_REASON) — or when the run is not detachable at all (no sandbox/journal behind it, so there is nothing to reattach to).
• Nothing when it is a plain disconnect on a detachable run (some other middleware, e.g. @tanstack/ai-sandbox, has provided DetachableRunCapability from @tanstack/ai). The record deliberately stays 'running' — the agent keeps running and a later attach can take it over. (The detaching middleware, not withPersistence, is what stamps detachedSince.)

Chat's onAbort and generation's onAbort (withGenerationPersistence) are asymmetric on purpose: a generation job has no journal and no agent loop to reattach to, so its onAbort always writes 'aborted' unconditionally. Do not "fix" that asymmetry by making generation conditional, or chat unconditional — both are correct for what they wrap.

Never build a client, or a persistence backend, that assumes a disconnect always finalizes the run — for a detachable run it usually does not, and inventing a finishedAt for a still-'running' record breaks takeover. Use isTerminalRunStatus(status) (from @tanstack/ai-persistence) to test whether a status is finished, rather than re-listing 'completed' | 'failed' | 'aborted' by hand.

Streaming snapshots default off (finish is authoritative). Enable only when partial-output durability is worth extra writes.

Resumes accepted in onConfig commit only at a success boundary (interrupt or finish). A failed run leaves interrupts pending so the same resume batch can retry.

Interrupt / resume flow

1. Middleware records pending interrupts and gates new input: if pending exist, the request must include a matching resume batch or onConfig throws.
2. On valid resume, middleware builds resumeToolState and clears config.resume so the engine does not double-reconstruct from client history (server owns transcript).
3. On success boundary, interrupts are marked resolved/cancelled.

Hydrate a thread for the client (reconstructChat)

Server-authoritative clients load history by threadId (often GET):

import { reconstructChat } from '@tanstack/ai-persistence'
import { persistence } from './persistence'
import { sessionUserId, userOwnsThread } from './auth'

export async function GET(request: Request) {
  return reconstructChat(persistence, request, {
    // Multi-user: required in production
    authorize: async (threadId, req) => {
      const userId = await sessionUserId(req)
      return userOwnsThread(userId, threadId)
    },
  })
}

Returns { messages, activeRun, interrupts, page? }:

• messages: UI messages for this window
• activeRun: { runId } if a run is still generating (runs.findActiveRun)
• interrupts: pending human-in-the-loop state for re-prompt
• page: { truncated, cursor } when the GET has a valid limit

Paging is opt-in. No limit returns the full transcript and can omit page. reconstructChat reads limit and before from the query. activeRun and interrupts are not paged.

Without authorize, anyone who guesses ?threadId= gets the transcript.

Generation activities

withGenerationPersistence(persistence) tracks run records for non-chat activities (image, audio, TTS, video, transcription). Do not fake threadId = requestId on chat run stores — use the generation helper.

Common mistakes

CRITICAL: Merge inside saveThread

Merge by id is withPersistence. saveThread must replace the merged list it receives.

HIGH: Omitting threadId / runId

Persistence keys and resume need stable ids. Use chatParamsFromRequest.

HIGH: Interrupts without runs

interrupts requires runs; withPersistence throws otherwise.

HIGH: Typing a factory as bare AIPersistence

AIPersistence defaults to the sparse all-optional bag, so withPersistence and reconstructChat reject the value. Return ChatPersistence (or ChatTranscriptPersistence) instead.

MEDIUM: Expecting withPersistence to reconnect a dropped stream

That is delivery durability (resumable streams), not state persistence.

Cross-references

• ai-persistence — layers and recommended stack
• ai-persistence/stores — implement the store interfaces
• ai-core/client-persistence (@tanstack/ai) — browser half
• ai-core/locks — multi-instance coordination