feat: resumable streams — reconnect to in-flight SSE responses via pluggable delivery durability

.changeset/resumable-streams.md

@@ -0,0 +1,53 @@
+---
+'@tanstack/ai': minor
+'@tanstack/ai-client': minor
+'@tanstack/ai-durable-stream': minor
+---
+
+Resumable streams: reconnect to an in-flight SSE **or NDJSON** response without
+re-running the provider.
+
+`toServerSentEventsResponse` and `toHttpResponse` both accept a
+`durability: { adapter, batch }` option. The adapter (`StreamDurability`)
+records every chunk to an ordered log before delivery and tags each event with
+an opaque, adapter-owned offset — an SSE `id:` line, or the `id` of an NDJSON
+`{ id, chunk }` envelope (NDJSON has no native event-id). A reconnect
+(`Last-Event-ID`) or an explicit `?offset` read replays strictly after that
+offset from the log — the lazy provider stream is never iterated on resume.
+Producers terminalize the log on cancellation and failure (`RUN_ERROR` append
+
+- `close()`) and on completion when the source stream emits its own terminal
+  event (`chat()` always does), so readers are never parked on a dead run.
+
+Two adapters ship: `memoryStream(request)` in `@tanstack/ai` (process-local,
+for development and tests) and the new `@tanstack/ai-durable-stream` package,
+a Durable Streams protocol adapter for production backends.
+
+For the `GET` handler that a reload or a second tab reconnects to,
+`resumeServerSentEventsResponse({ adapter })` and `resumeHttpResponse({ adapter })`
+replay a run straight from the durability log. They need no producer stream and
+return a 400 when the request carries no resume offset.
+
+On the client, all four HTTP adapters are now resumable — `fetchServerSentEvents`,
+`fetchHttpStream`, `xhrServerSentEvents`, and `xhrHttpStream`. Each tracks the
+per-event offset, auto-reconnects with `Last-Event-ID`, de-duplicates the
+replayed prefix, and exposes `joinRun(runId)` to attach to an in-flight or
+finished run from the start (read-only GET with `offset=-1`). Untagged streams
+behave exactly as before. A durable run that ends with no terminal event and no
+forward progress now throws `DurableStreamIncompleteError` instead of hanging.
+
+Reconnection and durability are bounded so failures surface rather than hang or
+loop:
+
+- `memoryStream` evicts completed logs after a grace window (unbounded growth
+  is gone); resuming an expired/unknown run throws, and a from-start join to a
+  run that never produces fails after `MemoryStreamOptions.firstChunkDeadlineMs`.
+- all four HTTP adapters accept `reconnect: { maxAttempts, delayMs }` — a
+  throttle plus a ceiling on CONSECUTIVE no-progress reconnects (default 5;
+  forward progress resets it) that fails with the new `StreamReconnectLimitError`
+  instead of reconnecting endlessly, without penalizing a healthy long-lived run.
+- `durableStream` accepts `reconnect: { maxReadFailures, delayMs }` to bound its
+  read-retry loop, and `server` is now optional when `fetch` is provided (e.g. a
+  Cloudflare service binding).
+- `toServerSentEventsResponse` accepts `debug` to record durability terminal /
+  close failures server-side, where a replaying joiner cannot observe them.

docs/chat/connection-adapters.md

@@ -73,13 +73,65 @@ import { useChat, fetchServerSentEvents } from "@tanstack/ai-react";
 
 const { messages } = useChat({
   connection: fetchServerSentEvents("/api/chat", {
-    body: { provider: "openai", model: "gpt-5.1" },
+    body: { provider: "openai", model: "gpt-5.5" },
   }),
 });
 ```
 
 > **Tip:** `body` and `forwardedProps` populate the same wire field. Use `body` for static defaults, the `forwardedProps` constructor option (or per-`sendMessage` `data`) for dynamic values. Runtime values always win.
 
+### Resumable SSE
+
+`fetchServerSentEvents` watches SSE `id:` values. If a connection drops after
+receiving an id, it reconnects with `Last-Event-ID` and de-duplicates the
+replayed prefix. `joinRun(runId)` performs a read-only GET with `offset=-1` and
+the run id, replaying an in-flight or finished run from the start.
+
+The ids only appear when the server passes a durability adapter to
+`toServerSentEventsResponse`. They are opaque tokens owned by that adapter; the
+chat client does not create, parse, or persist them. Without ids, behavior is
+identical to a plain single fetch. See
+[Resumable Streams](../resumable-streams/overview).
+
+Your route needs a `GET` handler alongside `POST` for `joinRun` (second tab or
+reload) to work. `POST` handles fresh runs and auto-reconnects (it re-sends the
+same body with `Last-Event-ID`); `GET` replays a known run from the start:
+
+```typescript
+import {
+  chat,
+  chatParamsFromRequest,
+  memoryStream,
+  resumeServerSentEventsResponse,
+  toServerSentEventsResponse,
+} from "@tanstack/ai";
+import { openaiText } from "@tanstack/ai-openai";
+
+export async function POST(request: Request) {
+  const { messages, threadId, runId } = await chatParamsFromRequest(request);
+  const stream = chat({ adapter: openaiText("gpt-5.5"), messages, threadId, runId });
+  return toServerSentEventsResponse(stream, {
+    durability: { adapter: memoryStream(request) },
+  });
+}
+
+// joinRun hits GET ?offset=-1&runId=... (replay only, no messages sent).
+export async function GET(request: Request) {
+  return resumeServerSentEventsResponse({ adapter: memoryStream(request) });
+}
+```
+
+The `GET` handler calls no provider: on a replay the durability adapter's
+`resumeFrom()` is non-null (from `?offset`), so the log is replayed instead.
+`resumeServerSentEventsResponse` returns a 400 when the request has no resume
+offset. Use `resumeHttpResponse` for the NDJSON adapters.
+
+`fetchHttpStream` and `xhrHttpStream` resume the same way over NDJSON, where the
+offset rides in an `{ id, chunk }` envelope (see below) instead of an SSE `id:`
+line. Enable it by passing a durability adapter to `toHttpResponse`.
+`xhrServerSentEvents` resumes over SSE exactly like `fetchServerSentEvents`
+(paired with `toServerSentEventsResponse` and its `id:` lines).
+
 ## HTTP Streaming (NDJSON)
 
 For environments that don't speak SSE — some edge runtimes, certain mobile WebViews, or anywhere a proxy strips `text/event-stream` — use raw newline-delimited JSON. The wire format is one JSON `StreamChunk` per line:
@@ -92,7 +144,9 @@ const { messages } = useChat({
 });
 ```
 
-Server-side, write each chunk as `JSON.stringify(chunk) + "\n"` to the response body. Options (`url`, `headers`, `body`, `fetchClient`, dynamic functions) match `fetchServerSentEvents` exactly.
+Server-side, write each chunk as `JSON.stringify(chunk) + "\n"` to the response body (or use `toHttpResponse(stream)`). Options (`url`, `headers`, `body`, `fetchClient`, dynamic functions) match `fetchServerSentEvents` exactly.
+
+`fetchHttpStream` is also resumable: pass a durability adapter to `toHttpResponse` and each line becomes an `{ id, chunk }` envelope. A dropped connection reconnects with `Last-Event-ID`, de-duplicates the replayed prefix, and `joinRun(runId)` attaches to an existing run. Same guarantees as [Resumable SSE](#resumable-sse), over NDJSON.
 
 ## React Native and Expo
 
@@ -129,6 +183,10 @@ const chat = useChat({
 });
 ```
 
+Mobile connections drop often, so this is where resumability pays off most.
+Both XHR adapters reconnect and `joinRun` when the server adds a durability
+adapter. See [Resumable Streams](../resumable-streams/overview).
+
 Use `xhrServerSentEvents()` when your server returns `text/event-stream` via
 `toServerSentEventsResponse()`:
 
@@ -203,7 +261,7 @@ export const chatFn = createServerFn({ method: "POST" })
   .inputValidator((data: { messages: Array<UIMessage> }) => data)
   .handler(({ data }) =>
     toServerSentEventsResponse(
-      chat({ adapter: openaiText("gpt-5.1"), messages: data.messages }),
+      chat({ adapter: openaiText("gpt-5.5"), messages: data.messages }),
     ),
   );
 ```
@@ -292,20 +350,30 @@ function websocketConnection(url: string): SubscribeConnectionAdapter {
 
   return {
     async *subscribe(abortSignal) {
-      while (!abortSignal?.aborted && !closed) {
-        const buffered = queue.shift();
-        if (buffered !== undefined) {
-          yield buffered;
-          continue;
-        }
-        const chunk = await new Promise<StreamChunk | null>((resolve) => {
-          pending = resolve;
-          abortSignal?.addEventListener("abort", () => resolve(null), {
-            once: true,
+      // Register the abort listener once (not per-iteration) so it can't
+      // accumulate on a long-lived socket.
+      const onAbort = () => deliver(null);
+      abortSignal?.addEventListener("abort", onAbort, { once: true });
+      try {
+        while (!abortSignal?.aborted) {
+          // Drain buffered chunks BEFORE honoring `closed`: a burst of messages
+          // followed by a close event (common within one macrotask) must still
+          // deliver the queued chunks (including a trailing RUN_FINISHED),
+          // otherwise the client would hang waiting for a terminal it dropped.
+          const buffered = queue.shift();
+          if (buffered !== undefined) {
+            yield buffered;
+            continue;
+          }
+          if (closed) return;
+          const chunk = await new Promise<StreamChunk | null>((resolve) => {
+            pending = resolve;
           });
-        });
-        if (chunk === null) return;
-        yield chunk;
+          if (chunk === null) return;
+          yield chunk;
+        }
+      } finally {
+        abortSignal?.removeEventListener("abort", onAbort);
       }
     },
 

docs/config.json

@@ -166,7 +166,8 @@
         {
           "label": "Connection Adapters",
           "to": "chat/connection-adapters",
-          "addedAt": "2026-04-15"
+          "addedAt": "2026-04-15",
+          "updatedAt": "2026-07-17"
         },
         {
           "label": "Thinking & Reasoning",
@@ -180,6 +181,26 @@
         }
       ]
     },
+    {
+      "label": "Resumable Streams",
+      "children": [
+        {
+          "label": "Overview",
+          "to": "resumable-streams/overview",
+          "addedAt": "2026-07-17"
+        },
+        {
+          "label": "Advanced",
+          "to": "resumable-streams/advanced",
+          "addedAt": "2026-07-17"
+        },
+        {
+          "label": "Custom Durability Adapter",
+          "to": "resumable-streams/custom-adapter",
+          "addedAt": "2026-07-17"
+        }
+      ]
+    },
     {
       "label": "Protocol",
       "children": [