Resumable Streams

Persist an in-flight LLM response on the server so the client can reload, lose its connection, or open a new tab and pick up the same stream.

assistant-stream/resumable lets you continue a streaming LLM response across client reconnects. The server keeps writing to a store while the original request is in flight; if the browser reloads or loses its connection, a follow-up request replays the persisted bytes plus any new ones until the producer finalizes.

It works with any encoder that already ships in assistant-stream (the AI SDK UI message stream, the data stream protocol, the assistant transport SSE format, or your own), because persistence happens at the byte level after encoding.

What it solves

A user sends a long prompt, walks away, and reloads the tab. Without resumable streams the LLM call is wasted; with them the client picks up where it left off. The same flow handles dropped mobile connections and lets a stream started on one device be read on another, gated by an opaque stream id.

If your responses are short or you do not care about reload survival, the standard streamText().toUIMessageStreamResponse() path is enough.

Server side: minimum wiring

Construct a ResumableStreamContext once per process and reuse it across requests. The context is the seam between your route handlers and the storage backend.

/lib/resumable-context.ts
import {
  createInMemoryResumableStreamStore,
  createResumableStreamContext,
} from "assistant-stream/resumable";

const store = createInMemoryResumableStreamStore();
export const resumableContext = createResumableStreamContext({ store });

In your chat route, wrap the response body in ctx.run(streamId, makeStream). The first caller for streamId becomes the producer (your makeStream callback runs); later callers and reconnects become consumers that replay the persisted bytes.

/app/api/chat/route.ts
import { streamText } from "ai";
import { RESUMABLE_STREAM_ID_HEADER } from "assistant-stream/resumable";
import { resumableContext } from "@/lib/resumable-context";

export async function POST(req: Request) {
  const { messages } = await req.json();
  const streamId = crypto.randomUUID();

  const result = streamText({ /* model, messages, tools, ... */ });
  const sourceBody = result.toUIMessageStreamResponse().body!;

  const stream = await resumableContext.run(streamId, () => sourceBody);

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      [RESUMABLE_STREAM_ID_HEADER]: streamId,
    },
  });
}

A separate GET endpoint replays the persisted bytes for reconnecting clients. ctx.resume(streamId) returns null when no stream exists; use ctx.requireResume(streamId) if you prefer to surface a ResumableStreamError with code "missing" instead.

/app/api/chat/resume/[streamId]/route.ts
import { RESUMABLE_STREAM_ID_HEADER } from "assistant-stream/resumable";
import { resumableContext } from "@/lib/resumable-context";

export async function GET(
  _req: Request,
  ctx: { params: Promise<{ streamId: string }> },
) {
  const { streamId } = await ctx.params;
  const stream = await resumableContext.resume(streamId);
  if (!stream) {
    return new Response(JSON.stringify({ error: "stream not found" }), {
      status: 404,
      headers: { "Content-Type": "application/json" },
    });
  }
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      [RESUMABLE_STREAM_ID_HEADER]: streamId,
    },
  });
}

The context exposes two more verbs: ctx.status(streamId) returns "streaming" | "done" | "error" | "missing", and ctx.delete(streamId) removes all persisted state for a stream and terminates active readers. The remaining options on createResumableStreamContext (onAcquire, onAppend, onFinalize, onError) are observability hooks covered in Resumable Stream Deployment.

Client side: native integration

@assistant-ui/react-ai-sdk ships a resumable option on AssistantChatTransport. It captures the stream id from the response header, redirects chat.resumeStream() reconnects to your resume route, and clears the stored id when the response finishes naturally. Pair it with useChatRuntime, which fires chat.resumeStream() whenever its resumable storage reports a pending id, including ids discovered after mount.

/app/page.tsx
"use client";

import { AssistantRuntimeProvider } from "@assistant-ui/react";
import {
  AssistantChatTransport,
  createResumableSessionStorage,
  useChatRuntime,
} from "@assistant-ui/react-ai-sdk";
import { useMemo } from "react";
import { Thread } from "@/components/assistant-ui/thread";

const storage = createResumableSessionStorage();

export default function Page() {
  const transport = useMemo(
    () =>
      new AssistantChatTransport({
        api: "/api/chat",
        resumable: {
          storage,
          resumeApi: (streamId) => `/api/chat/resume/${streamId}`,
        },
      }),
    [],
  );
  const runtime = useChatRuntime({
    transport,
    onResumeError: (error) => {
      console.error("Could not resume the previous response", error);
    },
  });

  return (
    <AssistantRuntimeProvider runtime={runtime}>
      <Thread />
    </AssistantRuntimeProvider>
  );
}

onResumeError runs when the client finds a stored stream id but the reconnect attempt fails. Use it to show a toast, report telemetry, or mark the thread as needing retry. Assistant-ui clears the failed stream id after the callback unless a newer id has replaced it.

createResumableSessionStorage returns a ResumableClientStorage backed by window.sessionStorage. Pass { key } to namespace per route or chat surface; key also accepts a getter that is read lazily on every access, so you can derive it from the active thread's identity (see Multiple threads). While the getter returns undefined, reads report no pending stream and writes are dropped. Reuse one storage instance per key, because separate instances do not synchronize their in-memory caches. The persisted entry survives reloads, while in-memory ownership prevents one mounted thread from attaching to another thread's pending stream. An idle follower that learns a stream id from its backend can call storage.setStreamId(streamId, threadId) after the corresponding message exists in local history; the runtime attaches when that thread is current, without requiring a remount. The built-in storage holds one pending stream id per key; the Multiple threads pattern gives each thread its own key, so several threads stay resumable at once. A custom ResumableClientStorage can instead key its records by the optional threadId arguments and implement subscribe(listener, threadId) for post-mount updates, or replace the storage methods entirely. If you are running on a transport that already wraps fetch or prepareReconnectToStreamRequest, the resumable option composes with your existing handlers.

The default finish detector scans the SSE body for the AI SDK "type":"finish" marker. Override isFinishEvent on the resumable option when you ship a custom encoder.

Multiple threads

The snippet above stores the pending stream id under one sessionStorage key, which is correct for a single chat surface. Under a thread list runtime (useRemoteThreadListRuntime, which useChatRuntime wraps) with more than one alive thread, a single shared key is written and cleared by whichever thread acts last: switch threads while a response is in flight and the response header writes its stream id to the shared key, which the thread you switched to reads on mount and resumes — replaying the other conversation's stream into this one. An unscoped clear when a stream finishes can also drop another thread's pending id. And after a reload, a pending id under the shared key is claimed by whichever thread is main at mount, which with a thread list is a fresh empty thread rather than the one that started the stream.

Scope the key to the thread, and construct the transport and storage per thread inside the per-thread runtime hook rather than once at module level. Derive the key from the thread's identity with a getter so it is read lazily on every access:

/app/page.tsx
"use client";

import {
  AssistantCloud,
  AssistantRuntimeProvider,
  useAui,
  useCloudThreadListAdapter,
  useRemoteThreadListRuntime,
} from "@assistant-ui/react";
import {
  AssistantChatTransport,
  createResumableSessionStorage,
  useChatRuntime,
} from "@assistant-ui/react-ai-sdk";
import { useMemo } from "react";
import { Thread } from "@/components/assistant-ui/thread";
import { ThreadList } from "@/components/assistant-ui/thread-list";

const cloud = new AssistantCloud({
  baseUrl: process.env.NEXT_PUBLIC_ASSISTANT_BASE_URL!,
  anonymous: true,
});

function ResumableThreadRuntime() {
  const aui = useAui();
  const transport = useMemo(
    () =>
      new AssistantChatTransport({
        api: "/api/chat",
        resumable: {
          storage: createResumableSessionStorage({
            key: () => {
              const item = aui.threadListItem.getState();
              return `aui-resumable-stream-id:${item.remoteId ?? item.id}`;
            },
          }),
          resumeApi: (streamId) => `/api/chat/resume/${streamId}`,
        },
      }),
    [aui],
  );
  return useChatRuntime({ transport });
}

export default function Page() {
  const adapter = useCloudThreadListAdapter({ cloud });
  const runtime = useRemoteThreadListRuntime({
    adapter,
    runtimeHook: ResumableThreadRuntime,
  });

  return (
    <AssistantRuntimeProvider runtime={runtime}>
      <ThreadList />
      <Thread />
    </AssistantRuntimeProvider>
  );
}

useCloudThreadListAdapter is the same adapter useChatRuntime({ cloud }) builds internally, so cloud thread history and attachments keep working; to bring your own backend, pass a custom RemoteThreadListAdapter instead. Each thread's runtime hook constructs its own transport and storage, keyed by that thread's identity, so one conversation's stream id can never be read or cleared by another. useChatRuntime is a no-op thread list when nested as a runtimeHook, so it only runs the per-thread chat runtime against the per-thread transport; the real thread list is the outer useRemoteThreadListRuntime.

Key by remoteId ?? id under a recognizable prefix, read lazily. A brand-new thread has no remoteId yet, but the transport initializes the thread before the request goes out, so by the time the response header delivers the stream id the getter resolves to the remoteId. After a reload the same conversation comes back with that remoteId as its id, so the stored entry is found again and the pending stream resumes. The local-id fallback covers a thread that has never sent, whose key nothing else reads or writes. Keying by the local id alone would break resume across reloads for threads created in the current session, because their local __LOCALID_ id is replaced by the remoteId when the list reloads.

Storage choices

The core package ships createInMemoryResumableStreamStore for development and tests. State lives in a process-local Map, so it does not survive a server restart. Useful options include defaultTtlMs, maxChunkBytes, maxEntriesPerStream, maxStreams, and gcIntervalMs for periodic eviction.

For production, use one of the optional Redis adapters via the assistant-stream/resumable/redis (node-redis v5) or assistant-stream/resumable/ioredis sub-paths. Both adapters batch the per-append XADD and TTL refresh into a single pipelined round trip, store chunk values as binary, and accept the same keyPrefix, defaultTtlMs, pollIntervalMs, and maxChunkBytes options. Cluster routing works because each stream's keys share a {streamId} hash tag.

/lib/resumable-context.ts
import {
  createResumableStreamContext,
  type ResumableStreamStore,
} from "assistant-stream/resumable";

async function createStore(): Promise<ResumableStreamStore> {
  if (!process.env.REDIS_URL) {
    const { createInMemoryResumableStreamStore } = await import(
      "assistant-stream/resumable"
    );
    return createInMemoryResumableStreamStore();
  }
  const { createClient } = await import("redis");
  const { createRedisResumableStreamStore } = await import(
    "assistant-stream/resumable/redis"
  );
  const client = createClient({ url: process.env.REDIS_URL });
  await client.connect();
  return createRedisResumableStreamStore(client);
}

export const resumableContext = createResumableStreamContext({
  store: await createStore(),
});

For Postgres, Cloudflare Durable Objects, Upstash REST, or any other backend, implement the ResumableStreamStore interface directly. See Custom Resumable Stream Stores for the contract walkthrough and a worked example.

Production checklist

  • Auth. The resume route in the snippets above will serve any caller that knows the stream id. Bind streamId to the requesting user at acquire time and verify the binding inside the resume handler. Treat the id as opaque, not as a credential; it leaks via response headers, sessionStorage, browser history, and access logs.
  • waitUntil on serverless. On Vercel and Cloudflare the request handler is killed once the response returns, which interrupts the producer task. Pass after from next/server (or your platform's ctx.waitUntil) when constructing the context so the task survives past the response: createResumableStreamContext({ store, waitUntil: after }).
  • TTL. Streams expire 24 hours after the last write by default. Configure with defaultTtlMs on the store, or override per deployment via ttlMs on the context. Match TTLs across the store, any owner-binding key, and any signed cookie that references a streamId.
  • Stream id format. The Redis adapters validate streamId against /^[A-Za-z0-9_.:-]{1,256}$/ to keep keys well-formed. UUIDv4 is fine.

For the full treatment of authorization, multi-tenant key prefixes, observability hooks, resource limits, and incident response, see Resumable Stream Deployment.

A new ResumableStreamError class is exported from assistant-stream/resumable with codes "missing" | "exists" | "finalized" | "invalid-id"; catch it in the resume route to distinguish "stream gone" from other failures.

Helpers for AssistantStreamController callbacks

If you produce streams via createAssistantStream rather than the AI SDK, the package ships two helpers that bridge the controller-callback style and any encoder to the store:

import {
  createResumableAssistantStreamResponse,
  createResumeAssistantStreamResponse,
} from "assistant-stream/resumable";
import { resumableContext } from "@/lib/resumable-context";

// POST handler
return createResumableAssistantStreamResponse({
  context: resumableContext,
  streamId,
  callback: (controller) => {
    /* same shape as createAssistantStreamResponse */
  },
});

// GET resume handler
return createResumeAssistantStreamResponse({
  context: resumableContext,
  streamId,
});

Both helpers default to the data-stream encoder; pass encoder: () => new AssistantTransportEncoder() (or any custom encoder) to override. They set the x-resumable-stream-id response header automatically, which is what AssistantChatTransport's resumable adapter looks for.

Example app

examples/with-resumable-stream is a runnable Next.js app that uses useChat, the resumable transport option, and useChatRuntime. It falls back to a built-in mock when OPENAI_API_KEY is unset, and switches the store from in-memory to Redis when REDIS_URL is set.

npx assistant-ui create my-app -e with-resumable-stream

To drop the resumable chat and resume routes into an existing project (in-memory store only; upgrade via Storage choices):

npx shadcn@latest add @assistant-ui/ai-sdk-backend-resumable

The @assistant-ui namespace resolves the Radix or Base UI flavor from your project's style through the style-aware registry entry in components.json. Without that entry, add by direct URL instead:

npx shadcn@latest add https://r.assistant-ui.com/base/ai-sdk-backend-resumable.json