WorkflowChatTransport overview and purpose
WorkflowChatTransport is a ChatTransport implementation for useChat that enables automatic stream reconnection for workflow-based chat apps. It posts messages to a chat endpoint, extracts the x-workflow-run-id response header, and reconnects to a /{runId}/stream endpoint on interruption such as network failures, page refreshes, or function timeouts. Unlike DefaultChatTransport which assumes the full response arrives in a single HTTP request, WorkflowChatTransport is designed for the Workflow SDK where the initial response stream may be interrupted by function timeouts. The transport automatically detects missing finish events and reconnects to resume from where the stream left off.
WorkflowChatTransport import statement
Import WorkflowChatTransport from '@ai-sdk/workflow' using: import { WorkflowChatTransport } from '@ai-sdk/workflow';
WorkflowChatTransport constructor parameters
WorkflowChatTransport constructor accepts the following optional parameters: api (string, default '/api/chat') - API endpoint for chat requests; the reconnection endpoint is derived as {api}/{runId}/stream. fetch (typeof fetch, default global fetch) - Custom fetch implementation. maxConsecutiveErrors (number, default 3) - Maximum consecutive errors allowed before giving up. initialStartIndex (number, default 0) - Default chunk index to start from when reconnecting; negative values read from end (e.g., -50 fetches last 50 chunks). onChatSendMessage ((response: Response, options: SendMessagesOptions) => void | Promise<void>) - Callback after initial POST succeeds, useful for extracting workflow run ID or tracking history. onChatEnd (({ chatId, chunkIndex }) => void | Promise<void>) - Callback when stream ends with finish chunk. prepareSendMessagesRequest (PrepareSendMessagesRequest) - Function to customize POST request. prepareReconnectToStreamRequest (PrepareReconnectToStreamRequest) - Function to customize reconnection GET request.
WorkflowChatTransport.sendMessages() method
The sendMessages() method sends messages to the chat endpoint via POST and returns a streaming response. If the stream is interrupted without a finish event, the transport automatically reconnects via GET to {api}/{runId}/stream?startIndex={chunkIndex} to resume from the last received chunk. The method accepts: chatId (string) - unique identifier for the chat session; trigger ('submit-message' | 'regenerate-message') - the type of message submission; messageId (string | undefined) - ID of message to regenerate or undefined for new messages; messages (UIMessage[]) - array of UI messages representing conversation history; abortSignal (AbortSignal | undefined) - signal to abort the request, propagated to both initial POST and reconnection GET. Returns Promise<ReadableStream<UIMessageChunk>> that includes chunks from both initial POST response and any automatic reconnection.
WorkflowChatTransport.reconnectToStream() method
The reconnectToStream() method reconnects to an existing chat stream that was previously interrupted, useful for resuming after page refresh. It accepts: chatId (string) - the chat ID to reconnect to, used to construct the reconnection URL; abortSignal (AbortSignal | undefined) - signal to abort the reconnection request; startIndex (number, optional) - override the start index for this reconnection; negative values read from end of stream, when omitted falls back to the constructor's initialStartIndex. Returns Promise<ReadableStream<UIMessageChunk> | null>.
WorkflowChatTransport reconnection flow
WorkflowChatTransport follows this reconnection flow: 1) POST to {api} with messages; response must include x-workflow-run-id header. 2) Stream the SSE response, counting chunks as they arrive. 3) Detect interruption: if stream closes without finish event (e.g., function timeout, network error), the response is incomplete. 4) Reconnect via GET to {api}/{runId}/stream?startIndex={chunkIndex} to resume from last received chunk. 5) Retry: if reconnection stream also interrupts, retry up to maxConsecutiveErrors times. 6) Complete: once finish event is received, call onChatEnd and close stream.
Negative start index behavior in WorkflowChatTransport
When initialStartIndex is negative (e.g., -50), the transport sends it as-is in the first reconnection request. The server should resolve this to an absolute position and return the x-workflow-stream-tail-index response header so the transport can compute the correct position for subsequent retries. If the header is missing or invalid, the transport falls back to replaying from the beginning (startIndex=0).
WorkflowChatTransport server requirements
WorkflowChatTransport requires two server endpoints: POST {api} (e.g., /api/chat) must accept messages as JSON body, return an SSE stream of UIMessageChunk events, and include an x-workflow-run-id response header. GET {api}/{runId}/stream (e.g., /api/chat/{runId}/stream) must accept a startIndex query parameter, return the SSE stream starting from the given chunk index, and for negative startIndex, resolve to the tail and include x-workflow-stream-tail-index response header.
WorkflowChatTransport basic usage example
Example showing basic usage with useChat:
```tsx
'use client';
import { useChat } from '@ai-sdk/react';
import { WorkflowChatTransport } from '@ai-sdk/workflow';
import { useMemo } from 'react';
export default function Chat() {
const transport = useMemo(
() => new WorkflowChatTransport({ api: '/api/chat' }),
[],
);
const { messages, sendMessage, status } = useChat({ transport });
return (
<div>
{messages.map(message => (
<div key={message.id}>
{message.role === 'user' ? 'User: ' : 'AI: '}
{message.parts.map((part, index) =>
part.type === 'text' ? <span key={index}>{part.text}</span> : null,
)}
</div>
))}
<button onClick={() => sendMessage({ text: 'Hello!' })}>Send</button>
</div>
);
}
```
WorkflowChatTransport with callbacks and page refresh recovery example
Example showing usage with callbacks and page refresh recovery:
```tsx
'use client';
import { useChat } from '@ai-sdk/react';
import { WorkflowChatTransport } from '@ai-sdk/workflow';
import { useMemo } from 'react';
export default function Chat() {
const transport = useMemo(
() =>
new WorkflowChatTransport({
api: '/api/chat',
maxConsecutiveErrors: 5,
initialStartIndex: -50, // Resume from last 50 chunks on page refresh
onChatSendMessage: response => {
const runId = response.headers.get('x-workflow-run-id');
console.log('Workflow run started:', runId);
},
onChatEnd: ({ chatId, chunkIndex }) => {
console.log(`Chat ${chatId} complete, ${chunkIndex} chunks`);
},
}),
[],
);
const { messages, sendMessage } = useChat({ transport });
// ... render chat UI
}
```
WorkflowChatTransport POST endpoint example
Example server-side POST endpoint for Next.js (app/api/chat/route.ts):
```ts
import { createModelCallToUIChunkTransform } from '@ai-sdk/workflow';
import { createUIMessageStreamResponse, type UIMessage } from 'ai';
import { start } from 'workflow/api';
import { chat } from '@/workflow/agent-chat';
export async function POST(request: Request) {
const { messages }: { messages: UIMessage[] } = await request.json();
const run = await start(chat, [messages]);
return createUIMessageStreamResponse({
stream: run.readable.pipeThrough(createModelCallToUIChunkTransform()),
headers: {
'x-workflow-run-id': run.runId,
},
});
}
```
WorkflowChatTransport GET reconnection endpoint example
Example server-side GET reconnection endpoint for Next.js (app/api/chat/[runId]/stream/route.ts):
```ts
import { createModelCallToUIChunkTransform } from '@ai-sdk/workflow';
import type { NextRequest } from 'next/server';
import { getRun } from 'workflow/api';
export async function GET(
request: NextRequest,
{ params }: { params: Promise<{ runId: string }> },
) {
const { runId } = await params;
const startIndex = Number(
new URL(request.url).searchParams.get('startIndex') ?? '0',
);
const run = await getRun(runId);
const readable = run
.getReadable({ startIndex })
.pipeThrough(createModelCallToUIChunkTransform());
return new Response(readable, {
headers: {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
'x-workflow-run-id': runId,
},
});
}
```
AI SDK Workflow overview
AI SDK Workflow enables building durable, resumable AI agents with the @ai-sdk/workflow package.
Phase 2 dual-write implementation pattern
During migration, upsertMessage writes to both schemas in a transaction. First write to v4 schema as before, then convert message using convertV4MessageToV5 and write to v5 schema. Use onConflictDoUpdate for both inserts to handle updates. This ensures new messages are available in both v4 and v5 formats during the migration period.
AI SDK 5.0 data migration overview
AI SDK 5.0 introduces changes to message structure and persistence patterns. Unlike code migrations that can be automated, data migration depends on specific persistence approach, database schema, and application requirements. A two-phase approach is recommended: Phase 1 uses runtime conversion to update the application without database changes, allowing immediate v5 adoption; Phase 2 involves migrating the data schema at your own pace by creating a new v5-compatible schema alongside the existing one.
AI SDK 4.0 vs 5.0 message structure changes
AI SDK 4.0 uses: content field for text, reasoning as top-level property, toolInvocations as top-level property, and parts (optional) ordered array. AI SDK 5.0 uses: parts array as single source of truth, content removed (accessed via text part), reasoning removed and replaced with reasoning part, toolInvocations removed and replaced with tool-${toolName} parts with input/output (renamed from args/result), data role removed (use data parts instead).
Phase 1 runtime conversion layer approach
Phase 1 gets the application working with AI SDK 5.0 without touching the database. Steps include: (1) Update dependencies by installing v4 types alongside v5 using npm aliases (ai@^5.0.0 and ai-legacy npm:ai@^4.3.2), (2) Add conversion functions to transform between v4 and v5 formats, (3) Convert messages when reading from database (v4 to v5), (4) Convert messages when saving (v5 to v4). The database schema remains unchanged. Timeline: can be completed in hours or days.
Phase 2 side-by-side schema migration approach
Phase 2 migrates data to v5-compatible schema after Phase 1 is complete. Steps include: (1) Create messages_v5 table alongside existing messages table, (2) Start dual-writing to both tables with conversion, (3) Run background migration to convert existing messages, (4) Switch reads to v5 schema, (5) Remove conversion from route handlers, (6) Remove dual-write and write only to v5, (7) Drop old tables. This eliminates runtime conversion overhead, eliminates technical debt early, provides type safety with v5 message format, and makes maintenance and extension easier. Timeline: do this soon after Phase 1.
V4 message type guards for conversion
Type guards detect which message format is being used. isV4Message returns true if message contains toolInvocations, has parts with tool-invocation type, has role 'data', has string reasoning field, has parts with args or result fields, or has parts with reasoning and details fields. isV4ToolInvocationPart checks type equals 'tool-invocation' and presence of toolInvocation field. isV4ReasoningPart checks type equals 'reasoning' and presence of reasoning field. isV4SourcePart checks type equals 'source' and presence of source field. isV4FilePart checks type equals 'file', mimeType, and data fields.
V4 to V5 tool invocation state mapping
Tool invocation states map from v4 to v5 as follows: 'partial-call' to 'input-streaming', 'call' to 'input-available', 'result' to 'output-available'. Default fallback is 'output-available'. The function convertToolInvocationState applies this mapping.
V4 to V5 tool invocation conversion
ToolInvocation conversion creates a ToolUIPart with type `tool-${toolName}`, maps toolCallId from toolCallId, maps input from args, maps output from result (only if state is 'result'), and converts state using convertToolInvocationState function.
V4 to V5 message parts conversion strategy
Conversion logic depends on message structure: if message has parts array, convert each part individually using convertPart function; if no parts array, build parts from top-level fields by checking for reasoning field (create reasoning part), toolInvocations field (create tool parts), and content field (create text part). The convertV4MessageToV5 function handles data role messages separately by converting to assistant role and creating a data-custom part.
V5 to V4 message conversion for storage
During Phase 1, v5 messages are converted back to v4 format before storage. For each part: text parts set content field, reasoning parts set reasoning field and create reasoning part with details, tool-${name} parts create toolInvocation entries with state mapping and renamed args/result fields, source-url parts create source parts with id/url/title/sourceType, file parts create file parts with mimeType/data, data-custom parts set data field. The base message includes id, role, content (from text parts), reasoning, toolInvocations, and parts arrays.
Runtime message loading conversion pattern
When loading messages from database during Phase 1, apply conversion using convertV4MessageToV5 for each raw message: rawMessages.map((msg, index) => convertV4MessageToV5(msg, index)). This transforms database v4 format to application v5 format at read time.
Runtime message saving conversion pattern
When saving messages during Phase 1, convert v5 format to v4 inline before passing to database functions: convertV5MessageToV4(message) for user messages and responseMessage in onFinish handler. Keep database functions unchanged to continue working with v4 format. This provides bidirectional conversion: reading v4→v5, writing v5→v4.
Phase 2 background migration script pattern
Migration script: (1) Get all v5 messages to find migrated IDs, (2) Select all v4 messages and filter out migrated ones, (3) Process in batches of 100 messages using transactions, (4) For each message convert using convertV4MessageToV5, (5) Insert into messages_v5 table, (6) Track progress and errors, (7) Can be run multiple times safely and stopped/resumed. Log progress every batch.
Phase 2 verification script pattern
Verification checks data integrity by counting messages in both v4 and v5 schemas using count() from drizzle-orm. Calculates migration progress percentage as (v5Count / v4Count * 100). This ensures all messages have been migrated before switching to v5 schema.
Phase 2 v5 schema reading after migration
After migration is complete and verified, update read functions to query messages_v5 directly. No conversion is needed since data is already in v5 format: return messages from messages_v5 table filtered by chatId and ordered by createdAt. Return type is MyUIMessage[].
Phase 2 v5 schema writing after migration
Once reads use v5 schema and background migration is complete, stop dual-writing and write only to messages_v5. Update upsertMessage to accept MyUIMessage (v5 format) instead of UIMessage (v4 format). Pass message directly without conversion. Update route handlers to pass v5 messages directly without calling convertV5MessageToV4.
Phase 2 completion cleanup steps
After v5 migration is verified and working in production for 1-2 weeks: (1) Remove conversion functions (delete v4↔v5 conversion utilities), (2) Remove ai-legacy dependency (uninstall v4 types package), (3) Test thoroughly to ensure app works with v5 schema, (4) Monitor for production issues, (5) Drop old messages table with DROP TABLE messages, (6) Optionally rename messages_v5 table to messages.
Migration skill for AI SDK v6 to v7
A migration skill is available to help guide migration from AI SDK v6 to v7. It can be added using the command: npx skills add vercel/ai --skill migrate-ai-sdk-v6-to-v7
Stream protocol changed from single chunks to start/delta/end pattern
The streaming pattern has changed in v5. Instead of chunk types like 'text-delta', the new pattern uses: 'text-start' (initialize with unique ID), 'text-delta' (now uses delta property instead of textDelta), 'text-end' (finalize). All chunks now include an ID for tracking content blocks.