prepareReconnectToStreamRequest?: PrepareReconnectToStreamRequest } /** * A transport implementation for managing chat workflows with support for * streaming responses and automatic reconnection to interrupted streams. * * This class implements the ChatTransport interface from the AI SDK and provides * reliable message streaming with automatic recovery from network interruptions * or function timeouts. * * @template UI_MESSAGE - The type of UI messages being sent and received, * must extend the UIMessage interface from the AI SDK. * * @implements {ChatTransport} */ export class WorkflowChatTransport implements ChatTransport { private readonly api: string private readonly fetch: typeof fetch private readonly onChatSendMessage?: OnChatSendMessage private readonly onChatEnd?: OnChatEnd private readonly maxConsecutiveErrors: number private readonly prepareSendMessagesRequest?: PrepareSendMessagesRequest private readonly prepareReconnectToStreamRequest?: PrepareReconnectToStreamRequest /** * Creates a new WorkflowChatTransport instance. * * @param options - Configuration options for the transport * @param options.api - API endpoint for chat requests (defaults to '/api/chat') * @param options.fetch - Custom fetch implementation (defaults to global fetch) * @param options.onChatSendMessage - Callback after sending messages * @param options.onChatEnd - Callback when chat stream ends * @param options.maxConsecutiveErrors - Maximum consecutive errors for reconnection * @param options.prepareSendMessagesRequest - Function to prepare send messages request * @param options.prepareReconnectToStreamRequest - Function to prepare reconnect request */ constructor(options: WorkflowChatTransportOptions = {}) { this.api = options.api ?? "/api/chat" this.fetch = options.fetch ?? fetch.bind(globalThis) this.onChatSendMessage = options.onChatSendMessage this.onChatEnd = options.onChatEnd this.maxConsecutiveErrors = options.maxConsecutiveErrors ?? 3 this.prepareSendMessagesRequest = options.prepareSendMessagesRequest this.prepareReconnectToStreamRequest = options.prepareReconnectToStreamRequest } /** * Sends messages to the chat endpoint and returns a stream of response chunks. * * This method handles the entire chat lifecycle including: * - Sending messages to the /api/chat endpoint * - Streaming response chunks * - Automatic reconnection if the stream is interrupted * * @param options - Options for sending messages * @param options.trigger - The type of message submission ('submit-message' or 'regenerate-message') * @param options.chatId - Unique identifier for this chat session * @param options.messageId - Optional message ID for tracking specific messages * @param options.messages - Array of UI messages to send * @param options.abortSignal - Optional AbortSignal to cancel the request * * @returns A ReadableStream of UIMessageChunk objects containing the response * @throws Error if the fetch request fails or returns a non-OK status */ async sendMessages( options: SendMessagesOptions & ChatRequestOptions ): Promise> { return iteratorToStream(this.sendMessagesIterator(options), { signal: options.abortSignal, }) } private async *sendMessagesIterator( options: SendMessagesOptions ): AsyncGenerator { const { chatId, messages, abortSignal, trigger, messageId } = options // We keep track of if the "finish" chunk is received to determine // if we need to reconnect, and keep track of the chunk index to resume from. let gotFinish = false let chunkIndex = 0 // Prepare the request using the configurator if provided const requestConfig = this.prepareSendMessagesRequest ? await this.prepareSendMessagesRequest({ id: chatId, messages, requestMetadata: undefined, body: undefined, credentials: undefined, headers: undefined, api: this.api, trigger, messageId, }) : undefined