fix(vercel): return real turn.stream to useChat

Fixes an issue where `sendMessages()` returned an empty `ReadableStream`
to `useChat`. The stream returned here is used to power important
features of useChat, such as:
- status changes (submitted -> streaming -> ready)
- turn lifecycle hooks (e.g. `onToolCall`, `onData`, `onFinish`)
- evaluating whether `sendAutomaticallyWhen` should send

With an empty stream, none of these features worked. This transport aims
to, as much as possible, be a drop-in replacement for the default
transport so we need to make these features work.

This commit returns the real stream from `turn.stream` to `useChat`,
which allows these features to work as expected. As a result of this
change, both `useChat` and `useMessageSync` now accumulate messages in
parallel but both produce identical messages from the same underlying
Ably events.

When the AI SDK doesn't set messageId on the start chunk (currently this
is logically equivalent to not being in 'persistence mode') useChat and
the transport will assign different IDs. This commit adds `messageId` to
`EncoderOptions` so that the server transport can pass its generated ID
to the vercel encoder which uses it as a fallback, which ensures that
both sides converge on the same ID.

Author: owenpearson

docs/internals/chat-transport.md

@@ -9,7 +9,7 @@ Vercel's `useChat` manages message state internally. When the user submits a mes
 1. Determine which messages are new vs history
 2. Compute fork metadata for regeneration
 3. Delegate to the core transport's `send()`
-4. Return a stream that signals completion without duplicating state
+4. Return the turn stream so `useChat` can drive status and callbacks
 
 ## sendMessages
 
@@ -28,11 +28,13 @@ The `prepareSendMessagesRequest` hook (optional) lets the server app customize t
 
 Without the hook, the adapter builds a default body with `history` (including per-message Ably headers), `id`, `trigger`, and fork metadata fields.
 
-### Empty stream return
+### Real stream return
 
-The adapter returns an **empty stream** that closes when the turn ends - not the real event stream. This is intentional: `useChat` consumes the returned stream to accumulate the assistant message, but `useMessageSync` (the companion React hook) already pushes the transport's authoritative message state into `useChat` via `setMessages`. Returning the real event stream would cause `useChat` to accumulate a duplicate assistant message.
+The adapter returns the real turn stream from `sendMessages()`. `useChat` consumes this stream to drive status transitions (`submitted` -> `streaming` -> `ready`), fire callbacks (`onToolCall`, `onData`, `onFinish`), and evaluate `sendAutomaticallyWhen`.
 
-The empty stream is created via a `TransformStream` whose writable side closes when the turn's real stream finishes.
+Both `useChat` and `useMessageSync` accumulate messages in parallel: `useChat` builds from the stream, while `useMessageSync` pushes from the transport's message store via `setMessages` (a full replacement). The transport's version is always authoritative - both accumulators produce identical messages from the same chunks, and `setMessages` overwrites `useChat`'s state on every transport event.
+
+The server encoder ensures `messageId` alignment by stamping the transport-assigned `x-ably-msg-id` as a fallback domain `messageId` on the `start` chunk. This ensures both accumulators assign the same message ID.
 
 ### Abort signal
 

src/core/codec/types.ts

@@ -223,6 +223,12 @@ export interface EncoderOptions {
   extras?: Extras;
   /** Hook called before each Ably message is published. Mutate the message in place to add transport-level headers. */
   onMessage?: (message: Ably.Message) => void;
+  /**
+   * Domain-level message identity. Domain encoders use this as a fallback
+   * messageId when a lifecycle chunk (e.g. `start`) does not provide one,
+   * ensuring useChat and the transport accumulator assign the same ID.
+   */
+  messageId?: string;
 }
 
 /**

src/core/transport/server-transport.ts

@@ -381,17 +381,19 @@ class DefaultServerTransport<TEvent, TMessage> implements ServerTransport<TEvent
         const assistantParent =
           streamOpts?.parent === undefined ? (turnParent ?? undefined) : (streamOpts.parent ?? undefined);
 
+        const msgId = crypto.randomUUID();
         const defaultHeaders = buildTransportHeaders({
           role: 'assistant',
           turnId,
-          msgId: crypto.randomUUID(),
+          msgId,
           turnClientId: turnOwnerClientId,
           parent: assistantParent,
           forkOf: streamOpts?.forkOf ?? turnForkOf,
         });
         const encoder = codec.createEncoder(channel, {
           extras: { headers: defaultHeaders },
           onMessage,
+          messageId: msgId,
         });
 
         const result = await pipeStream(stream, encoder, signal, onAbort, logger);

src/vercel/codec/encoder.ts

@@ -46,10 +46,12 @@ import { headerWriter } from '../../utils.js';
 
 class DefaultUIMessageEncoder implements StreamEncoder<AI.UIMessageChunk, AI.UIMessage> {
   private readonly _core: EncoderCore;
+  private readonly _messageId: string | undefined;
   private _aborted = false;
 
   constructor(writer: ChannelWriter, options: EncoderCoreOptions = {}) {
     this._core = createEncoderCore(writer, options);
+    this._messageId = options.messageId;
   }
 
   async appendEvent(chunk: AI.UIMessageChunk, perWrite?: WriteOptions): Promise<void> {
@@ -144,7 +146,7 @@ class DefaultUIMessageEncoder implements StreamEncoder<AI.UIMessageChunk, AI.UIM
 
       case 'start': {
         const h = headerWriter()
-          .str('messageId', chunk.messageId)
+          .str('messageId', chunk.messageId ?? this._messageId)
           .json('messageMetadata', chunk.messageMetadata)
           .build();
         await this._core.publishDiscrete({ name: 'start', data: '', headers: h }, perWrite);

src/vercel/transport/chat-transport.ts

@@ -242,34 +242,15 @@ export const createChatTransport = (
       });
     }
 
-    // Return an empty stream that closes when the turn ends.
-    // useChat consumes the returned stream to accumulate the assistant message,
-    // but useMessageSync already pushes the transport's authoritative message
-    // state into useChat via setMessages. Returning the real event stream would
-    // cause useChat to accumulate a duplicate assistant message. Instead, we
-    // return a stream that produces no chunks and closes when the turn's stream
-    // finishes, so useChat knows when streaming is done without duplicating state.
-    const { readable, writable } = new TransformStream<AI.UIMessageChunk>();
-    const writer = writable.getWriter();
-    // Fire-and-forget: we only care about the close/abort signal, not the piped data.
-    // Errors on the turn stream are surfaced via transport.on('error'), not here.
-    /* eslint-disable @typescript-eslint/no-empty-function -- swallow: writer.close() rejection after stream teardown is unrecoverable */
-    turn.stream
-      .pipeTo(
-        new WritableStream({
-          close: () => {
-            writer.close().catch(() => {});
-          },
-          abort: () => {
-            writer.close().catch(() => {});
-          },
-        }),
-      )
-      .catch(() => {
-        writer.close().catch(() => {});
-      });
-    /* eslint-enable @typescript-eslint/no-empty-function */
-    return readable;
+    // Return the real turn stream. useChat reads chunks from this stream to
+    // drive status transitions (submitted → streaming → ready), fire callbacks
+    // (onToolCall, onData, onFinish), and evaluate sendAutomaticallyWhen.
+    //
+    // Both useChat and useMessageSync accumulate messages in parallel:
+    // useChat builds from the stream, useMessageSync pushes from the
+    // transport's message store via setMessages (a full replacement).
+    // The transport's version is always authoritative.
+    return turn.stream;
   },
 
   // Observer mode handles in-progress streams automatically.

test/vercel/codec/encoder.test.ts

@@ -234,14 +234,30 @@ describe('Vercel encoder', () => {
       expect(headersOf(msg)[`${D}messageId`]).toBe('msg-1');
     });
 
-    it('omits messageId domain header when start chunk has no messageId', async () => {
+    it('omits messageId domain header when neither chunk nor options provide it', async () => {
       const encoder = createEncoder(writer);
       await encoder.appendEvent({ type: 'start' });
 
       const msg = firstPublish(writer);
       expect(headersOf(msg)[`${D}messageId`]).toBeUndefined();
     });
 
+    it('falls back to options.messageId when start chunk has no messageId', async () => {
+      const encoder = createEncoder(writer, { messageId: 'fallback-id' });
+      await encoder.appendEvent({ type: 'start' });
+
+      const msg = firstPublish(writer);
+      expect(headersOf(msg)[`${D}messageId`]).toBe('fallback-id');
+    });
+
+    it('prefers chunk.messageId over options.messageId', async () => {
+      const encoder = createEncoder(writer, { messageId: 'fallback-id' });
+      await encoder.appendEvent({ type: 'start', messageId: 'chunk-id' });
+
+      const msg = firstPublish(writer);
+      expect(headersOf(msg)[`${D}messageId`]).toBe('chunk-id');
+    });
+
     it('stamps x-ably-msg-id from WriteOptions on all publishes', async () => {
       const encoder = createEncoder(writer);
       const perWrite = { messageId: 'msg-1' };