Function processStreamChunks

packages/openai-base/src/adapters/responses-text.ts:760–1605

* Processes streamed chunks from the Responses API and yields AG-UI events. * Override this in subclasses to handle provider-specific stream behavior. * * Handles the following event types: * - response.created / response.incomplete / response.failed * - response.output_text.delta

(
    stream: AsyncIterable<ResponseStreamEvent>,
    toolCallMetadata: Map<
      string,
      {
        index: number
        name: string
        started: boolean
        ended?: boolean
        pendingArguments?: string | undefined
      }
    >,
    options: TextOptions<TProviderOptions>,
    aguiState: {
      runId: string
      threadId: string
      messageId: string
      hasEmittedRunStarted: boolean
    },
  )

   * - error
   */
  protected async *processStreamChunks(
    stream: AsyncIterable<ResponseStreamEvent>,
    toolCallMetadata: Map<
      string,
      {
        index: number
        name: string
        started: boolean
        ended?: boolean
        pendingArguments?: string | undefined
      }
    >,
    options: TextOptions<TProviderOptions>,
    aguiState: {
      runId: string
      threadId: string
      messageId: string
      hasEmittedRunStarted: boolean
    },
  ): AsyncIterable<StreamChunk> {
    let accumulatedContent = ''
    let accumulatedReasoning = ''

    // Track if we've been streaming deltas to avoid duplicating content from done events
    let hasStreamedContentDeltas = false
    let hasStreamedReasoningDeltas = false

    // Preserve response metadata across events
    let model: string = options.model

    // AG-UI lifecycle tracking
    let stepId: string | null = null
    let hasEmittedTextMessageStart = false
    let hasEmittedStepStarted = false
    // Track whether we've emitted a terminal RUN_FINISHED so the
    // end-of-stream fallback below knows to synthesise one when the upstream
    // cuts off without a response.completed event.
    let runFinishedEmitted = false

    try {
      for await (const chunk of stream) {
        options.logger.provider(`provider=${this.name} type=${chunk.type}`, {
          provider: this.name,
          type: chunk.type,
        })

        // Emit RUN_STARTED on first chunk
        if (!aguiState.hasEmittedRunStarted) {
          aguiState.hasEmittedRunStarted = true
          yield {
            type: EventType.RUN_STARTED,
            runId: aguiState.runId,
            threadId: aguiState.threadId,
            model: model || options.model,
            timestamp: Date.now(),
            parentRunId: options.parentRunId,
          }
        }