lukasa1993/tanstack-skills-extracted/tanstack-ai-persistence-build-custom-adapter

Custom Chat Persistence

The deliverable is one file in the app — src/lib/chat-persistence.ts — exporting a ChatPersistence built from the database client the app already has. Plus whatever DDL that database needs, added through the app's existing migration flow.

Do not create a package, a second client, or a migration runner.

Always write all four tables (or collections) and all four stores: thread, run, interrupt, metadata. The docs call interrupts and metadata optional for a hand-rolled adapter; this recipe is not that case. withPersistence needs all four to type-check as ChatPersistence, and the conformance suite fails without them. Do not stop at a thread table and a run table.

Route first. If the app already runs one of these, stop and use that skill — it has the driver-specific code:

Everything else lands here. The full contracts and their invariants are in ai-persistence/stores and docs/persistence/store-reference.md; the complete worked node:sqlite walkthrough is docs/persistence/build-your-own-chat-adapter.md and examples/ts-react-chat/src/lib/sqlite-persistence.ts.

1. Read the app before writing anything

2. Shape the storage

Four logical records. Whatever the engine, keep these keys — the store methods look records up by exactly these:

• Timestamps are epoch milliseconds (number) in records. Store them

however the engine prefers and convert in the mapper.

• (namespace, key) is a composite key. Never join with a separator —

('a:b','c') and ('a','b:c') must stay distinct records, and the conformance suite checks it.

• Index runs(threadId, status), runs(threadId, startedAt), and

interrupts(threadId, requestedAt) for the listing paths. If the backend implements listReclaimable, also index runs(status, detachedSince). If it implements listByParentRun, also index runs(parentRunId, startedAt).

• run.error is a structured RunError ({ message: string, code?: string }),

not a bare string. message is the provider's prose; code is the stable, machine-branchable classification an operator filters and groups by. In a SQL-backed table, store it as two columns (error, error_code) rather than one JSON blob, moved together in update so a later code-less failure can never leave a stale code from an earlier one behind. run.status is one of 'running' | 'interrupted' | 'completed' | 'failed' | 'aborted'; 'interrupted' is a pause, not terminal, and only 'completed' | 'failed' | 'aborted' are terminal.

• Extra app-owned columns are fine (a userId, audit columns) as long as they

are nullable or defaulted. The stores never read columns they do not know about.

3. The invariants

Getting one of these wrong is the usual source of stuck approvals and wiped history. They are engine-independent:

1. `saveThread` is a full overwrite, never an append. The argument is the

complete authoritative transcript.

1. `loadThread` returns `[]` for an unknown thread, never null.
2. `createOrResume` is insert-if-absent — an existing runId comes back

_unchanged_, ignoring the new field values. Resume and double-submit depend on it. After a racy insert, re-read rather than trusting your own write.

1. `runs.update` on an unknown id is a silent no-op — it must not throw and

must not insert. (Drivers that throw on zero rows affected need the updateMany-style call, not the update-one-or-throw call.)

1. **runs.update distinguishes "field omitted" from "field explicitly

cleared" for the durable-run fields** (sandboxKey, detachedSince, cancelRequested, driverEpoch). A reattach clears detachedSince by passing it explicitly as undefined — update(runId, { detachedSince: undefined }) — and that must write NULL, not be silently dropped. Check 'detachedSince' in patch, never patch.detachedSince !== undefined; the latter cannot tell a clear from an omission and leaves every reattached run looking permanently detached to the reaper. Same rule for cancelRequested (false is a real value, not "unset") and for sandboxKey / driverEpoch. See examples/ts-react-chat/src/lib/sqlite-persistence.ts for the pattern.

1. `interrupts.create` is insert-if-absent — never clobber a resolved

interrupt back to pending. Every list* is ordered by requestedAt ascending.

1. `runs.listReclaimable` uses an inclusive cutoff (if implemented):

status === 'running' AND detachedSince <= now - ttlMs. It is a query, not automatic reclamation: reapDetachedRuns from @tanstack/ai-sandbox is the sweep that consumes it, and the application schedules that sweep. A store without this method cannot be reaped. runs.findActiveRun is required; runs.listByThread, runs.listByParentRun, and runs.listReclaimable are optional: implement only what the app needs and leave the rest off the object. listByParentRun returns the child runs for one parentRunId, oldest startedAt first. reconstructChat uses that list to put subagent cards back. createOrResume copies parentRunId, subagentRunId, and name on the first insert and leaves them unchanged on resume.

Row mappers omit absent optionals (...(row.sandbox_key != null ? { sandboxKey: row.sandbox_key } : {})) so records compare cleanly against the reference in-memory backend. For a two-column error/error_code layout, the mapper is ...(row.error != null ? { error: { message: row.error, ...(row.error_code != null ? { code: row.error_code } : {}) } } : {}).

4. Write src/lib/chat-persistence.ts

Four factories and one assembly. Postgres via pg shown here; the shape is the same for any driver.

<!-- fence: custom-chat-persistence -->

import { defineAIPersistence } from '@tanstack/ai-persistence'
import type { QueryResultRow } from 'pg'
import type { Pool } from 'pg'
import type { ModelMessage, TokenUsage } from '@tanstack/ai'
import type {
  ChatPersistence,
  InterruptRecord,
  InterruptStatus,
  InterruptStore,
  MessageStore,
  MetadataStore,
  RunRecord,
  RunStatus,
  RunStore,
} from '@tanstack/ai-persistence'
import { pool } from '@/db'
type SqlParam = string | number | boolean | null
const RUN_STATUSES: ReadonlyArray<RunStatus> = [
  'running',
  'interrupted',
  'completed',
  'failed',
  'aborted',
]
const INTERRUPT_STATUSES: ReadonlyArray<InterruptStatus> = [
  'pending',
  'resolved',
  'cancelled',
]
function toRunStatus(value: string): RunStatus {
  const status = RUN_STATUSES.find((candidate) => candidate === value)
  if (!status) throw new Error(`Unknown run status: ${value}`)
  return status
}
function toInterruptStatus(value: string): InterruptStatus {
  const status = INTERRUPT_STATUSES.find((candidate) => candidate === value)
  if (!status) throw new Error(`Unknown interrupt status: ${value}`)
  return status
}
function epochMs(value: string | number): number {
  return typeof value === 'number' ? value : Number(value)
}
// `pg` parses jsonb into a value, so a JSON string is already a string.
// A text column still holds JSON source (`{`, `[`, or a quoted string).
// Re-encode a decoded string before parsing: JSON.parse("left") throws.
function readJson<T>(value: T | string): T {
  if (typeof value !== 'string') return value
  const first = value.trimStart()[0]
  const source =
    first === '{' || first === '[' || first === '"'
      ? value
      : JSON.stringify(value)
  return JSON.parse(source)
}
interface ThreadRow extends QueryResultRow {
  messages_json: Array<ModelMessage> | string
}
interface RunRow extends QueryResultRow {
  run_id: string
  thread_id: string
  status: string
  started_at: string | number
  finished_at: string | number | null
  error: string | null
  error_code: string | null
  usage_json: TokenUsage | string | null
  sandbox_key: string | null
  detached_since: string | number | null
  cancel_requested: boolean | null
  driver_epoch: number | null
  parent_run_id: string | null
  subagent_run_id: string | null
  name: string | null
}
interface InterruptRow extends QueryResultRow {
  interrupt_id: string
  run_id: string
  thread_id: string
  status: string
  requested_at: string | number
  resolved_at: string | number | null
  payload_json: Record<string, unknown> | string
  response_json: unknown
}
interface MetadataRow extends QueryResultRow {
  value_json: unknown
}
async function rowsOf<T extends QueryResultRow>(
  db: Pool,
  sql: string,
  params: Array<SqlParam> = [],
): Promise<Array<T>> {
  const result = await db.query<T>(sql, params)
  return result.rows
}