TanStack AI

S2 integrates with TanStack AI through the @s2-dev/resumable-stream package. The integration persists TanStack AI StreamChunk events to S2 and replays them through a useChat connection adapter.

Prerequisites

npm install @s2-dev/resumable-stream @tanstack/ai @tanstack/ai-react @tanstack/ai-client

If you use TanStack’s OpenAI adapter:

npm install @tanstack/ai-openai

Create an S2 access token and basin first:

Sign up here, generate an access token, and set it as S2_ACCESS_TOKEN.
Create a basin with Create Stream on Append and Create Stream on Read enabled, and set it as S2_BASIN.

The helper does not create streams manually. Stream creation should be handled by the basin configuration above.

Import paths

The integration has separate server and client entrypoints.

// Server routes only
import { createResumableChat } from '@s2-dev/resumable-stream/tanstack-ai';

// Browser/client code only
import { createS2Connection } from '@s2-dev/resumable-stream/tanstack-ai/client';

Keep these split. The server entrypoint uses S2 credentials. The client entrypoint is browser-safe and only talks to your HTTP routes.

Setup

Create one chat helper and share it across routes:

lib/s2.ts

import { createResumableChat } from '@s2-dev/resumable-stream/tanstack-ai';

export const chat = createResumableChat({
  accessToken: process.env.S2_ACCESS_TOKEN!,
  basin: process.env.S2_BASIN!,
  mode: 'session',
  enableStop: true,
});

enableStop is only needed if you expose a stop route. Use one stable S2 stream name per chat:

lib/stream-name.ts

const CHAT_ID_PATTERN = /^[a-zA-Z0-9_-]{1,64}$/;

export function isValidChatId(value: unknown): value is string {
  return typeof value === 'string' && CHAT_ID_PATTERN.test(value);
}

export function streamName(chatId: string): string {
  return `tanstack-ai-chat-${chatId}`;
}

Server: POST and DELETE routes

makeSessionResponse persists the latest user text event plus chunks from your TanStack source. The source receives the submitted TanStack UIMessage[].

src/routes/api.chat.ts

import { createFileRoute } from '@tanstack/react-router';
import {
  chat as tanstackChat,
  convertMessagesToModelMessages,
} from '@tanstack/ai';
import { openaiText } from '@tanstack/ai-openai';
import { chat } from '@/lib/s2';
import { isValidChatId, streamName } from '@/lib/stream-name';

export const Route = createFileRoute('/api/chat')({
  server: {
    handlers: {
      POST: async ({ request }) => {
        const body = await request.json();
        if (!isValidChatId(body.id)) {
          return new Response('Missing or invalid id', { status: 400 });
        }
        if (!Array.isArray(body.messages) || body.messages.length === 0) {
          return new Response('Expected at least one message', { status: 400 });
        }

        return chat.makeSessionResponse(streamName(body.id), {
          messages: body.messages,
          source: (messages, { abortController }) =>
            tanstackChat({
              adapter: openaiText(process.env.OPENAI_MODEL ?? 'gpt-4o-mini'),
              messages: convertMessagesToModelMessages(messages),
              abortController,
            }),
        });
      },
      DELETE: async ({ request }) => {
        const body = await request.json();
        if (!isValidChatId(body.id)) {
          return new Response('Missing or invalid id', { status: 400 });
        }

        return chat.stopSession(streamName(body.id));
      },
    },
  },
});

If your runtime can terminate work after a response returns, pass its background-task hook as waitUntil. Without waitUntil, makeSessionResponse waits for persistence before returning.

Server: replay

The replay route delivers both initial history and live chunks.

src/routes/api.chat.replay.ts

import { createFileRoute } from '@tanstack/react-router';
import { chat } from '@/lib/s2';
import { isValidChatId, streamName } from '@/lib/stream-name';

function parseFromSeqNum(value: string | null): number | undefined {
  if (value === null) return undefined;
  const parsed = Number.parseInt(value, 10);
  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : undefined;
}

export const Route = createFileRoute('/api/chat/replay')({
  server: {
    handlers: {
      GET: ({ request }) => {
        const url = new URL(request.url);
        const id = url.searchParams.get('id');
        if (!isValidChatId(id)) {
          return new Response('Missing id query parameter', { status: 400 });
        }

        return chat.replay(streamName(id), {
          fromSeqNum: parseFromSeqNum(url.searchParams.get('from')),
        });
      },
    },
  },
});

When fromSeqNum is omitted, replay compacts completed history into one in-memory MESSAGES_SNAPSHOT, then tails live records. Snapshots are not stored in S2. Replay frames include SSE id values; the client uses them as the next reconnect cursor.

Client

Pass the send and replay endpoints to the TanStack connection. Add stopUrl if you expose a stop route.

app/page.tsx

'use client';

import { useChat } from '@tanstack/ai-react';
import { createS2Connection } from '@s2-dev/resumable-stream/tanstack-ai/client';
import { useMemo } from 'react';

function Chat({ chatId }: { chatId: string }) {
  const connection = useMemo(
    () =>
      createS2Connection({
        mode: 'session',
        sendUrl: '/api/chat',
        stopUrl: '/api/chat',
        subscribeUrl: `/api/chat/replay?id=${encodeURIComponent(chatId)}`,
        body: { id: chatId },
      }),
    [chatId],
  );

  const chat = useChat({ connection, live: true });

  function stop() {
    chat.stop();
    void connection.stop?.();
  }

  // render your chat UI...
  return null;
}

Session mode requires live: true because chunks arrive through the replay subscription, not the POST response.

Flow

The browser mounts and calls connection.subscribe().
chat.replay emits an in-memory snapshot and keeps tailing S2.
The browser sends the normal TanStack messages array.
makeSessionResponse stores stream events in S2 and starts your source.
Model chunks are appended to S2 and delivered through replay.
Reconnects continue from the last SSE id.
If configured, connection.stop?.() calls stopSession.

Modes

mode controls how generations map to S2 streams.

mode	use when	refresh behavior
`single-use`	Each assistant turn has its own stream name.	Replays only the active turn.
`shared`	One stream name should hold the active generation only.	Replays the active generation after lease/fence coordination.
`session`	You are building a full chat session.	Replay bootstraps a snapshot, then tails later chunks.

For single-use and shared, you can omit subscribeUrl and stream chunks directly on the POST response. If you pass subscribeUrl, the client can recover an active generation on mount.

Configuration

Relevant options on createResumableChat:

option	default	what it controls
`mode`	`"single-use"`	`"single-use"` uses one stream per generation. `"shared"` reuses one active-generation stream. `"session"` stores a durable chat session log.
`enableStop`	`false`	Enables process-local `stopSession` support for a stop route. No active-generation map is created unless this is true.
`endpoints`	S2 defaults	Optional endpoint overrides, for example when using S2 Lite.
`leaseDurationMs`	`5000`	Only for `shared` mode. Max pause within an active generation before a new claim can take it over.
`onError`	generic message	Maps upstream errors to the `RUN_ERROR` chunk shown to the client.
`batchSize` / `lingerDuration`	`10` / `50ms`	S2 append batching knobs.

Relevant options on createS2Connection:

option	what it controls
`mode`	Must match the server mode. Use `"session"` for chat sessions.
`sendUrl`	POST endpoint that starts generation.
`stopUrl`	DELETE endpoint that stops process-local generation. Optional.
`subscribeUrl`	GET endpoint that streams chunks. Required for `session`; optional for `single-use` and `shared`.
`headers` / `credentials`	Auth and fetch options for send, stop, and subscribe requests.
`body`	Extra fields merged into the POST request body, and DELETE when `stopUrl` is used.
`fetch`	Custom fetch implementation for tests or framework integrations.

Example

A runnable TanStack Start chat app with session replay and stop support is available in the SDK repo: examples/tanstack-ai-chat.

AI SDKs

Observability

Pipelines

Sync

TanStack AI

Prerequisites

Import paths

Setup

Server: POST and DELETE routes

Server: replay

Client

Flow

Modes

Configuration

Example

AI SDKs

Observability

Pipelines

Sync

Documentation Index

​Prerequisites

​Import paths

​Setup

​Server: POST and DELETE routes

​Server: replay

​Client

​Flow

​Modes

​Configuration

​Example

Prerequisites

Import paths

Setup

Server: POST and DELETE routes

Server: replay

Client

Flow

Modes

Configuration

Example