Function toServerSentEventsStream

packages/ai/src/stream-to-response.ts:50–100

(
  stream: AsyncIterable<StreamChunk>,
  abortController?: AbortController,
)

 * @returns ReadableStream in Server-Sent Events format
 */
export function toServerSentEventsStream(
  stream: AsyncIterable<StreamChunk>,
  abortController?: AbortController,
): ReadableStream<Uint8Array> {
  const encoder = new TextEncoder()

  return new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of stream) {
          // Check if stream was cancelled/aborted
          if (abortController?.signal.aborted) {
            break
          }

          // Send each chunk as Server-Sent Events format
          controller.enqueue(
            encoder.encode(`data: ${JSON.stringify(chunk)}\n\n`),
          )
        }

        controller.close()
      } catch (error: unknown) {
        // Don't send error if aborted
        if (abortController?.signal.aborted) {
          controller.close()
          return
        }

        // Send error event (AG-UI RUN_ERROR)
        controller.enqueue(
          encoder.encode(
            `data: ${JSON.stringify({
              type: 'RUN_ERROR',
              timestamp: Date.now(),
              error: toRunErrorPayload(error),
            })}\n\n`,
          ),
        )
        controller.close()
      }
    },
    cancel() {
      // When the ReadableStream is cancelled (e.g., client disconnects),
      // abort the underlying stream
      if (abortController) {
        abortController.abort()
      }
    },
  })
}

/**
 * Convert a StreamChunk async iterable to a Response in Server-Sent Events format