Function createReadableStreamFromResponses

apps/sim/providers/openai/utils.ts:416–537

(
  response: Response,
  onComplete?: (content: string, usage?: ResponsesUsageTokens) => void
)

 * Creates a ReadableStream from a Responses API SSE stream.
 */
export function createReadableStreamFromResponses(
  response: Response,
  onComplete?: (content: string, usage?: ResponsesUsageTokens) => void
): ReadableStream<Uint8Array> {
  let fullContent = ''
  let finalUsage: ResponsesUsageTokens | undefined
  let activeEventType: string | undefined
  const encoder = new TextEncoder()

  return new ReadableStream<Uint8Array>({
    async start(controller) {
      const reader = response.body?.getReader()
      if (!reader) {
        controller.close()
        return
      }

      const decoder = new TextDecoder()
      let buffer = ''

      try {
        while (true) {
          const { done, value } = await reader.read()
          if (done) {
            break
          }

          buffer += decoder.decode(value, { stream: true })
          const lines = buffer.split('\n')
          buffer = lines.pop() || ''

          for (const line of lines) {
            const trimmed = line.trim()
            if (!trimmed) {
              continue
            }

            if (trimmed.startsWith('event:')) {
              activeEventType = trimmed.slice(6).trim()
              continue
            }

            if (!trimmed.startsWith('data:')) {
              continue
            }

            const data = trimmed.slice(5).trim()
            if (data === '[DONE]') {
              continue
            }

            let event: Record<string, unknown>
            try {
              event = JSON.parse(data)
            } catch (error) {
              logger.debug('Skipping non-JSON response stream chunk', {
                data: data.slice(0, 200),
                error,