feat: WebSocket transport — full-duplex resumable streams

.changeset/websocket-transport.md

@@ -0,0 +1,42 @@
+---
+'@tanstack/ai': minor
+'@tanstack/ai-client': minor
+'@tanstack/ai-react': minor
+'@tanstack/ai-solid': minor
+'@tanstack/ai-vue': minor
+'@tanstack/ai-svelte': minor
+'@tanstack/ai-angular': minor
+---
+
+WebSocket transport: a full-duplex, resumable third transport alongside SSE and
+NDJSON, reusing the same delivery-durability seam.
+
+On the server, `@tanstack/ai` adds `toWebSocketStream(socket, request, { onRun,
+durability, batch, heartbeatMs, idleTimeoutMs, debug })` — a portable core that
+pumps a conversation over an already-accepted WHATWG `WebSocketLike` server
+socket (Node via `ws`, Bun, etc.), and `toWebSocketResponse(request, { onRun,
+… })`, a thin wrapper that upgrades via `WebSocketPair` and returns a 101
+`Response` on Cloudflare Workers/Durable Objects (it throws elsewhere, pointing
+you at `toWebSocketStream`). Because one socket outlives many `chat()` turns
+(client-tool resubmits, follow-up user messages), you pass an `onRun(ctx) =>
+AsyncIterable<StreamChunk>` factory instead of a prebuilt stream — the helper
+calls it per inbound `RunAgentInput` frame. The socket is conversation-scoped:
+it stays open across turns and closes on client close, an `{ type: 'abort',
+runId }` control frame (which aborts only that turn), or the idle timeout, with a
+periodic `{ type: 'ping' }` heartbeat. Durability is keyed per turn and reuses
+the existing `durableStreamSource`, so server→client frames carry the same
+`{ id, chunk }` envelope as NDJSON. `resumeWebSocketStream(socket, { adapter })`
+and `resumeWebSocketResponse({ adapter })` replay a run read-only from the
+durability log (no model call).
+
+On the client, `webSocket(url, options)` (in `@tanstack/ai-client`, re-exported
+from `@tanstack/ai-react`, `-solid`, `-vue`, `-svelte`, and `-angular`) is a
+full-duplex `subscribe` + `send` connection adapter for `useChat`. `send()`
+writes a `RunAgentInput` frame; `subscribe()` yields inbound chunks, ignores
+heartbeats, unwraps durable envelopes, and auto-reconnects a dropped durable run
+by reopening with `?runId=&offset=` (browsers can't set a `Last-Event-ID`
+handshake header, so the offset rides in the URL). The reconnect bookkeeping
+(offset de-dupe, no-progress ceiling → `StreamReconnectLimitError`) is shared
+with the HTTP adapters via the new `createReconnectTracker`, and a fatal drop
+surfaces to the consumer (`StreamReadError` / `StreamReconnectLimitError`)
+instead of hanging.

docs/chat/connection-adapters.md

@@ -32,7 +32,8 @@ This page covers every supported transport, when to pick which, and how to build
 | Code that **synchronously** returns an `AsyncIterable<StreamChunk>` (in-process `chat()`, an RSC stream, tests) | [`stream`](#server-functions-and-direct-async-iterables) |
 | An **async** call — a TanStack Start server function or any `Promise`-returning function — resolving to a `Response` or an `AsyncIterable<StreamChunk>` | [`fetcher`](#server-functions-via-fetcher) |
 | An RPC framework like Cap'n Web, gRPC-Web, or tRPC | [`rpcStream`](#rpc-streams) |
-| A single long-lived WebSocket (or BroadcastChannel, postMessage, shared worker) serving many runs | [Custom `subscribe` / `send` adapter](#persistent-transports-websockets-and-friends) |
+| A single long-lived, resumable WebSocket serving many runs | [`webSocket`](#websockets) |
+| BroadcastChannel, postMessage, a shared worker, or another persistent transport | [Custom `subscribe` / `send` adapter](#persistent-transports-websockets-and-friends) |
 | Standard SSE but with custom fetch wrapping (auth refresh, retries) | [`fetchServerSentEvents` with `fetchClient`](#custom-fetch-client) |
 | Something else entirely (HTTP/3, Server-Sent Events over a different protocol, etc.) | [Custom `connect` adapter](#custom-request-scoped-adapters) |
 
@@ -295,11 +296,25 @@ const { messages } = useChat({
 });
 ```
 
+## WebSockets
+
+For a persistent, resumable WebSocket, use the built-in `webSocket()` adapter instead of hand-rolling a `SubscribeConnectionAdapter`. It opens one socket for the whole conversation, reconnects a dropped durable run automatically, and pairs with the server's `toWebSocketStream` / `toWebSocketResponse`:
+
+```typescript
+import { useChat, webSocket } from "@tanstack/ai-react";
+
+const connection = webSocket("/api/chat-ws");
+
+const { messages, sendMessage } = useChat({ connection });
+```
+
+See [WebSockets](../resumable-streams/websockets) for the server side, the wire protocol, reconnect details, and hosting on Node vs Cloudflare.
+
 ## Persistent Transports (WebSockets and Friends)
 
 A persistent transport — WebSocket, BroadcastChannel, postMessage between iframes, a shared worker — is fundamentally different from request/response. You open the channel **once**, then send and receive over it for the lifetime of the client. `stream()`/`connect()` can't model this cleanly because they assume one async iterable per request.
 
-For these cases, implement the `SubscribeConnectionAdapter` interface directly. The shape (full definition in [The Adapter Interface](#the-adapter-interface)):
+The built-in `webSocket()` adapter above covers the common resumable WebSocket case. For anything else persistent, implement the `SubscribeConnectionAdapter` interface directly. The shape (full definition in [The Adapter Interface](#the-adapter-interface)):
 
 ```typescript
 import type { SubscribeConnectionAdapter } from "@tanstack/ai-react";
@@ -313,7 +328,9 @@ import type { SubscribeConnectionAdapter } from "@tanstack/ai-react";
 
 The runtime correlates them: chunks emitted on the subscription queue between `send()` and the next terminal event (`RUN_FINISHED` / `RUN_ERROR`) are attributed to that run.
 
-### WebSocket example
+### Custom WebSocket example
+
+Building your own protocol instead of the built-in `webSocket()` adapter (a different wire format, no resume support needed, or a server you don't control)? Implement `SubscribeConnectionAdapter` by hand:
 
 ```typescript
 import { useChat, type SubscribeConnectionAdapter } from "@tanstack/ai-react";

docs/config.json

@@ -168,7 +168,7 @@
           "label": "Connection Adapters",
           "to": "chat/connection-adapters",
           "addedAt": "2026-04-15",
-          "updatedAt": "2026-07-17"
+          "updatedAt": "2026-07-20"
         },
         {
           "label": "Thinking & Reasoning",
@@ -188,12 +188,19 @@
         {
           "label": "Overview",
           "to": "resumable-streams/overview",
-          "addedAt": "2026-07-17"
+          "addedAt": "2026-07-17",
+          "updatedAt": "2026-07-20"
         },
         {
           "label": "Advanced",
           "to": "resumable-streams/advanced",
-          "addedAt": "2026-07-17"
+          "addedAt": "2026-07-17",
+          "updatedAt": "2026-07-20"
+        },
+        {
+          "label": "WebSockets",
+          "to": "resumable-streams/websockets",
+          "addedAt": "2026-07-20"
         },
         {
           "label": "Custom Durability Adapter",

docs/resumable-streams/advanced.md

@@ -165,6 +165,50 @@ read failure it retries from the last valid position, capping consecutive
 failures (`reconnect: { maxReadFailures: 10, delayMs: 250 }`). Normal long-poll
 advancement is never throttled.
 
+## WebSocket transport
+
+Everything above describes SSE and NDJSON: one connection per turn. A
+WebSocket instead stays open for the whole conversation, so a few things work
+differently. See [WebSockets](./websockets) for the full protocol and code;
+this section covers only what changes relative to the rest of this page.
+
+**Reconnect rides in the URL, not a header.** SSE and NDJSON resume with
+`Last-Event-ID` because `fetch`/XHR can set arbitrary headers before opening a
+request. A browser's `WebSocket` constructor can't set custom headers on the
+handshake, so the offset instead rides in the URL: `?runId=<id>&offset=<lastId>`.
+`webSocket()` reopens at that URL automatically on a durable run's drop, the
+same de-dupe guarantee as `fetchServerSentEvents`, just carried differently on
+the wire.
+
+**The socket is conversation-scoped, not run-scoped.** One socket carries many
+turns. `toWebSocketStream`'s `durability` option is therefore a factory keyed
+by each turn's `runId` (via a synthetic per-turn request), not a single value
+built once. The socket itself closes on client close, an idle timeout, or
+process shutdown, not when one turn's `RUN_FINISHED` arrives.
+
+**Heartbeat and idle timeout replace the transport-level keepalive SSE gets
+for free.** `toWebSocketStream` pings every `heartbeatMs` (default 30s) and
+closes the socket after `idleTimeoutMs` (default 5 minutes) with no inbound
+frame.
+
+**An abort frame targets one turn, not the socket.** `{ type: 'abort', runId }`
+aborts only that turn's `onRun` iteration; the socket stays open for the next
+turn. Closing the socket aborts every turn still in flight on it.
+
+**Hosting differs by runtime.** On Cloudflare Workers or Durable Objects,
+`toWebSocketResponse` uses the global `WebSocketPair` and needs no manual
+upgrade. On Node (or anywhere without `WebSocketPair`), you upgrade the
+connection yourself, typically with `ws`'s `WebSocketServer({ noServer: true })`
+hooked into the HTTP server's `upgrade` event, and hand the resulting socket to
+`toWebSocketStream` / `resumeWebSocketStream` directly.
+
+**The producer-vs-socket caveat from [memoryStream in
+production](#memorystream-in-production) applies unchanged.** With
+`memoryStream`, a dropped WebSocket aborts the `chat()` call backing it just
+like a dropped SSE connection does; reconnecting replays the log rather than
+resuming a still-running model call. `durableStream` decouples the two exactly
+as it does for SSE and NDJSON.
+
 ## Offset ownership
 
 `StreamDurability<TOffset>` owns its offset format. Core only passes returned

docs/resumable-streams/overview.md

@@ -138,6 +138,45 @@ For NDJSON, swap `fetchServerSentEvents` for `fetchHttpStream` (with the server
 on `toHttpResponse`). The XHR adapters (`xhrServerSentEvents`, `xhrHttpStream`)
 work the same way, for runtimes without streaming `fetch`.
 
+## Or go full-duplex: WebSockets
+
+SSE and NDJSON open one connection per turn. If you want a single persistent
+socket that carries the whole conversation instead, wrap it with
+`toWebSocketStream` (or `toWebSocketResponse` on Cloudflare) and pair it with
+the client's `webSocket()` adapter:
+
+```ts
+import { chat, memoryStream, toWebSocketStream } from '@tanstack/ai'
+import { openaiText } from '@tanstack/ai-openai'
+import type { WebSocketLike } from '@tanstack/ai'
+
+// `socket` is a server socket you already accepted (see WebSockets for how,
+// on Node vs Cloudflare) and `request` is the handshake request.
+function handleChatSocket(socket: WebSocketLike, request: Request) {
+  toWebSocketStream(socket, request, {
+    durability: (ctx) => memoryStream(ctx.request),
+    onRun: ({ messages, threadId, runId }) =>
+      chat({ adapter: openaiText('gpt-5.5'), messages, threadId, runId }),
+  })
+}
+```
+
+```tsx
+import { useChat, webSocket } from '@tanstack/ai-react'
+
+const connection = webSocket('/api/chat-ws')
+
+export function Chat() {
+  const { messages, sendMessage } = useChat({ connection })
+  return <button onClick={() => void sendMessage('Hello')}>Send</button>
+}
+```
+
+The socket resumes the same way SSE and NDJSON do: a dropped connection
+reopens with the last offset and replays only what's missing. See
+[WebSockets](./websockets) for the wire protocol, reconnect details, and
+hosting on Node vs Cloudflare.
+
 That covers the common case. For the durability contract, terminal and error
 handling, reconnect tuning, attaching to a run by id, Cloudflare deployment, and
 production process-death concerns, see [Advanced](./advanced).