import {
  getErrorMessage,
  UnsupportedFunctionalityError,
  type LanguageModelV4,
  type SharedV4Warning,
} from '@ai-sdk/provider';
import {
  asArray,
  createIdGenerator,
  DelayedPromise,
  filterNullable,
  isAbortError,
  type Arrayable,
  type Context,
  type Experimental_SandboxSession as SandboxSession,
  type IdGenerator,
  type InferToolSetContext,
  type ModelMessage,
  type ProviderOptions,
  type ToolApprovalResponse,
  type ToolContent,
  type ToolSet,
} from '@ai-sdk/provider-utils';
import type { ServerResponse } from 'node:http';
import { NoOutputGeneratedError, ToolChoiceViolationError } from '../error';
import { logWarnings } from '../logger/log-warnings';
import { resolveLanguageModel } from '../model/resolve-model';
import { cloneModelMessages } from '../prompt/clone-model-message';
import { createToolModelOutput } from '../prompt/create-tool-model-output';
import type { LanguageModelCallOptions } from '../prompt/language-model-call-options';
import { prepareLanguageModelCallOptions } from '../prompt/prepare-language-model-call-options';
import { prepareToolChoice } from '../prompt/prepare-tool-choice';
import { prepareTools } from '../prompt/prepare-tools';
import type { Prompt } from '../prompt/prompt';
import {
  getChunkTimeoutMs,
  getFirstChunkTimeoutMs,
  getStepTimeoutMs,
  getTotalTimeoutMs,
  type RequestOptions,
  type TimeoutConfiguration,
} from '../prompt/request-options';
import { standardizePrompt } from '../prompt/standardize-prompt';
import { wrapGatewayError } from '../prompt/wrap-gateway-error';
import type { TelemetryDispatcher } from '../telemetry/telemetry';
import type { TelemetryOptions } from '../telemetry/telemetry-options';
import { createTextStreamResponse } from '../text-stream/create-text-stream-response';
import { pipeTextStreamToResponse } from '../text-stream/pipe-text-stream-to-response';
import { toTextStream } from '../text-stream/to-text-stream';
import type { LanguageModelRequestMetadata } from '../types';
import type {
  CallWarning,
  FinishReason,
  LanguageModel,
  ToolChoice,
} from '../types/language-model';
import type { ProviderMetadata } from '../types/provider-metadata';
import {
  addLanguageModelUsage,
  createNullLanguageModelUsage,
  type LanguageModelUsage,
} from '../types/usage';
import type { UIMessage } from '../ui';
import { createUIMessageStreamResponse } from '../ui-message-stream/create-ui-message-stream-response';
import { pipeUIMessageStreamToResponse } from '../ui-message-stream/pipe-ui-message-stream-to-response';
import { toUIMessageStream as toUIMessageStreamHelper } from '../ui-message-stream/to-ui-message-stream';
import type { InferUIMessageChunk } from '../ui-message-stream/ui-message-chunks';
import type { UIMessageStreamResponseInit } from '../ui-message-stream/ui-message-stream-response-init';
import {
  createAsyncIterableStream,
  type AsyncIterableStream,
} from '../util/async-iterable-stream';
import type { Callback } from '../util/callback';
import { consumeStream } from '../util/consume-stream';
import { createIdMap } from '../util/create-id-map';
import { createStitchableStream } from '../util/create-stitchable-stream';
import type { DownloadFunction } from '../util/download/download-function';
import { getOwn } from '../util/get-own';
import { mergeAbortSignals } from '../util/merge-abort-signals';
import { mergeObjects } from '../util/merge-objects';
import { notify } from '../util/notify';
import { now as originalNow } from '../util/now';
import { prepareRetries } from '../util/prepare-retries';
import { setAbortTimeout } from '../util/set-abort-timeout';
import type { ActiveTools } from './active-tools';
import { collectToolApprovals } from './collect-tool-approvals';
import type { ContentPart } from './content-part';
import {
  executeToolsFromStream,
  type ExecuteToolsStreamPart,
} from './execute-tools-from-stream';
import { createToolSearchState } from '../tool-search/prepare-tool-search';
import { executeToolCall } from './execute-tool-call';
import {
  filterActiveTools,
  type ActiveToolSubset,
} from './filter-active-tools';
import type {
  GenerateTextAbortEvent,
  GenerateTextEndEvent,
  GenerateTextOnStartCallback,
  GenerateTextOnStepEndCallback,
  GenerateTextOnStepFinishCallback,
  GenerateTextOnStepStartCallback,
} from './generate-text-events';
import { invokeToolCallbacksFromStream } from './invoke-tool-callbacks-from-stream';
import type {
  OnLanguageModelCallEndCallback,
  OnLanguageModelCallStartCallback,
} from './language-model-events';
import { text, type Output } from './output';
import type {
  InferCompleteOutput,
  InferElementOutput,
  InferPartialOutput,
} from './output-utils';
import type { PrepareStepFunction } from './prepare-step';
import { prepareStepCallSettings } from './prepare-step-call-settings';
import { convertToReasoningOutputs } from './reasoning-output';
import type { ResponseMessage } from './response-message';
import { createRestrictedTelemetryDispatcher } from './restricted-telemetry-dispatcher';
import {
  DefaultStepResult,
  type StepResult,
  type StepResultPerformance,
} from './step-result';
import {
  isStepCount,
  isStopConditionMet,
  type StopCondition,
} from './stop-condition';
import {
  streamLanguageModelCall,
  type LanguageModelStreamPart,
} from './stream-language-model-call';
import {
  createStreamRetryAttemptBoundaryPart,
  isStreamRetryAttemptBoundaryPart,
  type StreamRetryAttemptBoundaryPart,
} from './stream-retry-attempt-boundary';
import type {
  ConsumeStreamOptions,
  StreamTextResult,
  TextStreamPart,
  UIMessageStreamOptions,
} from './stream-text-result';
import { toResponseMessages } from './to-response-messages';
import type { ToolApprovalConfiguration } from './tool-approval-configuration';
import {
  appendToolCallerMessages,
  prepareToolsForToolCallers,
  resolveToolCallerConfiguration,
  type Experimental_ToolCallers,
} from './tool-caller-configuration';
import type { TypedToolCall } from './tool-call';
import type { ToolCallRepairFunction } from './tool-call-repair-function';
import type {
  OnToolExecutionEndCallback,
  OnToolExecutionStartCallback,
} from './tool-execution-events';
import type { ToolInputRefinement } from './tool-input-refinement';
import type { ToolOrder } from './tool-order';
import type { ToolOutput } from './tool-output';
import type { StaticToolOutputDenied } from './tool-output-denied';
import type { ToolsContextParameter } from './tools-context-parameter';
import { validateApprovedToolApprovals } from './validate-tool-approvals';

const originalGenerateId = createIdGenerator({
  prefix: 'aitxt',
  size: 24,
});

const originalGenerateCallId = createIdGenerator({
  prefix: 'call',
  size: 24,
});

// Chunk types that contain semantic model output. This classification is used
// for first-content and inter-content timeouts as well as to distinguish empty
// incomplete streams from incomplete streams with partial results. It is
// exhaustive so that new chunk types must be classified explicitly.
const isOutputChunkType = {
  file: true,
  custom: false,
  source: false,
  'text-start': false,
  'text-end': false,
  'text-delta': true,
  'reasoning-start': false,
  'reasoning-end': false,
  'reasoning-delta': true,
  'reasoning-file': true,
  'tool-input-start': false,
  'tool-input-end': false,
  'tool-input-delta': true,
  'tool-approval-request': false,
  'tool-approval-response': false,
  'tool-call': true,
  'tool-result': false,
  'tool-error': false,
  'tool-output-denied': false,
  'tool-execution-end': false,
  'model-call-start': false,
  'model-call-response-metadata': false,
  'model-call-end': false,
  error: false,
  raw: false,
} as const satisfies Record<
  Exclude<ExecuteToolsStreamPart, StreamRetryAttemptBoundaryPart>['type'],
  boolean
>;

function isOutputChunk(
  chunk: Exclude<ExecuteToolsStreamPart, StreamRetryAttemptBoundaryPart>,
): boolean {
  if (!isOutputChunkType[chunk.type]) {
    return false;
  }

  switch (chunk.type) {
    case 'text-delta':
    case 'reasoning-delta':
      return chunk.text.length > 0;
    case 'tool-input-delta':
      return chunk.delta.length > 0;
    case 'file':
    case 'reasoning-file':
    case 'tool-call':
      return true;
    default:
      return false;
  }
}

export type StreamTextInclude = {
  /**
   * Whether to retain the request body in step results.
   * The request body can be large when sending images or files.
   *
   * @default false
   */
  requestBody?: boolean;

  /**
   * Whether to retain the request messages in step results.
   * The request messages can be large when sending images or files.
   *
   * @default false
   */
  requestMessages?: boolean;

  /**
   * Whether to include raw chunks from the provider in the stream.
   *
   * When enabled, you will receive raw chunks with type 'raw' that contain
   * the unprocessed data from the provider.
   *
   * This allows access to cutting-edge provider features not yet wrapped by
   * the AI SDK.
   *
   * @default false
   */
  rawChunks?: boolean;
};

/**
 * A transformation that is applied to the stream.
 *
 * @param stopStream - A function that stops the source stream.
 * @param tools - The tools that are accessible to and can be called by the model. The model needs to support calling tools.
 */
export type StreamTextTransform<TOOLS extends ToolSet> = (options: {
  tools: TOOLS; // for type inference
  stopStream: () => void;
}) => TransformStream<TextStreamPart<TOOLS>, TextStreamPart<TOOLS>>;

/** A result that requests recovery from a streamed provider error. */
export type StreamTextOnErrorResult = { retry: true };

/**
 * Existing observer callback that is set using the `onError` option.
 *
 * @param event - The event that is passed to the callback.
 */
export type StreamTextOnErrorCallback = Callback<{ error: unknown }>;

/**
 * Retry-capable callback that is set using the `onError` option.
 *
 * @param event - The event that is passed to the callback.
 */
export type StreamTextOnErrorRetryCallback = (event: {
  error: unknown;
}) =>
  | PromiseLike<void | StreamTextOnErrorResult>
  | void
  | StreamTextOnErrorResult;

type StreamTextOnErrorHandler = (event: { error: unknown }) => void;

/**
 * Callback that is set using the `onChunk` option.
 *
 * @param event - The event that is passed to the callback.
 */
export type StreamTextOnChunkCallback<TOOLS extends ToolSet> = (event: {
  chunk: TextStreamPart<TOOLS>;
}) => PromiseLike<void> | void;

export type StreamTextEndEvent<
  TOOLS extends ToolSet = ToolSet,
  RUNTIME_CONTEXT extends Context = Context,
  OUTPUT extends Output = Output,
> = GenerateTextEndEvent<TOOLS, RUNTIME_CONTEXT> & {
  /**
   * The parsed output when an output setting was provided and parsing
   * succeeded.
   */
  readonly output?: InferCompleteOutput<OUTPUT>;
};

export type StreamTextOnEndCallback<
  TOOLS extends ToolSet = ToolSet,
  RUNTIME_CONTEXT extends Context = Context,
  OUTPUT extends Output = Output,
> = Callback<StreamTextEndEvent<TOOLS, RUNTIME_CONTEXT, OUTPUT>>;

/**
 * Callback that is set using the `onAbort` option.
 *
 * @param event - The event that is passed to the callback.
 */
export type StreamTextOnAbortCallback<
  TOOLS extends ToolSet,
  RUNTIME_CONTEXT extends Context,
> = Callback<GenerateTextAbortEvent<TOOLS, RUNTIME_CONTEXT>>;

/**
 * Generate a text and call tools for a given prompt using a language model.
 *
 * This function streams the output. If you do not want to stream the output, use `generateText` instead.
 *
 * @param model - The language model to use.
 * @param tools - Tools that are accessible to and can be called by the model. The model needs to support calling tools.
 * @param toolOrder - Controls the order in which tools are sent to the provider. Tools not listed are appended alphabetically.
 *
 * @param system - A system message that will be part of the prompt.
 * @param prompt - A simple text prompt. You can either use `prompt` or `messages` but not both.
 * @param messages - A list of messages. You can either use `prompt` or `messages` but not both.
 * @param allowSystemInMessages - Whether system messages are allowed in the `prompt` or `messages` fields. Default: false.
 *
 * @param maxOutputTokens - Maximum number of tokens to generate.
 * @param temperature - Temperature setting.
 * The value is passed through to the provider. The range depends on the provider and model.
 * It is recommended to set either `temperature` or `topP`, but not both.
 * @param topP - Nucleus sampling.
 * The value is passed through to the provider. The range depends on the provider and model.
 * It is recommended to set either `temperature` or `topP`, but not both.
 * @param topK - Only sample from the top K options for each subsequent token.
 * Used to remove "long tail" low probability responses.
 * Recommended for advanced use cases only. You usually only need to use temperature.
 * @param presencePenalty - Presence penalty setting.
 * It affects the likelihood of the model to repeat information that is already in the prompt.
 * The value is passed through to the provider. The range depends on the provider and model.
 * @param frequencyPenalty - Frequency penalty setting.
 * It affects the likelihood of the model to repeatedly use the same words or phrases.
 * The value is passed through to the provider. The range depends on the provider and model.
 * @param stopSequences - Stop sequences.
 * If set, the model will stop generating text when one of the stop sequences is generated.
 * @param seed - The seed (integer) to use for random sampling.
 * If set and supported by the model, calls will generate deterministic results.
 *
 * @param maxRetries - Maximum number of retries. Set to 0 to disable retries. Default: 2.
 * @param streamRetries - Maximum number of retries for provider errors received after streaming starts. Set to 0 to disable automatic stream retries while allowing `onError` to request retries. Omit to disable all stream retry behavior. Default: 0.
 * @param abortSignal - An optional abort signal that can be used to cancel the call.
 * @param timeout - An optional timeout in milliseconds. The call will be aborted if it takes longer than the specified timeout.
 * @param headers - Additional HTTP headers to be sent with the request. Only applicable for HTTP-based providers.
 *
 * @param experimental_sandbox - The sandbox environment that is passed through to tool execution.
 * @param runtimeContext - User-defined runtime context that flows through the entire generation lifecycle.
 * @param experimental_refineToolInput - Optional mapping of tool names to functions that refine parsed tool inputs before tools are executed and before outputs, callbacks, and telemetry are recorded.
 *
 * @param onChunk - Callback that is called for each chunk of the stream. The stream processing will pause until the callback promise is resolved.
 * @param onError - Callback that is called when an error occurs during streaming. You can use it to log errors.
 * @param onStart - Callback invoked when generation begins, before any LLM calls.
 * @param experimental_onStart - Deprecated alias for `onStart`.
 * @param onStepStart - Callback invoked when each step begins, before the provider is called.
 * @param experimental_onStepStart - Deprecated alias for `onStepStart`.
 * @param onLanguageModelCallStart - Callback invoked immediately before each provider model call begins.
 * @param experimental_onLanguageModelCallStart - Deprecated alias for `onLanguageModelCallStart`.
 * @param onLanguageModelCallEnd - Callback invoked after each provider model call response is normalized and parsed.
 * @param experimental_onLanguageModelCallEnd - Deprecated alias for `onLanguageModelCallEnd`.
 * @param onToolExecutionStart - Callback invoked before each tool execution begins.
 * @param experimental_onToolCallStart - Deprecated alias for `onToolExecutionStart`.
 * @param onToolExecutionEnd - Callback invoked after each tool execution completes.
 * @param experimental_onToolCallFinish - Deprecated alias for `onToolExecutionEnd`.
 * @param onStepEnd - Callback that is called when each step (LLM call) ends, including intermediate steps.
 * @param onStepFinish - Deprecated alias for `onStepEnd`.
 * @param onEnd - Callback that is called when all steps are finished and the response is complete.
 * @param onFinish - Deprecated alias for `onEnd`.
 *
 * @returns
 * A result object for accessing different stream types and additional information.
 */
export function streamText<
  TOOLS extends ToolSet,
  RUNTIME_CONTEXT extends Context = Context,
  OUTPUT extends Output = Output<string, string, never>,
>({
  model,
  tools,
  toolChoice,
  instructions,
  system,
  prompt,
  messages,
  allowSystemInMessages,
  maxRetries,
  streamRetries,
  abortSignal,
  timeout,
  headers,
  stopWhen = isStepCount(1),
  experimental_sandbox: sandbox,
  output,
  toolApproval,
  experimental_toolCallers,
  experimental_toolApprovalSecret,
  experimental_telemetry,
  telemetry = experimental_telemetry,
  prepareStep,
  providerOptions,
  activeTools,
  toolOrder,
  experimental_repairToolCall,
  repairToolCall = experimental_repairToolCall,
  experimental_refineToolInput: refineToolInput,
  experimental_transform: transform,
  experimental_download: download,
  includeRawChunks,
  onChunk,
  onError: onErrorArg,
  onFinish,
  onEnd = onFinish,
  onAbort,
  onStepEnd,
  onStepFinish,
  onStart,
  experimental_onStart,
  onStepStart,
  experimental_onStepStart,
  onLanguageModelCallStart,
  experimental_onLanguageModelCallStart,
  onLanguageModelCallEnd,
  experimental_onLanguageModelCallEnd,
  onToolExecutionStart,
  onToolExecutionEnd,
  experimental_onToolCallStart,
  experimental_onToolCallFinish,
  runtimeContext = {} as RUNTIME_CONTEXT,
  toolsContext = {} as InferToolSetContext<TOOLS>,
  experimental_include,
  include = experimental_include,
  _internal: {
    now = originalNow,
    generateId = originalGenerateId,
    generateCallId = originalGenerateCallId,
  } = {},
  ...settings
}: LanguageModelCallOptions &
  RequestOptions<TOOLS> &
  Prompt &
  ToolsContextParameter<TOOLS> & {
    /**
     * The language model to use.
     */
    model: LanguageModel;

    /**
     * The tool choice strategy. Default: 'auto'.
     */
    toolChoice?: ToolChoice<TOOLS>;

    /**
     * Condition for stopping the generation when there are tool results in the last step.
     * When the condition is an array, any of the conditions can be met to stop the generation.
     *
     * @default isStepCount(1)
     */
    stopWhen?: Arrayable<StopCondition<NoInfer<TOOLS>, RUNTIME_CONTEXT>>;

    /**
     * Optional telemetry configuration.
     */
    telemetry?: TelemetryOptions<RUNTIME_CONTEXT, NoInfer<TOOLS>>;

    /**
     * Optional telemetry configuration.
     *
     * @deprecated Use `telemetry` instead. This alias will be removed in a future major release.
     */
    experimental_telemetry?: TelemetryOptions<RUNTIME_CONTEXT, NoInfer<TOOLS>>;

    /**
     * Additional provider-specific options. They are passed through
     * to the provider from the AI SDK and enable provider-specific
     * functionality that can be fully encapsulated in the provider.
     */
    providerOptions?: ProviderOptions;

    /**
     * The sandbox environment that is passed through to tool execution.
     */
    experimental_sandbox?: SandboxSession;

    /**
     * Runtime context. Treat runtime context as immutable.
     * If you need to mutate runtime context, update it in `prepareStep`.
     */
    runtimeContext?: RUNTIME_CONTEXT;

    /**
     * Limits the tools that are available for the model to call without
     * changing the tool call and result types in the result.
     */
    activeTools?: ActiveTools<NoInfer<TOOLS>>;

    /**
     * Controls the order in which tools are sent to the provider.
     *
     * The list can be partial. Tools not listed in `toolOrder` are sent after
     * the listed tools, sorted alphabetically. This can improve provider-side
     * caching by keeping tool definitions in a stable order.
     */
    toolOrder?: ToolOrder<NoInfer<TOOLS>>;

    /**
     * Optional specification for parsing structured outputs from the LLM response.
     */
    output?: OUTPUT;

    /**
     * Optional tool approval configuration.
     *
     * This configuration takes precedence over tool-defined approval settings.
     */
    toolApproval?: ToolApprovalConfiguration<TOOLS, RUNTIME_CONTEXT>;

    /**
     * Configures which caller tools may invoke each tool.
     */
    experimental_toolCallers?: Experimental_ToolCallers<NoInfer<TOOLS>>;

    /**
     * Secret for HMAC-signing tool approval requests. When set, the server
     * signs each approval request at issuance and verifies the signature when
     * the approval is replayed, preventing client-forged approvals.
     */
    experimental_toolApprovalSecret?: string | Uint8Array;

    /**
     * Optional function that you can use to provide different settings for a step.
     *
     * @param options - The options for the step.
     * @param options.steps - The steps that have been executed so far.
     * @param options.stepNumber - The number of the step that is being executed.
     * @param options.model - The model that is being used.
     *
     * @returns An object that contains the settings for the step.
     * If you return undefined (or for undefined settings), the settings from the outer level will be used.
     */
    prepareStep?: PrepareStepFunction<NoInfer<TOOLS>, RUNTIME_CONTEXT>;

    /**
     * A function that attempts to repair a tool call that failed to parse.
     */
    repairToolCall?: ToolCallRepairFunction<TOOLS>;

    /**
     * A function that attempts to repair a tool call that failed to parse.
     *
     * @deprecated Use `repairToolCall` instead.
     */
    experimental_repairToolCall?: ToolCallRepairFunction<TOOLS>;

    /**
     * Optional mapping of tool names to functions that refine parsed tool inputs.
     *
     * The refined input must have the same type shape as the tool input. Refined
     * inputs are used for tool execution, stream parts, callbacks, and telemetry.
     */
    experimental_refineToolInput?: ToolInputRefinement<NoInfer<TOOLS>>;

    /**
     * Optional stream transformations.
     * They are applied in the order they are provided.
     * The stream transformations must maintain the stream structure for streamText to work correctly.
     */
    experimental_transform?: Arrayable<StreamTextTransform<TOOLS>>;

    /**
     * Custom download function to use for URLs.
     *
     * By default, files are downloaded if the model does not support the URL for the given media type.
     */
    experimental_download?: DownloadFunction | undefined;

    /**
     * Whether to include raw chunks from the provider in the stream.
     * When enabled, you will receive raw chunks with type 'raw' that contain the unprocessed data from the provider.
     * This allows access to cutting-edge provider features not yet wrapped by the AI SDK.
     * Defaults to false.
     *
     * @deprecated Use `include.rawChunks` instead.
     */
    includeRawChunks?: boolean;

    /**
     * Callback that is called for each chunk of the stream.
     * The stream processing will pause until the callback promise is resolved.
     */
    onChunk?: StreamTextOnChunkCallback<TOOLS>;

    /**
     * Callback that is invoked when an error occurs during streaming.
     * You can use it to log errors.
     * Return `{ retry: true }` to retry the current model step after a provider
     * error is received from the response stream when `streamRetries` is
     * explicitly configured.
     * The stream processing will pause until the callback promise is resolved.
     */
    onError?:
      | StreamTextOnErrorCallback
      | StreamTextOnErrorRetryCallback
      | StreamTextOnErrorHandler;

    /**
     * Maximum number of automatic retries for provider errors received after
     * response streaming has started. Each retry reruns only the current model
     * step. Completed earlier steps and their tool results are preserved.
     *
     * Partial output from a failed attempt that was already emitted cannot be
     * retracted and remains in consumer-facing streams. It is excluded from
     * the recovered step result, structured output parsing, response messages,
     * and subsequent model steps.
     *
     * Set to `0` to disable automatic retries while allowing `onError` to
     * request one retry. When automatic retries are configured, `onError` can
     * request at most one additional retry after they are exhausted. Omit this
     * option to disable all stream retry behavior and preserve incremental tool
     * streaming for existing `onError` observers.
     *
     * @default 0 (stream retry behavior disabled when omitted)
     */
    streamRetries?: number;

    /**
     * Callback that is called when the LLM response and all request tool executions
     * (for tools that have an `execute` function) are finished.
     *
     * The usage is the combined usage of all steps.
     */
    onEnd?: StreamTextOnEndCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>,
      NoInfer<OUTPUT>
    >;

    /**
     * Callback that is called when the LLM response and all request tool executions
     * (for tools that have an `execute` function) are finished.
     *
     * The usage is the combined usage of all steps.
     *
     * @deprecated Use `onEnd` instead.
     */
    onFinish?: StreamTextOnEndCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>,
      NoInfer<OUTPUT>
    >;

    onAbort?: StreamTextOnAbortCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>
    >;

    /**
     * Callback that is called when each step (LLM call) ends, including intermediate steps.
     */
    onStepEnd?: GenerateTextOnStepEndCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>
    >;

    /**
     * Callback that is called when each step (LLM call) ends, including intermediate steps.
     *
     * @deprecated Use `onStepEnd` instead.
     */
    onStepFinish?: GenerateTextOnStepFinishCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>
    >;

    /**
     * Callback that is called when the streamText operation begins,
     * before any LLM calls are made.
     */
    onStart?: GenerateTextOnStartCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>,
      NoInfer<OUTPUT>
    >;

    /**
     * Callback that is called when the streamText operation begins,
     * before any LLM calls are made.
     *
     * @deprecated Use `onStart` instead.
     */
    experimental_onStart?: GenerateTextOnStartCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>,
      NoInfer<OUTPUT>
    >;

    /**
     * Callback that is called when a step (LLM call) begins,
     * before the provider is called.
     */
    onStepStart?: GenerateTextOnStepStartCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>,
      NoInfer<OUTPUT>
    >;

    /**
     * Callback that is called when a step (LLM call) begins,
     * before the provider is called.
     *
     * @deprecated Use `onStepStart` instead.
     */
    experimental_onStepStart?: GenerateTextOnStepStartCallback<
      NoInfer<TOOLS>,
      NoInfer<RUNTIME_CONTEXT>,
      NoInfer<OUTPUT>
    >;

    /**
     * Callback that is called immediately before the provider model call begins.
     */
    onLanguageModelCallStart?: OnLanguageModelCallStartCallback;

    /**
     * Callback that is called immediately before the provider model call begins.
     *
     * @deprecated Use `onLanguageModelCallStart` instead.
     */
    experimental_onLanguageModelCallStart?: OnLanguageModelCallStartCallback;

    /**
     * Callback that is called after the model response has been normalized and parsed,
     * but before any client-side tool execution begins.
     */
    onLanguageModelCallEnd?: OnLanguageModelCallEndCallback<NoInfer<TOOLS>>;

    /**
     * Callback that is called after the model response has been normalized and parsed,
     * but before any client-side tool execution begins.
     *
     * @deprecated Use `onLanguageModelCallEnd` instead.
     */
    experimental_onLanguageModelCallEnd?: OnLanguageModelCallEndCallback<
      NoInfer<TOOLS>
    >;

    /**
     * Callback that is called right before a tool's execute function runs.
     */
    onToolExecutionStart?: OnToolExecutionStartCallback<NoInfer<TOOLS>>;

    /**
     * Callback that is called right before a tool's execute function runs.
     *
     * @deprecated Use `onToolExecutionStart` instead.
     */
    experimental_onToolCallStart?: OnToolExecutionStartCallback<NoInfer<TOOLS>>;

    /**
     * Callback that is called right after a tool's execute function completes (or errors).
     */
    onToolExecutionEnd?: OnToolExecutionEndCallback<NoInfer<TOOLS>>;

    /**
     * Callback that is called right after a tool's execute function completes (or errors).
     *
     * @deprecated Use `onToolExecutionEnd` instead.
     */
    experimental_onToolCallFinish?: OnToolExecutionEndCallback<NoInfer<TOOLS>>;

    /**
     * Settings for controlling what data is included in step results.
     * Disabling inclusion can help reduce memory usage when processing
     * large payloads like images.
     *
     * By default, request bodies and request messages are excluded.
     */
    include?: StreamTextInclude;

    /**
     * Settings for controlling what data is included in step results.
     *
     * @deprecated Use `include` instead.
     */
    experimental_include?: StreamTextInclude;

    /**
     * Internal. For test use only. May change without notice.
     */
    _internal?: {
      now?: () => number;
      generateId?: IdGenerator;
      generateCallId?: IdGenerator;
    };
  }): StreamTextResult<TOOLS, RUNTIME_CONTEXT, OUTPUT> {
  const totalTimeoutMs = getTotalTimeoutMs(timeout);
  const stepTimeoutMs = getStepTimeoutMs(timeout);
  const firstChunkTimeoutMs = getFirstChunkTimeoutMs(timeout);
  const chunkTimeoutMs = getChunkTimeoutMs(timeout);
  const stepAbortController =
    stepTimeoutMs != null ? new AbortController() : undefined;
  const firstChunkAbortController =
    firstChunkTimeoutMs != null ? new AbortController() : undefined;
  const chunkAbortController =
    chunkTimeoutMs != null ? new AbortController() : undefined;
  const onError: StreamTextOnErrorHandler =
    onErrorArg ??
    (({ error }) => {
      console.error(error);
    });
  const resolvedOnStart = onStart ?? experimental_onStart;
  const resolvedOnStepStart = onStepStart ?? experimental_onStepStart;
  const resolvedOnLanguageModelCallStart =
    onLanguageModelCallStart ?? experimental_onLanguageModelCallStart;
  const resolvedOnLanguageModelCallEnd =
    onLanguageModelCallEnd ?? experimental_onLanguageModelCallEnd;
  const resolvedOnToolExecutionStart =
    onToolExecutionStart ?? experimental_onToolCallStart;
  const resolvedOnToolExecutionEnd =
    onToolExecutionEnd ?? experimental_onToolCallFinish;
  const resolvedOnStepEnd = onStepEnd ?? onStepFinish;
  return new DefaultStreamTextResult<TOOLS, RUNTIME_CONTEXT, OUTPUT>({
    model: resolveLanguageModel(model),
    telemetry,
    headers,
    settings,
    maxRetries,
    streamRetries,
    abortSignal: mergeAbortSignals(
      abortSignal,
      totalTimeoutMs,
      stepAbortController?.signal,
      firstChunkAbortController?.signal,
      chunkAbortController?.signal,
    ),
    stepTimeoutMs,
    stepAbortController,
    firstChunkTimeoutMs,
    firstChunkAbortController,
    chunkTimeoutMs,
    chunkAbortController,
    instructions,
    system,
    prompt,
    messages,
    allowSystemInMessages,
    experimental_sandbox: sandbox,
    tools,
    toolsContext,
    runtimeContext,
    toolChoice,
    transforms: asArray(transform),
    activeTools,
    toolOrder,
    repairToolCall,
    refineToolInput,
    stopConditions: asArray(stopWhen),
    output,
    toolApproval,
    experimental_toolCallers,
    experimental_toolApprovalSecret,
    providerOptions,
    prepareStep,
    timeout,
    onChunk,
    onError,
    canRetryStreamViaOnError: streamRetries !== undefined && onErrorArg != null,
    onEnd,
    onAbort,
    onStepFinish: resolvedOnStepEnd,
    onStart: resolvedOnStart,
    onStepStart: resolvedOnStepStart,
    onLanguageModelCallStart: resolvedOnLanguageModelCallStart,
    onLanguageModelCallEnd: resolvedOnLanguageModelCallEnd,
    onToolExecutionStart: resolvedOnToolExecutionStart,
    onToolExecutionEnd: resolvedOnToolExecutionEnd,
    now,
    generateId,
    generateCallId,
    download,

    // assign default values to include:
    include: {
      requestBody: include?.requestBody ?? false,
      requestMessages: include?.requestMessages ?? false,
      rawChunks: include?.rawChunks ?? includeRawChunks ?? false,
    },
  });
}

export type EnrichedStreamPart<TOOLS extends ToolSet, PARTIAL_OUTPUT> = {
  part: TextStreamPart<TOOLS>;
  partialOutput: PARTIAL_OUTPUT | undefined;
};

type StreamRetryBoundaryMetadata = {
  request: LanguageModelRequestMetadata;
  warnings: Array<CallWarning>;
};

const streamRetryBoundarySymbol = Symbol('streamRetryBoundary');

type StreamRetryBoundaryPart = {
  [streamRetryBoundarySymbol]: StreamRetryBoundaryMetadata;
};

type InternalTextStreamPart<TOOLS extends ToolSet> =
  | TextStreamPart<TOOLS>
  | StreamRetryBoundaryPart;

type InternalEnrichedStreamPart<TOOLS extends ToolSet, PARTIAL_OUTPUT> =
  | EnrichedStreamPart<TOOLS, PARTIAL_OUTPUT>
  | StreamRetryBoundaryPart;

function isStreamRetryBoundaryPart(
  part: object,
): part is StreamRetryBoundaryPart {
  return streamRetryBoundarySymbol in part;
}

async function markPromiseAsHandled<T>(promise: Promise<T>): Promise<void> {
  try {
    await promise;
  } catch {}
}

function createOutputTransformStream<
  TOOLS extends ToolSet,
  OUTPUT extends Output,
>(
  output: OUTPUT,
): TransformStream<
  InternalTextStreamPart<TOOLS>,
  InternalEnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>
> {
  let firstTextChunkId: string | undefined = undefined;
  let text = '';
  let textChunk = '';
  let textProviderMetadata: ProviderMetadata | undefined = undefined;
  let lastPublishedValue: string | undefined = undefined;
  let hasPublishedValue = false;

  function resetOutputState() {
    firstTextChunkId = undefined;
    text = '';
    textChunk = '';
    textProviderMetadata = undefined;
    lastPublishedValue = undefined;
    hasPublishedValue = false;
  }

  function enqueueChunk({
    controller,
    chunk,
  }: {
    controller: TransformStreamDefaultController<
      InternalEnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>
    >;
    chunk: EnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>;
  }) {
    controller.enqueue(chunk);
  }

  function publishTextChunk({
    controller,
    partialOutput = undefined,
  }: {
    controller: TransformStreamDefaultController<
      InternalEnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>
    >;
    partialOutput?: InferPartialOutput<OUTPUT>;
  }) {
    enqueueChunk({
      controller,
      chunk: {
        part: {
          type: 'text-delta',
          id: firstTextChunkId!,
          text: textChunk,
          providerMetadata: textProviderMetadata,
        },
        partialOutput,
      },
    });
    textChunk = '';
  }

  return new TransformStream<
    InternalTextStreamPart<TOOLS>,
    InternalEnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>
  >({
    async transform(chunk, controller) {
      if (isStreamRetryBoundaryPart(chunk)) {
        resetOutputState();
        controller.enqueue(chunk);
        return;
      }

      if (chunk.type === 'start-step') {
        resetOutputState();
      }

      // ensure that we publish the last text chunk before the step finish:
      if (chunk.type === 'finish-step' && textChunk.length > 0) {
        publishTextChunk({ controller });
      }

      if (
        chunk.type !== 'text-delta' &&
        chunk.type !== 'text-start' &&
        chunk.type !== 'text-end'
      ) {
        enqueueChunk({
          controller,
          chunk: { part: chunk, partialOutput: undefined },
        });
        return;
      }

      // we have to pick a text chunk which contains the json text
      // since we are streaming, we have to pick the first text chunk
      if (firstTextChunkId == null) {
        firstTextChunkId = chunk.id;
      } else if (chunk.id !== firstTextChunkId) {
        enqueueChunk({
          controller,
          chunk: { part: chunk, partialOutput: undefined },
        });
        return;
      }

      if (chunk.type === 'text-start') {
        enqueueChunk({
          controller,
          chunk: { part: chunk, partialOutput: undefined },
        });
        return;
      }

      if (chunk.type === 'text-end') {
        if (textChunk.length > 0) {
          publishTextChunk({ controller });
        }
        enqueueChunk({
          controller,
          chunk: { part: chunk, partialOutput: undefined },
        });
        return;
      }

      text += chunk.text;
      textChunk += chunk.text;
      textProviderMetadata = chunk.providerMetadata ?? textProviderMetadata;

      if (chunk.text.length === 0 && chunk.providerMetadata != null) {
        enqueueChunk({
          controller,
          chunk: { part: chunk, partialOutput: undefined },
        });
        return;
      }

      // only publish if partial json can be parsed:
      const result = await output.parsePartialOutput({ text });

      // null should be allowed (valid JSON value) but undefined should not:
      if (result !== undefined) {
        // only send new value if it has changed:
        // For string partials (text output), compare directly to avoid unnecessary JSON.stringify overhead
        const currentValue =
          typeof result.partial === 'string'
            ? result.partial
            : JSON.stringify(result.partial);
        if (!hasPublishedValue || currentValue !== lastPublishedValue) {
          publishTextChunk({ controller, partialOutput: result.partial });
          lastPublishedValue = currentValue;
          hasPublishedValue = true;
        }
      }
    },
  });
}

function applyStreamTextTransforms<TOOLS extends ToolSet>({
  stream,
  transforms,
  tools,
  stopStream,
}: {
  stream: ReadableStream<InternalTextStreamPart<TOOLS>>;
  transforms: Array<StreamTextTransform<TOOLS>>;
  tools: TOOLS;
  stopStream: () => void;
}): ReadableStream<InternalTextStreamPart<TOOLS>> {
  const sourceReader = stream.getReader();
  let sourceDone = false;
  let pendingBoundary: StreamRetryBoundaryPart | undefined;
  let transformedSegmentReader:
    | ReadableStreamDefaultReader<TextStreamPart<TOOLS>>
    | undefined;

  const createTransformedSegmentReader = () => {
    let segment = new ReadableStream<TextStreamPart<TOOLS>>(
      {
        async pull(controller) {
          const { done, value } = await sourceReader.read();

          if (done) {
            sourceDone = true;
            controller.close();
            return;
          }

          if (isStreamRetryBoundaryPart(value)) {
            pendingBoundary = value;
            controller.close();
            return;
          }

          controller.enqueue(value);
        },
        cancel(reason) {
          return sourceReader.cancel(reason);
        },
      },
      // Do not prefetch the next source chunk. `stopStream` is invoked from a
      // user transform and must close the gate before another chunk enters it.
      { highWaterMark: 0 },
    );

    for (const transform of transforms) {
      segment = segment.pipeThrough(
        transform({
          tools,
          stopStream,
        }),
      );
    }

    return segment.getReader();
  };

  return new ReadableStream<InternalTextStreamPart<TOOLS>>({
    async pull(controller) {
      transformedSegmentReader ??= createTransformedSegmentReader();

      const { done, value } = await transformedSegmentReader.read();

      if (!done) {
        controller.enqueue(value);
        return;
      }

      transformedSegmentReader = undefined;

      if (pendingBoundary != null) {
        controller.enqueue(pendingBoundary);
        pendingBoundary = undefined;
        return;
      }

      if (sourceDone) {
        controller.close();
      }
    },
    async cancel(reason) {
      await transformedSegmentReader?.cancel(reason);
      await sourceReader.cancel(reason);
    },
  });
}

class DefaultStreamTextResult<
  TOOLS extends ToolSet,
  RUNTIME_CONTEXT extends Context,
  OUTPUT extends Output,
> implements StreamTextResult<TOOLS, RUNTIME_CONTEXT, OUTPUT> {
  private readonly _totalUsage = new DelayedPromise<
    Awaited<StreamTextResult<TOOLS, RUNTIME_CONTEXT, OUTPUT>['usage']>
  >();
  private readonly _finishReason = new DelayedPromise<
    Awaited<StreamTextResult<TOOLS, RUNTIME_CONTEXT, OUTPUT>['finishReason']>
  >();
  private readonly _rawFinishReason = new DelayedPromise<
    Awaited<StreamTextResult<TOOLS, RUNTIME_CONTEXT, OUTPUT>['rawFinishReason']>
  >();
  private readonly _steps = new DelayedPromise<
    Awaited<StreamTextResult<TOOLS, RUNTIME_CONTEXT, OUTPUT>['steps']>
  >();
  private readonly _initialResponseMessages = new DelayedPromise<
    Array<ResponseMessage>
  >();

  private outputPromise: Promise<InferCompleteOutput<OUTPUT>> | undefined;

  private readonly addStream: (
    stream: ReadableStream<InternalTextStreamPart<TOOLS>>,
    callbacks?: {
      onError?: (error: unknown) => void;
      onCancel?: () => void;
    },
  ) => void;

  private readonly closeStream: () => void;

  private baseStream: ReadableStream<
    EnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>
  >;

  private outputSpecification: OUTPUT | undefined;

  private tools: TOOLS | undefined;

  constructor({
    model,
    telemetry,
    headers,
    settings,
    maxRetries: maxRetriesArg,
    streamRetries: streamRetriesArg,
    abortSignal,
    stepTimeoutMs,
    stepAbortController,
    firstChunkTimeoutMs,
    firstChunkAbortController,
    chunkTimeoutMs,
    chunkAbortController,
    instructions,
    system,
    prompt,
    messages,
    allowSystemInMessages,
    experimental_sandbox: sandbox,
    tools,
    toolChoice,
    transforms,
    activeTools,
    toolOrder,
    repairToolCall,
    refineToolInput,
    stopConditions,
    output,
    toolApproval,
    experimental_toolCallers,
    experimental_toolApprovalSecret,
    providerOptions,
    prepareStep,
    now,
    generateId,
    generateCallId,
    timeout,
    onChunk,
    onError,
    canRetryStreamViaOnError,
    onEnd,
    onAbort,
    onStepFinish,
    onStart,
    onStepStart,
    onLanguageModelCallStart,
    onLanguageModelCallEnd,
    onToolExecutionStart,
    onToolExecutionEnd,
    runtimeContext,
    toolsContext,
    download,
    include,
  }: {
    model: LanguageModelV4;
    telemetry: TelemetryOptions<RUNTIME_CONTEXT, TOOLS> | undefined;
    headers: Record<string, string | undefined> | undefined;
    settings: LanguageModelCallOptions;
    maxRetries: number | undefined;
    streamRetries: number | undefined;
    abortSignal: AbortSignal | undefined;
    stepTimeoutMs: number | undefined;
    stepAbortController: AbortController | undefined;
    firstChunkTimeoutMs: number | undefined;
    firstChunkAbortController: AbortController | undefined;
    chunkTimeoutMs: number | undefined;
    chunkAbortController: AbortController | undefined;
    toolsContext: InferToolSetContext<TOOLS>;
    runtimeContext: RUNTIME_CONTEXT;
    instructions: Prompt['instructions'];
    system: Prompt['system'];
    prompt: Prompt['prompt'];
    messages: Prompt['messages'];
    allowSystemInMessages: Prompt['allowSystemInMessages'];
    experimental_sandbox: SandboxSession | undefined;
    tools: TOOLS | undefined;
    toolChoice: ToolChoice<TOOLS> | undefined;
    transforms: Array<StreamTextTransform<TOOLS>>;
    activeTools: ActiveTools<TOOLS>;
    toolOrder: ToolOrder<TOOLS>;
    repairToolCall: ToolCallRepairFunction<TOOLS> | undefined;
    refineToolInput: ToolInputRefinement<TOOLS> | undefined;
    stopConditions: Array<
      StopCondition<NoInfer<TOOLS>, NoInfer<RUNTIME_CONTEXT>>
    >;
    output: OUTPUT | undefined;
    toolApproval: ToolApprovalConfiguration<TOOLS, RUNTIME_CONTEXT> | undefined;
    experimental_toolCallers: Experimental_ToolCallers<TOOLS> | undefined;
    experimental_toolApprovalSecret: string | Uint8Array | undefined;
    providerOptions: ProviderOptions | undefined;
    prepareStep:
      | PrepareStepFunction<NoInfer<TOOLS>, NoInfer<RUNTIME_CONTEXT>>
      | undefined;
    now: () => number;
    generateId: () => string;
    generateCallId: () => string;
    timeout: TimeoutConfiguration<TOOLS> | undefined;
    download: DownloadFunction | undefined;
    include: Required<StreamTextInclude>;

    // callbacks:
    onChunk: undefined | StreamTextOnChunkCallback<TOOLS>;
    onError: StreamTextOnErrorHandler;
    canRetryStreamViaOnError: boolean;
    onEnd:
      | undefined
      | StreamTextOnEndCallback<
          NoInfer<TOOLS>,
          NoInfer<RUNTIME_CONTEXT>,
          NoInfer<OUTPUT>
        >;
    onAbort:
      | undefined
      | StreamTextOnAbortCallback<NoInfer<TOOLS>, NoInfer<RUNTIME_CONTEXT>>;
    onStepFinish:
      | undefined
      | GenerateTextOnStepFinishCallback<
          NoInfer<TOOLS>,
          NoInfer<RUNTIME_CONTEXT>
        >;
    onStart:
      | undefined
      | GenerateTextOnStartCallback<
          NoInfer<TOOLS>,
          NoInfer<RUNTIME_CONTEXT>,
          NoInfer<OUTPUT>
        >;
    onStepStart:
      | undefined
      | GenerateTextOnStepStartCallback<
          NoInfer<TOOLS>,
          NoInfer<RUNTIME_CONTEXT>,
          NoInfer<OUTPUT>
        >;
    onLanguageModelCallStart: undefined | OnLanguageModelCallStartCallback;
    onLanguageModelCallEnd:
      | undefined
      | OnLanguageModelCallEndCallback<NoInfer<TOOLS>>;
    onToolExecutionStart: undefined | OnToolExecutionStartCallback<TOOLS>;
    onToolExecutionEnd: undefined | OnToolExecutionEndCallback<TOOLS>;
  }) {
    this.outputSpecification = output;
    this.tools = tools;
    const resolvedToolCallers = resolveToolCallerConfiguration({
      tools,
      toolCallers: experimental_toolCallers,
    });
    const prepareToolSearch = createToolSearchState({
      tools,
      toolCallers: resolvedToolCallers,
    });

    const telemetryDispatcher = createRestrictedTelemetryDispatcher<
      TOOLS,
      RUNTIME_CONTEXT,
      OUTPUT
    >({
      telemetry,
      includeRuntimeContext: telemetry?.includeRuntimeContext,
      includeToolsContext: telemetry?.includeToolsContext,
    });

    // promise to ensure that the step has been fully processed by the event processor
    // before a new step is started. This is required because the continuation condition
    // needs the updated steps to determine if another step is needed.
    let stepFinish!: DelayedPromise<void>;

    let recordedContent: Array<ContentPart<TOOLS>> = [];
    let recordedFinishReason: FinishReason | undefined = undefined;
    let recordedRawFinishReason: string | undefined = undefined;
    let recordedTotalUsage: LanguageModelUsage | undefined = undefined;
    let recordedRequest: Omit<LanguageModelRequestMetadata, 'messages'> = {};
    let recordedRequestMessages: Array<ModelMessage> = [];
    let recordedWarnings: Array<CallWarning> = [];
    const recordedSteps: StepResult<TOOLS, RUNTIME_CONTEXT>[] = [];
    const initialResponseMessages: Array<ResponseMessage> = [];
    let stepMessagesForNextStep: Array<ModelMessage> | undefined;
    let currentStepMessages: Array<ModelMessage> = [];
    let isAborted = false;
    let currentStepModel = model;

    // provider-assigned text/reasoning part IDs are only unique within a
    // single model call (e.g. Anthropic uses the content block index, which
    // restarts at 0 for every call), so colliding IDs are remapped to keep
    // them unique across the whole multi-step stream:
    const createPartIdReserver = () => {
      const usedIds = new Set<string>();

      return (id: string) => {
        if (!usedIds.has(id)) {
          usedIds.add(id);
          return id;
        }

        const generatedId = generateId();
        let uniqueId = generatedId;
        let suffix = 0;

        while (usedIds.has(uniqueId)) {
          uniqueId = `${generatedId}-${++suffix}`;
        }

        usedIds.add(uniqueId);
        return uniqueId;
      };
    };

    const reserveTextPartId = createPartIdReserver();
    const reserveReasoningPartId = createPartIdReserver();

    // Track provider-executed tool calls that support deferred results
    // (e.g., code_execution in programmatic tool calling scenarios).
    // These tools may not return their results in the same turn as their call.
    const pendingDeferredToolCalls = new Map<string, { toolName: string }>();

    let activeTextContent: Record<
      string,
      {
        type: 'text';
        text: string;
        providerMetadata: ProviderMetadata | undefined;
      }
    > = createIdMap();

    let activeReasoningContent: Record<
      string,
      {
        type: 'reasoning';
        text: string;
        providerMetadata: ProviderMetadata | undefined;
      }
    > = createIdMap();
    let recordedNoOutputError: NoOutputGeneratedError | undefined;
    const errorsHandledForStreamRetry = new Set<unknown>();

    const eventProcessor = new TransformStream<
      InternalEnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>,
      EnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>
    >({
      async transform(chunk, controller) {
        if (isStreamRetryBoundaryPart(chunk)) {
          const retryBoundary = chunk[streamRetryBoundarySymbol];
          recordedContent = [];
          activeReasoningContent = createIdMap();
          activeTextContent = createIdMap();
          recordedRequest = retryBoundary.request;
          recordedRequestMessages = retryBoundary.request.messages ?? [];
          recordedWarnings = retryBoundary.warnings;
          return;
        }

        const { part } = chunk;
        controller.enqueue(chunk); // forward the chunk to the next stream

        const callbacksHandledForStreamRetry =
          part.type === 'error' && errorsHandledForStreamRetry.has(part.error);

        if (!callbacksHandledForStreamRetry) {
          await notify({
            event: { chunk: part },
            callbacks: onChunk,
          });
        }

        if (part.type === 'error') {
          const error = wrapGatewayError(part.error);

          if (NoOutputGeneratedError.isInstance(error)) {
            recordedNoOutputError = error;
          }

          if (callbacksHandledForStreamRetry) {
            errorsHandledForStreamRetry.delete(part.error);
          } else {
            await notify({
              event: { error },
              callbacks: async event => {
                await onError(event);
              },
            });
          }
        }

        if (
          part.type === 'custom' ||
          part.type === 'source' ||
          part.type === 'tool-call' ||
          part.type === 'tool-approval-request' ||
          part.type === 'tool-approval-response' ||
          part.type === 'tool-error'
        ) {
          recordedContent.push(part);
        }

        if (part.type === 'text-start') {
          activeTextContent[part.id] = {
            type: 'text',
            text: '',
            providerMetadata: part.providerMetadata,
          };

          recordedContent.push(activeTextContent[part.id]);
        }

        if (part.type === 'text-delta') {
          const activeText = activeTextContent[part.id];

          if (activeText == null) {
            controller.enqueue({
              part: {
                type: 'error',
                error: `text part ${part.id} not found`,
              },
              partialOutput: undefined,
            });
            return;
          }

          activeText.text += part.text;
          activeText.providerMetadata =
            part.providerMetadata ?? activeText.providerMetadata;
        }

        if (part.type === 'text-end') {
          const activeText = activeTextContent[part.id];

          if (activeText == null) {
            controller.enqueue({
              part: {
                type: 'error',
                error: `text part ${part.id} not found`,
              },
              partialOutput: undefined,
            });
            return;
          }

          activeText.providerMetadata =
            part.providerMetadata ?? activeText.providerMetadata;

          delete activeTextContent[part.id];
        }

        if (part.type === 'reasoning-start') {
          activeReasoningContent[part.id] = {
            type: 'reasoning',
            text: '',
            providerMetadata: part.providerMetadata,
          };

          recordedContent.push(activeReasoningContent[part.id]);
        }

        if (part.type === 'reasoning-delta') {
          const activeReasoning = activeReasoningContent[part.id];

          if (activeReasoning == null) {
            controller.enqueue({
              part: {
                type: 'error',
                error: `reasoning part ${part.id} not found`,
              },
              partialOutput: undefined,
            });
            return;
          }

          activeReasoning.text += part.text;
          activeReasoning.providerMetadata =
            part.providerMetadata ?? activeReasoning.providerMetadata;
        }

        if (part.type === 'reasoning-end') {
          const activeReasoning = activeReasoningContent[part.id];

          if (activeReasoning == null) {
            controller.enqueue({
              part: {
                type: 'error',
                error: `reasoning part ${part.id} not found`,
              },
              partialOutput: undefined,
            });
            return;
          }

          activeReasoning.providerMetadata =
            part.providerMetadata ?? activeReasoning.providerMetadata;

          delete activeReasoningContent[part.id];
        }

        if (part.type === 'file' || part.type === 'reasoning-file') {
          recordedContent.push({
            type: part.type,
            file: part.file,
            ...(part.providerMetadata != null
              ? { providerMetadata: part.providerMetadata }
              : {}),
          });
        }

        if (part.type === 'tool-result' && !part.preliminary) {
          recordedContent.push(part);
        }

        if (part.type === 'start-step') {
          // reset the recorded data when a new step starts:
          recordedContent = [];
          activeReasoningContent = createIdMap();
          activeTextContent = createIdMap();

          recordedRequest = part.request;
          recordedWarnings = part.warnings;
        }

        if (part.type === 'finish-step') {
          const stepResponseMessages = await toResponseMessages({
            content: recordedContent,
            tools,
          });

          // Add step information (after response messages are updated):
          const currentStepResult: StepResult<TOOLS, RUNTIME_CONTEXT> =
            new DefaultStepResult({
              callId,
              stepNumber: recordedSteps.length,
              provider: currentStepModel.provider,
              modelId: currentStepModel.modelId,
              runtimeContext,
              toolsContext,
              content: recordedContent,
              finishReason: part.finishReason,
              rawFinishReason: part.rawFinishReason,
              usage: part.usage,
              performance: part.performance,
              warnings: recordedWarnings,
              request: {
                ...recordedRequest,
                messages: include.requestMessages
                  ? cloneModelMessages(recordedRequestMessages)
                  : undefined,
              },
              response: {
                ...part.response,
                messages: cloneModelMessages(stepResponseMessages),
              },
              providerMetadata: part.providerMetadata,
            });

          await notify({
            event: currentStepResult,
            callbacks: [onStepFinish, telemetryDispatcher.onStepEnd],
          });

          logWarnings({
            warnings: recordedWarnings,
            provider: currentStepModel.provider,
            model: currentStepModel.modelId,
          });

          recordedSteps.push(currentStepResult);
          stepMessagesForNextStep = [
            ...currentStepMessages,
            ...stepResponseMessages,
          ];

          // resolve the promise to signal that the step has been fully processed
          // by the event processor:
          stepFinish.resolve();
        }

        if (part.type === 'finish') {
          recordedTotalUsage = part.totalUsage;
          recordedFinishReason = part.finishReason;
          recordedRawFinishReason = part.rawFinishReason;
        }
      },

      async flush(controller) {
        try {
          // reject when no output was generated or an incomplete model stream
          // ended a continuation step:
          if (recordedSteps.length === 0 || recordedNoOutputError != null) {
            const error = abortSignal?.aborted
              ? abortSignal.reason
              : (recordedNoOutputError ??
                new NoOutputGeneratedError({
                  message: 'No output generated. Check the stream for errors.',
                }));

            self.rejectResultPromises(error);

            return; // no steps recorded (e.g. in error scenario)
          }

          // derived:
          const finishReason = recordedFinishReason ?? 'other';
          const totalUsage =
            recordedTotalUsage ?? createNullLanguageModelUsage();

          // from finish:
          self._finishReason.resolve(finishReason);
          self._rawFinishReason.resolve(recordedRawFinishReason);
          self._totalUsage.resolve(totalUsage);

          // aggregate results:
          self._steps.resolve(recordedSteps);

          if (isAborted) {
            return;
          }

          // call onEnd callback:
          const finalStep = recordedSteps[recordedSteps.length - 1];
          const content = recordedSteps.flatMap(step => step.content);
          const files = recordedSteps.flatMap(step => step.files);
          const sources = recordedSteps.flatMap(step => step.sources);
          const toolCalls = recordedSteps.flatMap(step => step.toolCalls);
          const staticToolCalls = recordedSteps.flatMap(
            step => step.staticToolCalls,
          );
          const dynamicToolCalls = recordedSteps.flatMap(
            step => step.dynamicToolCalls,
          );
          const toolResults = recordedSteps.flatMap(step => step.toolResults);
          const staticToolResults = recordedSteps.flatMap(
            step => step.staticToolResults,
          );
          const dynamicToolResults = recordedSteps.flatMap(
            step => step.dynamicToolResults,
          );
          const warnings = recordedSteps.flatMap(step => step.warnings ?? []);
          const onEndWithOutput =
            onEnd == null
              ? undefined
              : async (event: GenerateTextEndEvent<TOOLS, RUNTIME_CONTEXT>) => {
                  const parsedOutput =
                    output == null
                      ? undefined
                      : await self.getOutputPromise().catch(() => undefined);

                  await onEnd({
                    ...event,
                    ...(output != null ? { output: parsedOutput } : {}),
                  });
                };

          const onEndEvent = {
            callId,
            toolsContext: finalStep.toolsContext,
            stepNumber: finalStep.stepNumber,
            model: finalStep.model,
            runtimeContext: finalStep.runtimeContext,
            finishReason: finalStep.finishReason,
            rawFinishReason: finalStep.rawFinishReason,
            usage: totalUsage,
            totalUsage,
            content,
            text: finalStep.text,
            reasoning: finalStep.reasoning,
            reasoningText: finalStep.reasoningText,
            files,
            sources,
            toolCalls,
            staticToolCalls,
            dynamicToolCalls,
            toolResults,
            staticToolResults,
            dynamicToolResults,
            responseMessages: [
              ...initialResponseMessages,
              ...recordedSteps.flatMap(step => step.response.messages),
            ],
            warnings,
            request: finalStep.request,
            response: finalStep.response,
            providerMetadata: finalStep.providerMetadata,
            steps: recordedSteps,
            finalStep,
          };

          await Promise.all([
            notify({
              event: onEndEvent,
              callbacks: onEndWithOutput,
            }),
            notify({
              event: onEndEvent,
              callbacks: telemetryDispatcher.onEnd,
            }),
          ]);
        } catch (error) {
          controller.error(error);
        }
      },
    });

    // initialize the stitchable stream and the transformed stream:
    const stitchableStream =
      createStitchableStream<InternalTextStreamPart<TOOLS>>();
    this.addStream = stitchableStream.addStream;
    this.closeStream = stitchableStream.close;

    // resilient stream that handles abort signals and errors:
    const reader = stitchableStream.stream.getReader();
    let stream = new ReadableStream<InternalTextStreamPart<TOOLS>>({
      async start(controller) {
        // send start event:
        controller.enqueue({ type: 'start' });
      },

      async pull(controller) {
        // abort handling:
        async function abort() {
          isAborted = true;

          await notify({
            event: {
              callId,
              steps: recordedSteps,
              ...(abortSignal?.reason !== undefined
                ? { reason: abortSignal.reason }
                : {}),
            },
            callbacks: [onAbort, telemetryDispatcher.onAbort],
          });
          controller.enqueue({
            type: 'abort',
            // The `reason` is usually of type DOMException, but it can also be of any type,
            // so we use getErrorMessage for serialization because it is already designed to accept values of the unknown type.
            // See: https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal/reason
            ...(abortSignal?.reason !== undefined
              ? { reason: getErrorMessage(abortSignal.reason) }
              : {}),
          });
          controller.close();
        }

        try {
          const { done, value } = await reader.read();

          if (done) {
            controller.close();
            return;
          }

          if (abortSignal?.aborted) {
            await abort();
            return;
          }

          controller.enqueue(value);
        } catch (error) {
          if (isAbortError(error) && abortSignal?.aborted) {
            await abort();
          } else {
            await telemetryDispatcher.onError?.({ callId, error });
            controller.error(error);
          }
        }
      },

      cancel(reason) {
        return stitchableStream.stream.cancel(reason);
      },
    });

    // introduce a gate that prevent further tokens from
    // being emitted after a transform calls stopStream
    let isRunning = true;
    stream = stream.pipeThrough(
      new TransformStream({
        async transform(chunk, controller) {
          if (isRunning) {
            controller.enqueue(chunk);
          }
        },
      }),
    );

    // Transform each retry attempt independently. The retry boundary is
    // intercepted outside user transforms, so transforms may filter or
    // reconstruct any public stream part without losing logical isolation.
    stream = applyStreamTextTransforms({
      stream,
      transforms,
      tools: tools as TOOLS,
      stopStream() {
        stitchableStream.terminate();
        isRunning = false;
      },
    });

    this.baseStream = stream
      .pipeThrough(createOutputTransformStream(output ?? text()))
      .pipeThrough(eventProcessor);

    const { maxRetries } = prepareRetries({
      maxRetries: maxRetriesArg,
      abortSignal,
    });
    const { maxRetries: streamRetries } = prepareRetries({
      maxRetries: streamRetriesArg,
      abortSignal,
      parameter: 'streamRetries',
      defaultMaxRetries: 0,
    });

    const callSettings = prepareLanguageModelCallOptions(settings);

    const self = this;

    const callId = generateCallId();

    (async () => {
      const initialPrompt = await standardizePrompt({
        instructions,
        system,
        prompt,
        messages,
        allowSystemInMessages,
      } as Prompt);

      const startEvent = {
        callId,
        operationId: 'ai.streamText',
        provider: model.provider,
        modelId: model.modelId,
        instructions: initialPrompt.instructions,
        messages: initialPrompt.messages,
        tools,
        toolChoice,
        activeTools,
        toolOrder,
        maxOutputTokens: callSettings.maxOutputTokens,
        temperature: callSettings.temperature,
        topP: callSettings.topP,
        topK: callSettings.topK,
        presencePenalty: callSettings.presencePenalty,
        frequencyPenalty: callSettings.frequencyPenalty,
        stopSequences: callSettings.stopSequences,
        seed: callSettings.seed,
        reasoning: callSettings.reasoning,
        maxRetries,
        timeout,
        headers,
        providerOptions,
        output,
        runtimeContext,
        toolsContext,
      };

      const streamTextTracingChannelContext =
        telemetryDispatcher.startTracingChannelContext?.({
          type: 'streamText',
          event: startEvent,
          completion: self._totalUsage.promise.then(() => undefined),
        });
      // Re-enter the streamText tracing context after stream setup returns.
      const runInStreamTextTracingChannelContext = <T>(execute: () => T): T =>
        streamTextTracingChannelContext?.run(execute) ?? execute();
      const runInTracingChannelSpanInStreamText =
        telemetryDispatcher.runInTracingChannelSpan == null
          ? undefined
          : <T>(
              options: Parameters<
                NonNullable<TelemetryDispatcher['runInTracingChannelSpan']>
              >[0] & {
                execute: () => PromiseLike<T>;
              },
            ) =>
              runInStreamTextTracingChannelContext(() =>
                telemetryDispatcher.runInTracingChannelSpan!(options),
              );

      await notify({
        event: startEvent,
        callbacks: [onStart, telemetryDispatcher.onStart],
      });

      const initialMessages = initialPrompt.messages;
      let instructionsForNextStep = initialPrompt.instructions;

      const { approvedToolApprovals, deniedToolApprovals } =
        collectToolApprovals<TOOLS>({ messages: initialMessages });

      // initial tool execution step stream
      if (deniedToolApprovals.length > 0 || approvedToolApprovals.length > 0) {
        const {
          approvedToolApprovals: localApprovedToolApprovals,
          deniedToolApprovals: revalidationDeniedToolApprovals,
          invalidToolApprovals,
        } = await validateApprovedToolApprovals<TOOLS, RUNTIME_CONTEXT>({
          approvedToolApprovals: approvedToolApprovals.filter(
            toolApproval => !toolApproval.toolCall.providerExecuted,
          ),
          tools,
          toolApproval,
          messages: initialMessages,
          toolsContext,
          runtimeContext,
          toolApprovalSecret: experimental_toolApprovalSecret,
          refineToolInput,
        });

        const localDeniedToolApprovals = [
          ...deniedToolApprovals.filter(
            toolApproval => !toolApproval.toolCall.providerExecuted,
          ),
          ...revalidationDeniedToolApprovals,
        ];
        const localDeniedToolApprovalsWithoutResults =
          localDeniedToolApprovals.filter(
            toolApproval => toolApproval.existingToolResult == null,
          );

        const deniedProviderExecutedToolApprovals = deniedToolApprovals.filter(
          toolApproval => toolApproval.toolCall.providerExecuted,
        );

        let toolExecutionStepStreamController:
          | ReadableStreamDefaultController<TextStreamPart<TOOLS>>
          | undefined;
        const toolExecutionStepStream = new ReadableStream<
          TextStreamPart<TOOLS>
        >({
          start(controller) {
            toolExecutionStepStreamController = controller;
          },
        });

        self.addStream(toolExecutionStepStream);

        try {
          for (const toolApproval of [
            ...localDeniedToolApprovals,
            ...deniedProviderExecutedToolApprovals,
          ]) {
            toolExecutionStepStreamController?.enqueue({
              type: 'tool-output-denied',
              toolCallId: toolApproval.toolCall.toolCallId,
              toolName: toolApproval.toolCall.toolName,
            } as StaticToolOutputDenied<TOOLS>);
          }

          for (const toolApproval of invalidToolApprovals) {
            toolExecutionStepStreamController?.enqueue({
              type: 'tool-error',
              toolCallId: toolApproval.toolCall.toolCallId,
              toolName: toolApproval.toolCall.toolName,
              input: toolApproval.toolCall.input,
              error: getErrorMessage(toolApproval.error),
              title: toolApproval.toolCall.title,
              ...(toolApproval.toolCall.dynamic === true
                ? { dynamic: true as const }
                : {}),
              ...(toolApproval.toolCall.toolMetadata != null
                ? { toolMetadata: toolApproval.toolCall.toolMetadata }
                : {}),
            } as TextStreamPart<TOOLS>);
          }

          const toolOutputs: Array<ToolOutput<TOOLS>> = [];

          await Promise.all(
            localApprovedToolApprovals.map(async toolApproval => {
              const result = await executeToolCall({
                toolCall: toolApproval.toolCall,
                tools,
                callId,
                messages: initialMessages,
                abortSignal,
                timeout,
                experimental_sandbox: sandbox,
                toolsContext,
                onToolExecutionStart: filterNullable(
                  onToolExecutionStart,
                  telemetryDispatcher.onToolExecutionStart,
                ),
                onToolExecutionEnd: filterNullable(
                  onToolExecutionEnd,
                  telemetryDispatcher.onToolExecutionEnd,
                ),
                executeToolInTelemetryContext: telemetryDispatcher.executeTool,
                runInTracingChannelSpan: runInTracingChannelSpanInStreamText,
                onPreliminaryToolResult: result => {
                  toolExecutionStepStreamController?.enqueue(result);
                },
              });

              if (result != null) {
                toolExecutionStepStreamController?.enqueue(result.output);
                toolOutputs.push(result.output);
              }
            }),
          );

          // Local tool results (approved + denied) are sent as tool results:
          if (
            toolOutputs.length > 0 ||
            localDeniedToolApprovalsWithoutResults.length > 0 ||
            invalidToolApprovals.length > 0
          ) {
            const localToolContent: ToolContent = [];

            // add regular tool results for approved tool calls:
            for (const output of toolOutputs) {
              localToolContent.push({
                type: 'tool-result' as const,
                toolCallId: output.toolCallId,
                toolName: output.toolName,
                output: await createToolModelOutput({
                  toolCallId: output.toolCallId,
                  input: output.input,
                  tool: getOwn(tools, output.toolName),
                  output:
                    output.type === 'tool-result'
                      ? output.output
                      : output.error,
                  errorMode: output.type === 'tool-error' ? 'text' : 'none',
                }),
              });
            }

            // Report invalid approved tool calls to the model without
            // executing them. Repairing the input after approval would change
            // the operation that the user authorized.
            for (const toolApproval of invalidToolApprovals) {
              localToolContent.push({
                type: 'tool-result' as const,
                toolCallId: toolApproval.toolCall.toolCallId,
                toolName: toolApproval.toolCall.toolName,
                output: await createToolModelOutput({
                  toolCallId: toolApproval.toolCall.toolCallId,
                  input: toolApproval.toolCall.input,
                  tool: getOwn(tools, toolApproval.toolCall.toolName),
                  output: toolApproval.error,
                  errorMode: 'text',
                }),
              });
            }

            // add execution denied tool results for denied local tool approvals:
            for (const toolApproval of localDeniedToolApprovalsWithoutResults) {
              localToolContent.push({
                type: 'tool-result' as const,
                toolCallId: toolApproval.toolCall.toolCallId,
                toolName: toolApproval.toolCall.toolName,
                output: {
                  type: 'execution-denied' as const,
                  reason: toolApproval.approvalResponse.reason,
                },
              });
            }

            initialResponseMessages.push({
              role: 'tool',
              content: localToolContent,
            });
          }
        } finally {
          toolExecutionStepStreamController?.close();
        }
      }

      self._initialResponseMessages.resolve(initialResponseMessages);

      async function streamStep({
        currentStep,
        usage,
      }: {
        currentStep: number;
        usage: LanguageModelUsage;
      }) {
        // Set up step timeout if configured
        const stepTimeoutId = setAbortTimeout({
          abortController: stepAbortController,
          label: 'Step',
          timeoutMs: stepTimeoutMs,
        });

        // The first-content timeout is armed when the provider response stream
        // starts and is cleared by the first semantic output chunk.
        let firstChunkTimeoutId: ReturnType<typeof setTimeout> | undefined =
          undefined;

        function startFirstChunkTimeout() {
          if (abortSignal?.aborted) {
            return;
          }

          firstChunkTimeoutId = setAbortTimeout({
            abortController: firstChunkAbortController,
            label: 'First chunk',
            timeoutMs: firstChunkTimeoutMs,
          });
        }

        function clearFirstChunkTimeout() {
          if (firstChunkTimeoutId != null) {
            clearTimeout(firstChunkTimeoutId);
            firstChunkTimeoutId = undefined;
          }
        }

        // Chunk timeout tracking starts after semantic output begins and is
        // reset only by subsequent semantic output chunks.
        let chunkTimeoutId: ReturnType<typeof setTimeout> | undefined =
          undefined;

        function resetChunkTimeout() {
          if (chunkTimeoutId != null) {
            clearTimeout(chunkTimeoutId);
          }
          chunkTimeoutId = setAbortTimeout({
            abortController: chunkAbortController,
            label: 'Chunk',
            timeoutMs: chunkTimeoutMs,
          });
        }

        function clearChunkTimeout() {
          if (chunkTimeoutId != null) {
            clearTimeout(chunkTimeoutId);
            chunkTimeoutId = undefined;
          }
        }

        function clearStepTimeout() {
          if (stepTimeoutId != null) {
            clearTimeout(stepTimeoutId);
          }
        }

        function clearStepTimeouts() {
          clearStepTimeout();
          clearFirstChunkTimeout();
          clearChunkTimeout();
        }

        function cleanupStepTimeouts() {
          abortSignal?.removeEventListener('abort', cleanupStepTimeouts);
          clearStepTimeouts();
        }

        // The step's stream is registered lazily and consumed long after this
        // function returns, so its timers must stay armed past setup. When the
        // merged abort signal fires, drop all step-scoped timers so none
        // outlives the step.
        abortSignal?.addEventListener('abort', cleanupStepTimeouts, {
          once: true,
        });

        try {
          stepFinish = new DelayedPromise<void>();

          const stepTracingChannelContext =
            telemetryDispatcher.startTracingChannelContext?.({
              type: 'step',
              event: { callId, stepNumber: currentStep },
              completion: stepFinish.promise,
            });
          // Re-enter the current step before creating child spans.
          const runInStepTracingChannelContext = <T>(execute: () => T): T =>
            stepTracingChannelContext?.run(execute) ?? execute();

          const responseMessagesFromPreviousSteps = recordedSteps.flatMap(
            step => step.response.messages,
          );
          const accumulatedResponseMessages = [
            ...initialResponseMessages,
            ...responseMessagesFromPreviousSteps,
          ];
          const stepInputMessages = stepMessagesForNextStep ?? [
            ...initialMessages,
            ...initialResponseMessages,
          ];

          const prepareStepResult = await prepareStep?.({
            model,
            steps: recordedSteps,
            stepNumber: recordedSteps.length,
            instructions: instructionsForNextStep,
            initialInstructions: initialPrompt.instructions,
            messages: stepInputMessages,
            initialMessages,
            responseMessages: accumulatedResponseMessages,
            toolsContext,
            runtimeContext,
            experimental_sandbox: sandbox,
          });

          const stepSandbox =
            prepareStepResult?.experimental_sandbox ?? sandbox;

          runtimeContext = prepareStepResult?.runtimeContext ?? runtimeContext;
          toolsContext = prepareStepResult?.toolsContext ?? toolsContext;

          const stepModel = resolveLanguageModel(
            prepareStepResult?.model ?? model,
          );
          currentStepModel = stepModel;

          const stepActiveTools = filterActiveTools({
            tools,
            activeTools: prepareStepResult?.activeTools ?? activeTools,
          });
          const {
            executionTools: stepExecutionTools,
            modelTools: stepModelTools,
            toolCallerMessages,
          } = prepareToolsForToolCallers({
            tools: prepareToolSearch(stepActiveTools, {
              toolsContext,
              experimental_sandbox: stepSandbox,
            }),
            toolCallers: resolvedToolCallers,
          });
          const stepToolOrder = prepareStepResult?.toolOrder ?? toolOrder;

          const stepTools = await prepareTools({
            tools: stepModelTools as ActiveToolSubset<
              TOOLS,
              ActiveTools<NoInfer<TOOLS>>
            >,
            toolOrder: stepToolOrder as ToolOrder<
              ActiveToolSubset<TOOLS, ActiveTools<NoInfer<TOOLS>>>
            >,
            // active tools context is a subset of the tools context, so we can cast to the unknown type
            toolsContext: toolsContext as unknown as InferToolSetContext<
              ActiveToolSubset<TOOLS, ActiveTools<NoInfer<TOOLS>>>
            >,
            experimental_sandbox: stepSandbox,
          });

          const stepToolChoice = prepareToolChoice({
            toolChoice: prepareStepResult?.toolChoice ?? toolChoice,
          });

          const stepMessages = appendToolCallerMessages({
            messages: prepareStepResult?.messages ?? stepInputMessages,
            toolCallerMessages,
          });
          currentStepMessages = stepMessages;
          const stepInstructions =
            prepareStepResult?.instructions ??
            prepareStepResult?.system ??
            instructionsForNextStep;
          instructionsForNextStep = stepInstructions;

          const stepProviderOptions = mergeObjects(
            providerOptions,
            prepareStepResult?.providerOptions,
          );

          const stepCallSettings = prepareStepCallSettings({
            callSettings,
            stepSettings: prepareStepResult,
          });

          const stepStartTimestampMs = now();

          const { retry } = prepareRetries({ maxRetries, abortSignal });
          let hasNotifiedStepStart = false;

          const callLanguageModel = () =>
            runInStepTracingChannelContext(() =>
              retry(async () =>
                streamLanguageModelCall({
                  model: prepareStepResult?.model ?? model,
                  tools: stepModelTools as TOOLS,
                  toolOrder: stepToolOrder,
                  toolChoice: prepareStepResult?.toolChoice ?? toolChoice,
                  instructions: stepInstructions,
                  messages: stepMessages,
                  allowSystemInMessages,
                  repairToolCall,
                  refineToolInput,
                  abortSignal,
                  headers,
                  includeRawChunks: include.rawChunks,
                  providerOptions: stepProviderOptions,
                  download,
                  output,
                  callId,
                  executeLanguageModelCallInTelemetryContext:
                    telemetryDispatcher.executeLanguageModelCall,
                  toolsContext,
                  experimental_sandbox: stepSandbox,
                  onLanguageModelCallStart: filterNullable(
                    onLanguageModelCallStart,
                    telemetryDispatcher.onLanguageModelCallStart as
                      | undefined
                      | OnLanguageModelCallStartCallback,
                  ),
                  onLanguageModelCallEnd: filterNullable(
                    onLanguageModelCallEnd,
                    telemetryDispatcher.onLanguageModelCallEnd as
                      | undefined
                      | OnLanguageModelCallEndCallback<TOOLS>,
                  ),
                  onStart: async ({ promptMessages }) => {
                    if (hasNotifiedStepStart) {
                      return;
                    }
                    hasNotifiedStepStart = true;

                    await notify({
                      event: {
                        callId,
                        provider: stepModel.provider,
                        modelId: stepModel.modelId,
                        stepNumber: recordedSteps.length,
                        instructions: stepInstructions,
                        messages: stepMessages,
                        tools,
                        toolChoice: prepareStepResult?.toolChoice ?? toolChoice,
                        activeTools:
                          prepareStepResult?.activeTools ?? activeTools,
                        toolOrder: stepToolOrder,
                        steps: [...recordedSteps],
                        providerOptions: stepProviderOptions,
                        runtimeContext,
                        toolsContext,
                        output,
                        promptMessages,
                        stepTools,
                        stepToolChoice,
                      },
                      callbacks: [onStepStart, telemetryDispatcher.onStepStart],
                    });
                  },
                  _internal: {
                    now,
                  },
                  ...stepCallSettings,
                }),
              ),
            );

          const initialLanguageModelCall = await callLanguageModel();
          let request = initialLanguageModelCall.request;
          let response = initialLanguageModelCall.response;
          let languageModelStreamReader =
            initialLanguageModelCall.stream.getReader();
          let automaticStreamRetryCount = 0;
          let callbackStreamRetryCount = 0;
          let bufferedAttemptParts: LanguageModelStreamPart<TOOLS>[] = [];
          const outputChunksHandledBeforeBuffering = new WeakSet<object>();
          const openTextParts = new Set<string>();
          const openReasoningParts = new Set<string>();
          let enqueueStreamRetryAttemptBoundary = false;
          const shouldBufferToolParts =
            streamRetries > 0 || canRetryStreamViaOnError;

          const languageModelStream = new ReadableStream<
            LanguageModelStreamPart<TOOLS> | StreamRetryAttemptBoundaryPart
          >({
            async pull(controller) {
              const enqueueAttemptPart = (
                part: LanguageModelStreamPart<TOOLS>,
              ) => {
                switch (part.type) {
                  case 'text-start':
                    openTextParts.add(part.id);
                    break;
                  case 'text-end':
                    openTextParts.delete(part.id);
                    break;
                  case 'reasoning-start':
                    openReasoningParts.add(part.id);
                    break;
                  case 'reasoning-end':
                    openReasoningParts.delete(part.id);
                    break;
                }

                controller.enqueue(part);
              };

              const flushBufferedAttemptParts = () => {
                for (const part of bufferedAttemptParts) {
                  enqueueAttemptPart(part);
                }
                bufferedAttemptParts = [];
              };

              const closeOpenAttemptParts = () => {
                for (const id of openTextParts) {
                  controller.enqueue({ type: 'text-end', id });
                }
                openTextParts.clear();

                for (const id of openReasoningParts) {
                  controller.enqueue({ type: 'reasoning-end', id });
                }
                openReasoningParts.clear();
              };

              while (true) {
                const { done, value } = await languageModelStreamReader.read();

                if (enqueueStreamRetryAttemptBoundary) {
                  controller.enqueue(
                    createStreamRetryAttemptBoundaryPart({
                      warnings:
                        !done && value.type === 'model-call-start'
                          ? value.warnings
                          : [],
                    }),
                  );
                  enqueueStreamRetryAttemptBoundary = false;
                }

                if (done) {
                  flushBufferedAttemptParts();
                  controller.close();
                  return;
                }

                const isToolPart =
                  value.type === 'tool-input-start' ||
                  value.type === 'tool-input-delta' ||
                  value.type === 'tool-input-end' ||
                  value.type === 'tool-call' ||
                  value.type === 'tool-approval-request' ||
                  value.type === 'tool-approval-response' ||
                  value.type === 'tool-result' ||
                  value.type === 'tool-error';

                if (value.type === 'model-call-end') {
                  flushBufferedAttemptParts();
                  enqueueAttemptPart(value);
                  return;
                }

                if (
                  shouldBufferToolParts &&
                  value.type !== 'error' &&
                  (isToolPart || bufferedAttemptParts.length > 0)
                ) {
                  if (isOutputChunk(value)) {
                    clearFirstChunkTimeout();
                    resetChunkTimeout();
                    outputChunksHandledBeforeBuffering.add(value);
                  }

                  bufferedAttemptParts.push(value);
                  continue;
                }

                if (value.type !== 'error') {
                  enqueueAttemptPart(value);
                  return;
                }

                await notify({
                  event: { chunk: value },
                  callbacks: onChunk,
                });
                const error = wrapGatewayError(value.error);
                const isToolChoiceViolation =
                  ToolChoiceViolationError.isInstance(error);
                let onErrorResult: unknown;
                try {
                  onErrorResult = await onError({ error });
                } catch {}
                const callbackRequestedRetry =
                  canRetryStreamViaOnError &&
                  typeof onErrorResult === 'object' &&
                  onErrorResult != null &&
                  'retry' in onErrorResult &&
                  onErrorResult.retry === true;
                const automaticRetry =
                  !isToolChoiceViolation &&
                  automaticStreamRetryCount < streamRetries;
                const callbackRetry =
                  !isToolChoiceViolation &&
                  !automaticRetry &&
                  callbackRequestedRetry &&
                  callbackStreamRetryCount < 1;

                if (!automaticRetry && !callbackRetry) {
                  flushBufferedAttemptParts();
                  errorsHandledForStreamRetry.add(value.error);
                  controller.enqueue(value);
                  return;
                }

                if (automaticRetry) {
                  automaticStreamRetryCount++;
                } else {
                  callbackStreamRetryCount++;
                }

                await languageModelStreamReader.cancel(error);
                bufferedAttemptParts = [];
                closeOpenAttemptParts();

                let retryLanguageModelCall: Awaited<
                  ReturnType<typeof callLanguageModel>
                >;
                try {
                  retryLanguageModelCall = await callLanguageModel();
                } catch (retryError) {
                  controller.enqueue({
                    type: 'error',
                    error: retryError,
                  });
                  controller.close();
                  return;
                }
                request = retryLanguageModelCall.request;
                response = retryLanguageModelCall.response;
                languageModelStreamReader =
                  retryLanguageModelCall.stream.getReader();
                enqueueStreamRetryAttemptBoundary = true;
              }
            },
            cancel(reason) {
              return languageModelStreamReader.cancel(reason);
            },
          });

          startFirstChunkTimeout();

          const streamAfterToolCallbackInvocation =
            invokeToolCallbacksFromStream({
              stream: languageModelStream,
              tools: stepExecutionTools as TOOLS,
              stepInputMessages: stepMessages,
              abortSignal,
              toolsContext,
            });

          // Create child spans under the current step context.
          const runInTracingChannelSpanInStep =
            telemetryDispatcher.runInTracingChannelSpan == null
              ? undefined
              : <T>(
                  options: Parameters<
                    NonNullable<TelemetryDispatcher['runInTracingChannelSpan']>
                  >[0] & {
                    execute: () => PromiseLike<T>;
                  },
                ) =>
                  runInStepTracingChannelContext(() =>
                    telemetryDispatcher.runInTracingChannelSpan!(options),
                  );

          const streamWithToolResults = executeToolsFromStream({
            stream: streamAfterToolCallbackInvocation,
            tools: stepExecutionTools as TOOLS,
            callId,
            messages: stepMessages,
            abortSignal,
            timeout,
            experimental_sandbox: stepSandbox,
            toolsContext,
            toolApproval,
            runtimeContext,
            toolApprovalSecret: experimental_toolApprovalSecret,
            generateId,

            // the callbacks need to be passed down and handled by executeToolCall
            // to guarantee that the onToolExecutionStart callback is invoked before the tool execute function
            onToolExecutionStart: filterNullable(
              onToolExecutionStart,
              telemetryDispatcher.onToolExecutionStart,
            ),
            onToolExecutionEnd: filterNullable(
              onToolExecutionEnd,
              telemetryDispatcher.onToolExecutionEnd,
            ),

            executeToolInTelemetryContext: telemetryDispatcher.executeTool,
            runInTracingChannelSpan: runInTracingChannelSpanInStep,
          });

          // Conditionally include request.body based on include settings.
          // Large payloads (e.g., base64-encoded images) can cause memory issues.
          const getStepRequest = (): LanguageModelRequestMetadata => ({
            ...request,
            body: include.requestBody ? request?.body : undefined,
            messages: include.requestMessages
              ? cloneModelMessages(stepMessages)
              : undefined,
          });
          recordedRequestMessages = getStepRequest().messages ?? [];

          const stepToolCalls: TypedToolCall<TOOLS>[] = [];
          const stepToolOutputs: ToolOutput<TOOLS>[] = [];
          const stepToolApprovalResponses: ToolApprovalResponse[] = [];
          let warnings: SharedV4Warning[] | undefined;

          let stepFinishReason: FinishReason = 'other';
          let stepRawFinishReason: string | undefined = undefined;

          // terminal chunk = 'model-call-end' or 'error'; absence on stream
          // end means the model stream is incomplete:
          let hasReceivedTerminalChunk = false;

          // output chunk = any content chunk (text, tool calls, etc.);
          // used to distinguish empty incomplete streams from partial results:
          let hasReceivedOutputChunk = false;

          let stepUsage: LanguageModelUsage = createNullLanguageModelUsage();
          let stepProviderMetadata: ProviderMetadata | undefined;
          let stepFirstChunk = true;
          const createModelCallPerformance = (): Omit<
            StepResultPerformance,
            'stepTimeMs' | 'toolExecutionMs'
          > => ({
            responseTimeMs: 0,
            effectiveOutputTokensPerSecond: 0,
            outputTokensPerSecond: undefined,
            inputTokensPerSecond: undefined,
            effectiveTotalTokensPerSecond: 0,
            timeToFirstOutputMs: undefined,
            timeBetweenOutputChunksMs: undefined,
          });
          let modelCallPerformance = createModelCallPerformance();
          const toolExecutionMs: Record<string, number> = {};
          const createStepResponse = () => ({
            id: generateId(),
            timestamp: new Date(),
            modelId: stepModel.modelId,
          });
          let stepResponse: {
            id: string;
            timestamp: Date;
            modelId: string;
          } = createStepResponse();

          // maps provider-assigned IDs to stream-unique IDs for the text and
          // reasoning parts that are active in this step
          const textPartIds = new Map<string, string>();
          const reasoningPartIds = new Map<string, string>();

          const enqueueStepPart = (
            controller: TransformStreamDefaultController<
              InternalTextStreamPart<TOOLS>
            >,
            part: TextStreamPart<TOOLS>,
          ) => {
            controller.enqueue(part);
          };

          self.addStream(
            streamWithToolResults.pipeThrough(
              new TransformStream<
                ExecuteToolsStreamPart<TOOLS>,
                InternalTextStreamPart<TOOLS>
              >({
                async transform(chunk, controller): Promise<void> {
                  if (isStreamRetryAttemptBoundaryPart(chunk)) {
                    warnings = chunk.warnings;
                    stepFinishReason = 'other';
                    stepRawFinishReason = undefined;
                    hasReceivedTerminalChunk = false;
                    hasReceivedOutputChunk = false;
                    stepUsage = createNullLanguageModelUsage();
                    stepProviderMetadata = undefined;
                    modelCallPerformance = createModelCallPerformance();
                    stepResponse = createStepResponse();
                    textPartIds.clear();
                    reasoningPartIds.clear();

                    controller.enqueue({
                      [streamRetryBoundarySymbol]: {
                        request: getStepRequest(),
                        warnings,
                      } satisfies StreamRetryBoundaryMetadata,
                    });
                    return;
                  }

                  if (chunk.type === 'model-call-start') {
                    warnings = chunk.warnings;
                    return; // stream start chunks are sent immediately and do not count as first chunk
                  }

                  if (stepFirstChunk) {
                    stepFirstChunk = false;

                    // Step start:
                    enqueueStepPart(controller, {
                      type: 'start-step',
                      request: getStepRequest(),
                      warnings: warnings ?? [],
                    });
                  }

                  const chunkType = chunk.type;

                  if (isOutputChunk(chunk)) {
                    const timeoutHandledBeforeBuffering =
                      outputChunksHandledBeforeBuffering.has(chunk);

                    if (
                      !hasReceivedOutputChunk &&
                      !timeoutHandledBeforeBuffering
                    ) {
                      // Clear before forwarding the first output so a timeout
                      // cannot race with already-visible generated content.
                      clearFirstChunkTimeout();
                    }
                    hasReceivedOutputChunk = true;
                    if (!timeoutHandledBeforeBuffering) {
                      resetChunkTimeout();
                    }
                  }

                  switch (chunkType) {
                    case 'file':
                    case 'custom':
                    case 'source':
                    case 'reasoning-file':
                    case 'tool-input-start':
                    case 'tool-input-end':
                    case 'tool-input-delta':
                    case 'tool-approval-request':
                    case 'tool-output-denied': {
                      enqueueStepPart(controller, chunk);
                      break;
                    }

                    case 'text-start': {
                      const id = reserveTextPartId(chunk.id);
                      textPartIds.set(chunk.id, id);
                      enqueueStepPart(controller, { ...chunk, id });
                      break;
                    }

                    case 'text-delta': {
                      if (
                        chunk.text.length > 0 ||
                        chunk.providerMetadata != null
                      ) {
                        enqueueStepPart(controller, {
                          ...chunk,
                          id: textPartIds.get(chunk.id) ?? chunk.id,
                        });
                      }
                      break;
                    }

                    case 'text-end': {
                      enqueueStepPart(controller, {
                        ...chunk,
                        id: textPartIds.get(chunk.id) ?? chunk.id,
                      });
                      textPartIds.delete(chunk.id);
                      break;
                    }

                    case 'reasoning-start': {
                      const id = reserveReasoningPartId(chunk.id);
                      reasoningPartIds.set(chunk.id, id);
                      enqueueStepPart(controller, { ...chunk, id });
                      break;
                    }

                    case 'reasoning-delta': {
                      enqueueStepPart(controller, {
                        ...chunk,
                        id: reasoningPartIds.get(chunk.id) ?? chunk.id,
                      });
                      break;
                    }

                    case 'reasoning-end': {
                      enqueueStepPart(controller, {
                        ...chunk,
                        id: reasoningPartIds.get(chunk.id) ?? chunk.id,
                      });
                      reasoningPartIds.delete(chunk.id);
                      break;
                    }

                    case 'tool-call': {
                      enqueueStepPart(controller, chunk);
                      // store tool calls for onEnd callback and toolCalls promise:
                      stepToolCalls.push(chunk);
                      break;
                    }

                    case 'tool-approval-response': {
                      enqueueStepPart(controller, chunk);
                      stepToolApprovalResponses.push(chunk);
                      break;
                    }

                    case 'tool-result': {
                      enqueueStepPart(controller, chunk);

                      if (!chunk.preliminary) {
                        stepToolOutputs.push(chunk);
                      }

                      break;
                    }

                    case 'tool-error': {
                      enqueueStepPart(controller, chunk);
                      stepToolOutputs.push(chunk);
                      break;
                    }

                    case 'tool-execution-end': {
                      toolExecutionMs[chunk.toolCallId] = chunk.toolExecutionMs;
                      break;
                    }

                    case 'model-call-response-metadata': {
                      stepResponse = {
                        id: chunk.id ?? stepResponse.id,
                        timestamp: chunk.timestamp ?? stepResponse.timestamp,
                        modelId: chunk.modelId ?? stepResponse.modelId,
                      };
                      break;
                    }

                    case 'model-call-end': {
                      hasReceivedTerminalChunk = true;

                      // Note: tool executions might not be finished yet when the finish event is emitted.
                      // store usage and finish reason for promises and onEnd callback:
                      stepUsage = chunk.usage;
                      stepFinishReason = chunk.finishReason;
                      stepRawFinishReason = chunk.rawFinishReason;
                      stepProviderMetadata = chunk.providerMetadata;
                      modelCallPerformance = chunk.performance;

                      break;
                    }

                    case 'error': {
                      hasReceivedTerminalChunk = true;
                      enqueueStepPart(controller, chunk);
                      stepFinishReason = 'error';
                      break;
                    }

                    case 'raw': {
                      if (include.rawChunks) {
                        enqueueStepPart(controller, chunk);
                      }
                      break;
                    }

                    default: {
                      const exhaustiveCheck: never = chunkType;
                      throw new Error(`Unknown chunk type: ${exhaustiveCheck}`);
                    }
                  }
                },

                // invoke onEnd callback and resolve toolResults promise when the stream is about to close:
                async flush(controller) {
                  // emit an error when an incomplete model stream produced no
                  // output instead of recording an empty step. incomplete
                  // streams with partial output retain the partial result:
                  if (!hasReceivedTerminalChunk && !hasReceivedOutputChunk) {
                    enqueueStepPart(controller, {
                      type: 'error',
                      error: new NoOutputGeneratedError({
                        message:
                          'No output generated. The model stream ended without a finish chunk.',
                      }),
                    });

                    cleanupStepTimeouts();
                    self.closeStream();
                    return;
                  }

                  const stepTimeMs = now() - stepStartTimestampMs;

                  const finishStepPart: TextStreamPart<TOOLS> = {
                    type: 'finish-step',
                    finishReason: stepFinishReason,
                    rawFinishReason: stepRawFinishReason,
                    usage: stepUsage,
                    performance: {
                      stepTimeMs,
                      toolExecutionMs,
                      ...modelCallPerformance,
                    },
                    providerMetadata: stepProviderMetadata,
                    response: {
                      ...stepResponse,
                      headers: response?.headers,
                    },
                  };

                  enqueueStepPart(controller, finishStepPart);

                  const combinedUsage = addLanguageModelUsage(usage, stepUsage);

                  // wait for the step to be fully processed by the event processor
                  // to ensure that the recorded steps are complete:
                  await stepFinish.promise;

                  const clientToolCalls = stepToolCalls.filter(
                    toolCall => toolCall.providerExecuted !== true,
                  );
                  const clientToolOutputs = stepToolOutputs.filter(
                    toolOutput => toolOutput.providerExecuted !== true,
                  );
                  const deniedToolApprovalResponses =
                    stepToolApprovalResponses.filter(
                      toolApprovalResponse =>
                        toolApprovalResponse.approved === false,
                    );

                  // Track provider-executed tool calls that support deferred results.
                  // In programmatic tool calling, a server tool (e.g., code_execution) may
                  // trigger a client tool, and the server tool's result is deferred until
                  // the client tool's result is sent back.
                  for (const toolCall of stepToolCalls) {
                    if (toolCall.providerExecuted !== true) continue;
                    const tool = getOwn(stepExecutionTools, toolCall.toolName);
                    if (
                      tool?.type === 'provider' &&
                      tool.supportsDeferredResults
                    ) {
                      // Check if this tool call already has a result in the current step
                      const hasResultInStep = stepToolOutputs.some(
                        output =>
                          (output.type === 'tool-result' ||
                            output.type === 'tool-error') &&
                          output.toolCallId === toolCall.toolCallId,
                      );
                      if (!hasResultInStep) {
                        pendingDeferredToolCalls.set(toolCall.toolCallId, {
                          toolName: toolCall.toolName,
                        });
                      }
                    }
                  }

                  // Mark deferred tool calls as resolved when we receive their results
                  for (const output of stepToolOutputs) {
                    if (
                      output.type === 'tool-result' ||
                      output.type === 'tool-error'
                    ) {
                      pendingDeferredToolCalls.delete(output.toolCallId);
                    }
                  }

                  // Clear this step's timeouts before the next step is started.
                  cleanupStepTimeouts();

                  if (
                    // Continue only after all client tool calls have been executed or denied,
                    // and if there are client results or pending deferred provider results.
                    clientToolCalls.length ===
                      clientToolOutputs.length +
                        deniedToolApprovalResponses.length &&
                    (clientToolCalls.length > 0 ||
                      pendingDeferredToolCalls.size > 0) &&
                    // continue until a stop condition is met:
                    !(await isStopConditionMet({
                      stopConditions,
                      steps: recordedSteps,
                    }))
                  ) {
                    try {
                      await runInStreamTextTracingChannelContext(() =>
                        streamStep({
                          currentStep: currentStep + 1,
                          usage: combinedUsage,
                        }),
                      );
                    } catch (error) {
                      enqueueStepPart(controller, {
                        type: 'error',
                        error,
                      });

                      self.closeStream();
                    }
                  } else {
                    enqueueStepPart(controller, {
                      type: 'finish',
                      finishReason: stepFinishReason,
                      rawFinishReason: stepRawFinishReason,
                      totalUsage: combinedUsage,
                    });

                    self.closeStream(); // close the stitchable stream
                  }
                },
              }),
            ),
            {
              onError: cleanupStepTimeouts,
              onCancel: cleanupStepTimeouts,
            },
          );
        } catch (error) {
          // Setup failed before the stream was registered, so neither the
          // stream's flush nor an abort will clear the timers — clear them here.
          cleanupStepTimeouts();
          throw error;
        }
      }

      await runInStreamTextTracingChannelContext(() =>
        // add the initial stream to the stitchable stream
        streamStep({
          currentStep: 0,
          usage: createNullLanguageModelUsage(),
        }),
      );
    })().catch(async error => {
      await telemetryDispatcher.onError?.({ callId, error });
      self._initialResponseMessages.reject(error);
      markPromiseAsHandled(self._initialResponseMessages.promise);

      // add an error stream part and close the streams:
      self.addStream(
        new ReadableStream({
          start(controller) {
            controller.enqueue({ type: 'error', error });
            controller.close();
          },
        }),
      );
      self.closeStream();
    });
  }

  get steps() {
    // when any of the promises are accessed, the stream is consumed
    // so it resolves without needing to consume the stream separately
    this.consumeStream();

    return this._steps.promise;
  }

  get finalStep() {
    return this.steps.then(steps => steps.at(-1)!);
  }

  get content() {
    return this.steps.then(steps => steps.flatMap(step => step.content));
  }

  get warnings() {
    return this.steps.then(steps => steps.flatMap(step => step.warnings ?? []));
  }

  get providerMetadata() {
    return this.finalStep.then(step => step.providerMetadata);
  }

  get text() {
    return this.finalStep.then(step => step.text);
  }

  get reasoningText() {
    return this.finalStep.then(step => step.reasoningText);
  }

  get reasoning() {
    return this.finalStep.then(step =>
      convertToReasoningOutputs(step.reasoning),
    );
  }

  get sources() {
    return this.steps.then(steps => steps.flatMap(step => step.sources));
  }

  get files() {
    return this.steps.then(steps => steps.flatMap(step => step.files));
  }

  get toolCalls() {
    return this.steps.then(steps => steps.flatMap(step => step.toolCalls));
  }

  get staticToolCalls() {
    return this.steps.then(steps =>
      steps.flatMap(step => step.staticToolCalls),
    );
  }

  get dynamicToolCalls() {
    return this.steps.then(steps =>
      steps.flatMap(step => step.dynamicToolCalls),
    );
  }

  get toolResults() {
    return this.steps.then(steps => steps.flatMap(step => step.toolResults));
  }

  get staticToolResults() {
    return this.steps.then(steps =>
      steps.flatMap(step => step.staticToolResults),
    );
  }

  get dynamicToolResults() {
    return this.steps.then(steps =>
      steps.flatMap(step => step.dynamicToolResults),
    );
  }

  get usage() {
    return this.totalUsage;
  }

  get request() {
    return this.finalStep.then(step => step.request);
  }

  get response() {
    return this.finalStep.then(step => step.response);
  }

  get responseMessages() {
    return Promise.all([
      this._initialResponseMessages.promise,
      this.steps,
    ]).then(([initialResponseMessages, steps]) => [
      ...initialResponseMessages,
      ...steps.flatMap(step => step.response.messages),
    ]);
  }

  get totalUsage() {
    // when any of the promises are accessed, the stream is consumed
    // so it resolves without needing to consume the stream separately
    this.consumeStream();

    return this._totalUsage.promise;
  }

  get finishReason() {
    // when any of the promises are accessed, the stream is consumed
    // so it resolves without needing to consume the stream separately
    this.consumeStream();

    return this._finishReason.promise;
  }

  get rawFinishReason() {
    // when any of the promises are accessed, the stream is consumed
    // so it resolves without needing to consume the stream separately
    this.consumeStream();

    return this._rawFinishReason.promise;
  }

  /**
   * Split out a new stream from the original stream.
   * The original stream is replaced to allow for further splitting,
   * since we do not know how many times the stream will be split.
   *
   * Note: this leads to buffering the stream content on the server.
   * However, the LLM results are expected to be small enough to not cause issues.
   */
  private teeStream() {
    const [stream1, stream2] = this.baseStream.tee();
    this.baseStream = stream2;
    return stream1;
  }

  get textStream(): AsyncIterableStream<string> {
    return createAsyncIterableStream(toTextStream({ stream: this.stream }));
  }

  get stream(): AsyncIterableStream<TextStreamPart<TOOLS>> {
    return createAsyncIterableStream(
      this.teeStream().pipeThrough(
        new TransformStream<
          EnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>,
          TextStreamPart<TOOLS>
        >({
          transform({ part }, controller) {
            controller.enqueue(part);
          },
        }),
      ),
    );
  }

  get fullStream(): AsyncIterableStream<TextStreamPart<TOOLS>> {
    return this.stream;
  }

  private rejectResultPromises(error: unknown) {
    this.rejectResultPromise({ delayedPromise: this._finishReason, error });
    this.rejectResultPromise({ delayedPromise: this._rawFinishReason, error });
    this.rejectResultPromise({ delayedPromise: this._totalUsage, error });
    this.rejectResultPromise({ delayedPromise: this._steps, error });
    this.rejectResultPromise({
      delayedPromise: this._initialResponseMessages,
      error,
    });
  }

  private rejectResultPromise<T>({
    delayedPromise,
    error,
  }: {
    delayedPromise: DelayedPromise<T>;
    error: unknown;
  }) {
    if (delayedPromise.isPending()) {
      delayedPromise.reject(error);
      markPromiseAsHandled(delayedPromise.promise);
    }
  }

  async consumeStream(options?: ConsumeStreamOptions): Promise<void> {
    try {
      await consumeStream({
        stream: this.stream,
        onError: error => {
          this.rejectResultPromises(error);
          options?.onError?.(error);
        },
      });
    } catch (error) {
      this.rejectResultPromises(error);
      options?.onError?.(error);
    }
  }

  get experimental_partialOutputStream(): AsyncIterableStream<
    InferPartialOutput<OUTPUT>
  > {
    return this.partialOutputStream;
  }

  get partialOutputStream(): AsyncIterableStream<InferPartialOutput<OUTPUT>> {
    return createAsyncIterableStream(
      this.teeStream().pipeThrough(
        new TransformStream<
          EnrichedStreamPart<TOOLS, InferPartialOutput<OUTPUT>>,
          InferPartialOutput<OUTPUT>
        >({
          transform({ partialOutput }, controller) {
            if (partialOutput !== undefined) {
              controller.enqueue(partialOutput);
            }
          },
        }),
      ),
    );
  }

  get elementStream(): AsyncIterableStream<InferElementOutput<OUTPUT>> {
    const transform = this.outputSpecification?.createElementStreamTransform();

    if (transform == null) {
      throw new UnsupportedFunctionalityError({
        functionality: `element streams in ${
          this.outputSpecification?.name ?? 'text'
        } mode`,
      });
    }

    return createAsyncIterableStream(this.teeStream().pipeThrough(transform));
  }

  private getOutputPromise(): Promise<InferCompleteOutput<OUTPUT>> {
    if (this.outputPromise == null) {
      this.outputPromise = this.finalStep.then(step => {
        const output = this.outputSpecification ?? text();
        return output.parseCompleteOutput(
          { text: step.text },
          {
            response: step.response,
            usage: step.usage,
            finishReason: step.finishReason,
          },
        );
      });
    }

    return this.outputPromise;
  }

  get output(): Promise<InferCompleteOutput<OUTPUT>> {
    return this.getOutputPromise();
  }

  toUIMessageStream<UI_MESSAGE extends UIMessage>({
    originalMessages,
    generateMessageId,
    onEnd,
    onFinish,
    messageMetadata,
    sendReasoning,
    sendSources,
    sendStart,
    sendFinish,
    onError,
  }: UIMessageStreamOptions<UI_MESSAGE> = {}): AsyncIterableStream<
    InferUIMessageChunk<UI_MESSAGE>
  > {
    return createAsyncIterableStream(
      toUIMessageStreamHelper({
        stream: this.stream,
        tools: this.tools,
        originalMessages,
        generateMessageId,
        onEnd: onEnd ?? onFinish,
        messageMetadata,
        sendReasoning,
        sendSources,
        sendStart,
        sendFinish,
        onError,
      }),
    );
  }

  pipeUIMessageStreamToResponse<UI_MESSAGE extends UIMessage>(
    response: ServerResponse,
    {
      originalMessages,
      generateMessageId,
      onEnd,
      onFinish,
      messageMetadata,
      sendReasoning,
      sendSources,
      sendFinish,
      sendStart,
      onError,
      ...init
    }: UIMessageStreamResponseInit & UIMessageStreamOptions<UI_MESSAGE> = {},
  ) {
    return pipeUIMessageStreamToResponse({
      response,
      stream: this.toUIMessageStream({
        originalMessages,
        generateMessageId,
        onEnd: onEnd ?? onFinish,
        messageMetadata,
        sendReasoning,
        sendSources,
        sendFinish,
        sendStart,
        onError,
      }),
      ...init,
    });
  }

  pipeTextStreamToResponse(response: ServerResponse, init?: ResponseInit) {
    return pipeTextStreamToResponse({
      response,
      stream: this.textStream,
      ...init,
    });
  }

  toUIMessageStreamResponse<UI_MESSAGE extends UIMessage>({
    originalMessages,
    generateMessageId,
    onEnd,
    onFinish,
    messageMetadata,
    sendReasoning,
    sendSources,
    sendFinish,
    sendStart,
    onError,
    ...init
  }: UIMessageStreamResponseInit &
    UIMessageStreamOptions<UI_MESSAGE> = {}): Response {
    return createUIMessageStreamResponse({
      stream: this.toUIMessageStream({
        originalMessages,
        generateMessageId,
        onEnd: onEnd ?? onFinish,
        messageMetadata,
        sendReasoning,
        sendSources,
        sendFinish,
        sendStart,
        onError,
      }),
      ...init,
    });
  }

  toTextStreamResponse(init?: ResponseInit): Response {
    return createTextStreamResponse({
      stream: this.textStream,
      ...init,
    });
  }
}