# What is this?
## File for 'response_cost' calculation in Logging
import logging
import time
from collections.abc import Mapping, Sequence
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Final, Literal, cast

from httpx import Response
from pydantic import BaseModel

import litellm
import litellm._logging
from litellm import verbose_logger
from litellm.constants import (
    DEFAULT_MAX_LRU_CACHE_SIZE,
    DEFAULT_REPLICATE_GPU_PRICE_PER_SECOND,
)
from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import (
    StandardBuiltInToolCostTracking,
)
from litellm.litellm_core_utils.llm_cost_calc.usage_object_transformation import (
    InteractionsUsageObjectTransformation,
    TranscriptionUsageObjectTransformation,
)
from litellm.litellm_core_utils.llm_cost_calc.utils import (
    CostCalculatorUtils,
    _generic_cost_per_character,
    _get_regional_uplift_multiplier,
    _get_service_tier_cost_key,
    calculate_cost_component,
    generic_cost_per_token,
    get_billable_input_tokens,
    get_token_type_cost_breakdown,
    parse_prompt_tokens_details,
    select_cost_metric_for_model,
)
from litellm.llms.anthropic.cost_calculation import (
    cost_per_token as anthropic_cost_per_token,
)
from litellm.llms.azure.cost_calculation import (
    cost_per_token as azure_openai_cost_per_token,
)
from litellm.llms.azure_ai.cost_calculator import (
    cost_per_token as azure_ai_cost_per_token,
)
from litellm.llms.base_llm.search.transformation import SearchResponse
from litellm.llms.bedrock.cost_calculation import (
    cost_per_token as bedrock_cost_per_token,
)
from litellm.llms.databricks.cost_calculator import (
    cost_per_token as databricks_cost_per_token,
)
from litellm.llms.deepseek.cost_calculator import (
    cost_per_token as deepseek_cost_per_token,
)
from litellm.llms.fireworks_ai.cost_calculator import (
    cost_per_token as fireworks_ai_cost_per_token,
)
from litellm.llms.gemini.cost_calculator import cost_per_token as gemini_cost_per_token
from litellm.llms.lemonade.cost_calculator import (
    cost_per_token as lemonade_cost_per_token,
)
from litellm.llms.openai.cost_calculation import (
    _video_output_cost_per_second,
)
from litellm.llms.openai.cost_calculation import (
    cost_per_second as openai_cost_per_second,
)
from litellm.llms.openai.cost_calculation import (
    cost_per_token as openai_cost_per_token,
)
from litellm.llms.perplexity.cost_calculator import (
    cost_per_token as perplexity_cost_per_token,
)
from litellm.llms.tencent.cost_calculator import (
    cost_per_token as tencent_cost_per_token,
)
from litellm.llms.together_ai.cost_calculator import (
    get_model_params_and_category,
    has_together_registry_pricing,
)
from litellm.llms.vertex_ai.cost_calculator import (
    cost_per_character as google_cost_per_character,
)
from litellm.llms.vertex_ai.cost_calculator import (
    cost_per_token as google_cost_per_token,
)
from litellm.llms.vertex_ai.cost_calculator import cost_router as google_cost_router
from litellm.llms.xai.cost_calculator import cost_per_token as xai_cost_per_token
from litellm.responses.utils import ResponseAPILoggingUtils
from litellm.types.agents import LiteLLMSendMessageResponse
from litellm.types.llms.openai import (
    HttpxBinaryResponseContent,
    ImageGenerationRequestQuality,
    OpenAIModerationResponse,
    OpenAIRealtimeStreamList,
    OpenAIRealtimeStreamResponseBaseObject,
    OpenAIRealtimeStreamSessionEvents,
    ResponseAPIUsage,
    ResponsesAPIResponse,
)
from litellm.types.rerank import RerankBilledUnits, RerankResponse
from litellm.types.utils import (
    CallTypesLiteral,
    LiteLLMRealtimeStreamLoggingObject,
    LlmProviders,
    LlmProvidersSet,
    ModelInfo,
    PromptTokensDetailsWrapper,
    ServiceTier,
    StandardBuiltInToolsParams,
    TranscriptionUsageDurationObject,
    TranscriptionUsageTokensObject,
    Usage,
    VectorStoreSearchResponse,
)
from litellm.utils import (
    CallTypes,
    CostPerToken,
    EmbeddingResponse,
    ImageResponse,
    ModelResponse,
    ModelResponseStream,
    ProviderConfigManager,
    TextCompletionResponse,
    TranscriptionResponse,
    _cached_get_model_info_helper,
    token_counter,
)

if TYPE_CHECKING:
    from litellm.litellm_core_utils.litellm_logging import (
        Logging as LitellmLoggingObject,
    )
else:
    LitellmLoggingObject = Any

# Pre-resolved CallTypes enum values for fast membership checks
_A2A_CALL_TYPES: Final = frozenset(
    {
        CallTypes.asend_message.value,
        CallTypes.send_message.value,
    }
)

_VIDEO_CALL_TYPES: Final = frozenset(
    {
        CallTypes.create_video.value,
        CallTypes.acreate_video.value,
        CallTypes.video_edit.value,
        CallTypes.avideo_edit.value,
        CallTypes.video_remix.value,
        CallTypes.avideo_remix.value,
    }
)


_SPEECH_CALL_TYPES: Final = frozenset(
    {
        CallTypes.speech.value,
        CallTypes.aspeech.value,
    }
)

_TRANSCRIPTION_CALL_TYPES: Final = frozenset(
    {
        CallTypes.atranscription.value,
        CallTypes.transcription.value,
    }
)

_RERANK_CALL_TYPES: Final = frozenset(
    {
        CallTypes.rerank.value,
        CallTypes.arerank.value,
    }
)

_SEARCH_CALL_TYPES: Final = frozenset(
    {
        CallTypes.search.value,
        CallTypes.asearch.value,
    }
)

_AREALTIME_CALL_TYPE: Final = CallTypes.arealtime.value
_MCP_CALL_TYPE: Final = CallTypes.call_mcp_tool.value


def _cost_per_token_custom_pricing_helper(
    prompt_tokens: float = 0,
    completion_tokens: float = 0,
    response_time_ms: float | None = 0.0,
    cached_tokens: float = 0,
    cache_creation_tokens: float = 0,
    ### CUSTOM PRICING ###
    custom_cost_per_token: CostPerToken | None = None,
    custom_cost_per_second: float | None = None,
) -> tuple[float, float] | None:
    """Internal helper function for calculating cost, if custom pricing given.

    prompt_tokens is assumed to include both cached_tokens and cache_creation_tokens
    (OpenAI-compatible convention). Anthropic-style usage where prompt_tokens excludes
    cache tokens is handled at the caller (cost_per_token) before invoking this helper.
    """
    if custom_cost_per_token is None and custom_cost_per_second is None:
        return None

    if custom_cost_per_token is not None:
        input_cost_per_token: Final = custom_cost_per_token["input_cost_per_token"]
        output_cost_per_token: Final = custom_cost_per_token["output_cost_per_token"]

        cache_read_input_token_cost: Final = custom_cost_per_token.get(
            "cache_read_input_token_cost",
            input_cost_per_token,
        )
        cache_creation_input_token_cost: Final = custom_cost_per_token.get(
            "cache_creation_input_token_cost",
            input_cost_per_token,
        )

        regular_prompt_tokens: Final = max(
            prompt_tokens - cached_tokens - cache_creation_tokens,
            0,
        )

        input_cost: Final = (
            regular_prompt_tokens * input_cost_per_token
            + cached_tokens * cache_read_input_token_cost
            + cache_creation_tokens * cache_creation_input_token_cost
        )
        output_cost = completion_tokens * output_cost_per_token
        return input_cost, output_cost
    elif custom_cost_per_second is not None:
        output_cost = custom_cost_per_second * (response_time_ms or 0.0) / 1000
        return 0, output_cost

    return None


def _get_additional_costs(
    model: str,
    custom_llm_provider: str | None,
    prompt_tokens: int,
    completion_tokens: int,
) -> dict | None:
    """
    Calculate additional costs beyond standard token costs.

    This function delegates to provider-specific config classes to calculate
    any additional costs like routing fees, infrastructure costs, etc.

    Args:
        model: The model name
        custom_llm_provider: The provider name (optional)
        prompt_tokens: Number of prompt tokens
        completion_tokens: Number of completion tokens

    Returns:
        Optional dictionary with cost names and amounts, or None if no additional costs
    """
    if not custom_llm_provider:
        return None

    try:
        config_class = None
        if custom_llm_provider == "azure_ai":
            from litellm.llms.azure_ai.common_utils import AzureFoundryModelInfo

            config_class = AzureFoundryModelInfo.get_azure_ai_config_for_model(model)
        # Add more providers here as needed
        # elif custom_llm_provider == "other_provider":
        #     config_class = get_other_provider_config(model)

        if config_class and hasattr(config_class, "calculate_additional_costs"):
            return config_class.calculate_additional_costs(
                model=model,
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
            )
    except Exception as e:
        verbose_logger.debug("Error calculating additional costs: %s", e)

    return None


def _transcription_usage_has_token_details(
    usage_block: Usage | None,
) -> bool:
    if usage_block is None:
        return False

    prompt_tokens_val: Final = getattr(usage_block, "prompt_tokens", 0) or 0
    completion_tokens_val: Final = getattr(usage_block, "completion_tokens", 0) or 0
    prompt_details: Final[PromptTokensDetailsWrapper | None] = getattr(usage_block, "prompt_tokens_details", None)

    if prompt_details is not None:
        audio_token_count: Final = getattr(prompt_details, "audio_tokens", 0) or 0
        text_token_count: Final = getattr(prompt_details, "text_tokens", 0) or 0
        if audio_token_count > 0 or text_token_count > 0:
            return True

    return (prompt_tokens_val > 0) or (completion_tokens_val > 0)


def cost_per_token(
    model: str = "",
    prompt_tokens: int = 0,
    completion_tokens: int = 0,
    response_time_ms: float | None = 0.0,
    custom_llm_provider: str | None = None,
    region_name=None,
    ### CHARACTER PRICING ###
    prompt_characters: int | None = None,
    completion_characters: int | None = None,
    ### PROMPT CACHING PRICING ### - used for anthropic
    cache_creation_input_tokens: int | None = 0,
    cache_read_input_tokens: int | None = 0,
    ### CUSTOM PRICING ###
    custom_cost_per_token: CostPerToken | None = None,
    custom_cost_per_second: float | None = None,
    ### NUMBER OF QUERIES ###
    number_of_queries: int | None = None,
    ### USAGE OBJECT ###
    usage_object: Usage | None = None,  # just read the usage object if provided
    ### BILLED UNITS ###
    rerank_billed_units: RerankBilledUnits | None = None,
    ### CALL TYPE ###
    call_type: CallTypesLiteral = "completion",
    audio_transcription_file_duration: float = 0.0,  # for audio transcription calls - the file time in seconds
    ### SERVICE TIER ###
    service_tier: str | None = None,  # for OpenAI service tier pricing
    ### DATA RESIDENCY ###
    data_residency: str | None = None,  # for OpenAI regional-processing uplift (e.g. "eu", "us")
    ### VERTEX LOCATION ###
    vertex_location: str | None = None,  # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global")
    response: Any | None = None,
    ### REQUEST MODEL ###
    request_model: str | None = None,  # original request model for router detection
) -> tuple[float, float]:
    """
    Calculates the cost per token for a given model, prompt tokens, and completion tokens.

    Parameters:
        model (str): The name of the model to use. Default is ""
        prompt_tokens (int): The number of tokens in the prompt.
        completion_tokens (int): The number of tokens in the completion.
        response_time (float): The amount of time, in milliseconds, it took the call to complete.
        prompt_characters (float): The number of characters in the prompt. Used for vertex ai cost calculation.
        completion_characters (float): The number of characters in the completion response. Used for vertex ai cost calculation.
        custom_llm_provider (str): The llm provider to whom the call was made (see init.py for full list)
        custom_cost_per_token: Optional[CostPerToken]: the cost per input + output token for the llm api call.
        custom_cost_per_second: Optional[float]: the cost per second for the llm api call.
        call_type: Optional[str]: the call type

    Returns:
        tuple: A tuple containing the cost in USD dollars for prompt tokens and completion tokens, respectively.
    """

    if model is None:
        raise Exception("Invalid arg. Model cannot be none.")

    ## RECONSTRUCT USAGE BLOCK ##
    if usage_object is not None:
        usage_block = usage_object
    else:
        usage_block = Usage(
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=prompt_tokens + completion_tokens,
            cache_creation_input_tokens=cache_creation_input_tokens,
            cache_read_input_tokens=cache_read_input_tokens,
        )

    ## CUSTOM PRICING ##
    # Normalize cache token counts across providers:
    #   - OpenAI-compatible: usage.prompt_tokens_details.cached_tokens
    #     (prompt_tokens already INCLUDES cached_tokens)
    #   - Anthropic: usage.cache_read_input_tokens / cache_creation_input_tokens
    #     (prompt_tokens does NOT include these — adjust before calling helper)
    _cache_read_tokens: float = 0
    _cache_creation_tokens: float = 0
    _is_anthropic_style = False

    if usage_object is not None:
        _pt_details: Final[PromptTokensDetailsWrapper | None] = getattr(usage_object, "prompt_tokens_details", None)
        if _pt_details is not None:
            _cache_read_tokens = float(getattr(_pt_details, "cached_tokens", 0) or 0)
            # OpenAI-compatible providers report cache-write tokens under
            # either `cache_write_tokens` (kimi-k2) or `cache_creation_tokens`.
            # Mirror db_spend_update_writer to stay symmetric.
            _cache_creation_tokens = float(
                getattr(_pt_details, "cache_write_tokens", 0) or getattr(_pt_details, "cache_creation_tokens", 0) or 0
            )

        _anthropic_read: Final[int | None] = getattr(usage_object, "cache_read_input_tokens", None)
        _anthropic_create: Final[int | None] = getattr(usage_object, "cache_creation_input_tokens", None)
        if _anthropic_read is not None or _anthropic_create is not None:
            _is_anthropic_style = True
            if _anthropic_read is not None:
                _cache_read_tokens = float(_anthropic_read)
            if _anthropic_create is not None:
                _cache_creation_tokens = float(_anthropic_create)

    if not _cache_read_tokens and cache_read_input_tokens:
        _cache_read_tokens = float(cache_read_input_tokens)
        _is_anthropic_style = True
    if not _cache_creation_tokens and cache_creation_input_tokens:
        _cache_creation_tokens = float(cache_creation_input_tokens)
        _is_anthropic_style = True

    # Anthropic reports prompt_tokens as input_tokens (excluding cache tokens).
    # Adjust so the helper's "prompt_tokens includes cache tokens" invariant holds.
    _normalized_prompt_tokens = float(prompt_tokens)
    if _is_anthropic_style:
        _normalized_prompt_tokens += _cache_read_tokens + _cache_creation_tokens

    response_cost: Final = _cost_per_token_custom_pricing_helper(
        prompt_tokens=_normalized_prompt_tokens,
        completion_tokens=completion_tokens,
        response_time_ms=response_time_ms,
        cached_tokens=_cache_read_tokens,
        cache_creation_tokens=_cache_creation_tokens,
        custom_cost_per_second=custom_cost_per_second,
        custom_cost_per_token=custom_cost_per_token,
    )

    if response_cost is not None:
        return response_cost[0], response_cost[1]

    # given
    prompt_tokens_cost_usd_dollar: float = 0
    completion_tokens_cost_usd_dollar: float = 0
    model_cost_ref: Final = litellm.model_cost
    # Only callers that explicitly pass `custom_llm_provider` get the
    # dedup/prefix-join treatment. When provider is omitted, preserve legacy
    # behavior: `model_with_provider` stays equal to the raw `model` string
    # (provider is detected below for downstream use only).
    caller_supplied_provider: Final = custom_llm_provider is not None

    # `model` is normally a string, but callers that mock the transport can pass
    # non-string objects. Only run the string-based dedup/prefix-join when it is
    # actually a string — e.g. a MagicMock's `.startswith()` is always truthy and
    # its slices return new mocks, which would spin the dedup loop forever.
    model_is_str: Final = isinstance(model, str)

    # Router/proxy deployments may repeat the provider segment (e.g. model_name
    # "openai/openai/gpt-5.5"). Strip duplicated `{provider}/` chains before joining.
    if caller_supplied_provider and model_is_str:
        _dup_prefix: Final = f"{custom_llm_provider}/"
        while model.startswith(_dup_prefix):
            _remainder = model[len(_dup_prefix) :]
            if _remainder.startswith(_dup_prefix):
                model = _remainder
            else:
                break

    model_with_provider = model
    if caller_supplied_provider:
        _prov_prefix: Final = f"{custom_llm_provider}/"
        if model_is_str and model.startswith(_prov_prefix):
            model_with_provider = model
        else:
            model_with_provider = f"{custom_llm_provider}/{model}"
        if region_name is not None:
            model_with_provider_and_region: Final = f"{custom_llm_provider}/{region_name}/{model}"
            if model_with_provider_and_region in model_cost_ref:  # use region based pricing, if it's available
                model_with_provider = model_with_provider_and_region
    else:
        _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)

    assert custom_llm_provider is not None  # caller-supplied or get_llm_provider

    model_without_prefix = model
    model_parts: Final = model.split("/", 1)
    if len(model_parts) > 1:
        model_without_prefix = model_parts[1]
    else:
        model_without_prefix = model
    """
    Code block that formats model to lookup in litellm.model_cost
    Option1. model = "bedrock/ap-northeast-1/anthropic.claude-instant-v1". This is the most accurate since it is region based. Should always be option 1
    Option2. model = "openai/gpt-4"       - model = provider/model
    Option3. model = "anthropic.claude-3" - model = model
    """
    if model_with_provider in model_cost_ref:  # Option 2. use model with provider, model = "openai/gpt-4"
        model = model_with_provider
    elif model in model_cost_ref:  # Option 1. use model passed, model="gpt-4"
        model = model
    elif (
        model_without_prefix in model_cost_ref
    ):  # Option 3. if user passed model="bedrock/anthropic.claude-3", use model="anthropic.claude-3"
        model = model_without_prefix

    # see this https://learn.microsoft.com/en-us/azure/ai-services/openai/concepts/models
    if call_type == "speech" or call_type == "aspeech":
        speech_model_info = litellm.get_model_info(model=model_without_prefix, custom_llm_provider=custom_llm_provider)
        cost_metric: Final = select_cost_metric_for_model(speech_model_info)
        prompt_cost: float = 0.0
        completion_cost: float = 0.0
        if cost_metric == "cost_per_character":
            if prompt_characters is None:
                raise ValueError(
                    f"prompt_characters must be provided for tts calls. prompt_characters={prompt_characters}, model={model}, custom_llm_provider={custom_llm_provider}, call_type={call_type}"
                )
            _prompt_cost, _completion_cost = _generic_cost_per_character(
                model=model_without_prefix,
                custom_llm_provider=custom_llm_provider,
                prompt_characters=prompt_characters,
                completion_characters=0,
                custom_prompt_cost=None,
                custom_completion_cost=0,
            )
            if _prompt_cost is None or _completion_cost is None:
                raise ValueError(
                    f"cost for tts call is None. prompt_cost={_prompt_cost}, completion_cost={_completion_cost}, model={model_without_prefix}, custom_llm_provider={custom_llm_provider}, prompt_characters={prompt_characters}, completion_characters={completion_characters}"
                )
            prompt_cost = _prompt_cost
            completion_cost = _completion_cost
        elif cost_metric == "cost_per_token":
            prompt_cost, completion_cost = generic_cost_per_token(
                model=model_without_prefix,
                usage=usage_block,
                custom_llm_provider=custom_llm_provider,
                service_tier=service_tier,
                data_residency=data_residency,
            )

        return prompt_cost, completion_cost
    elif call_type == "arerank" or call_type == "rerank":
        return rerank_cost(
            model=model,
            custom_llm_provider=custom_llm_provider,
            billed_units=rerank_billed_units,
        )
    elif call_type == "avector_store_search" or call_type == "vector_store_search":
        return vector_store_search_cost(
            model=model,
            custom_llm_provider=custom_llm_provider,
            response=cast(VectorStoreSearchResponse, response),
        )
    elif call_type == "ocr" or call_type == "aocr":
        return ocr_cost(
            model=model,
            custom_llm_provider=custom_llm_provider,
            response=response,
        )
    elif (
        call_type == "aretrieve_batch"
        or call_type == "retrieve_batch"
        or call_type == CallTypes.aretrieve_batch
        or call_type == CallTypes.retrieve_batch
    ):
        return batch_cost_calculator(
            usage=usage_block,
            model=model,
            custom_llm_provider=custom_llm_provider,
            data_residency=data_residency,
        )
    elif call_type == "atranscription" or call_type == "transcription":
        if _transcription_usage_has_token_details(usage_block):
            return generic_cost_per_token(
                model=model_without_prefix,
                usage=usage_block,
                custom_llm_provider=custom_llm_provider,
                service_tier=service_tier,
                data_residency=data_residency,
            )

        return openai_cost_per_second(
            model=model_without_prefix,
            custom_llm_provider=custom_llm_provider,
            duration=audio_transcription_file_duration,
        )
    elif call_type == "search" or call_type == "asearch":
        # Search providers use per-query pricing
        from litellm.search import search_provider_cost_per_query

        return search_provider_cost_per_query(
            model=model,
            custom_llm_provider=custom_llm_provider,
            number_of_queries=number_of_queries or 1,
            optional_params=(response._hidden_params if response and hasattr(response, "_hidden_params") else None),
        )
    elif custom_llm_provider == "vertex_ai":
        cost_router: Final = google_cost_router(
            model=model_without_prefix,
            custom_llm_provider=custom_llm_provider,
            call_type=call_type,
        )
        if cost_router == "cost_per_character":
            return google_cost_per_character(
                model=model_without_prefix,
                custom_llm_provider=custom_llm_provider,
                prompt_characters=prompt_characters,
                completion_characters=completion_characters,
                usage=usage_block,
                service_tier=service_tier,
                vertex_location=vertex_location,
            )
        elif cost_router == "cost_per_token":
            return google_cost_per_token(
                model=model_without_prefix,
                custom_llm_provider=custom_llm_provider,
                usage=usage_block,
                service_tier=service_tier,
                vertex_location=vertex_location,
            )
    elif custom_llm_provider == "anthropic":
        return anthropic_cost_per_token(model=model, usage=usage_block, service_tier=service_tier)
    elif custom_llm_provider == "bedrock":
        return bedrock_cost_per_token(model=model, usage=usage_block, service_tier=service_tier)
    elif custom_llm_provider == "openai":
        return openai_cost_per_token(
            model=model,
            usage=usage_block,
            service_tier=service_tier,
            data_residency=data_residency,
        )
    elif custom_llm_provider == "databricks":
        return databricks_cost_per_token(model=model, usage=usage_block)
    elif custom_llm_provider == "fireworks_ai":
        return fireworks_ai_cost_per_token(model=model, usage=usage_block)
    elif custom_llm_provider == "azure":
        return azure_openai_cost_per_token(
            model=model,
            usage=usage_block,
            response_time_ms=response_time_ms,
            service_tier=service_tier,
        )
    elif custom_llm_provider == "gemini":
        return gemini_cost_per_token(model=model, usage=usage_block, service_tier=service_tier)
    elif custom_llm_provider == "deepseek":
        return deepseek_cost_per_token(model=model, usage=usage_block)
    elif custom_llm_provider == "tencent":
        return tencent_cost_per_token(model=model, usage=usage_block)
    elif custom_llm_provider == "perplexity":
        return perplexity_cost_per_token(model=model, usage=usage_block)
    elif custom_llm_provider == "xai":
        return xai_cost_per_token(model=model, usage=usage_block)
    elif custom_llm_provider == "lemonade":
        return lemonade_cost_per_token(model=model, usage=usage_block)
    elif custom_llm_provider in ("dashscope", "qwencloud", "qwen_ai_platform"):
        from litellm.llms.dashscope.cost_calculator import (
            cost_per_token as dashscope_cost_per_token,
        )

        return dashscope_cost_per_token(model=model, usage=usage_block, custom_llm_provider=custom_llm_provider)
    elif custom_llm_provider == "azure_ai":
        return azure_ai_cost_per_token(
            model=model,
            usage=usage_block,
            response_time_ms=response_time_ms,
            request_model=request_model,
            service_tier=service_tier,
        )
    else:
        model_info: Final = _cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)

        if (
            (model_info.get("input_cost_per_token") or 0.0) > 0
            or (model_info.get("output_cost_per_token") or 0.0) > 0
            or model_info.get("tiered_pricing") is not None
        ):
            return generic_cost_per_token(
                model=model,
                usage=usage_block,
                custom_llm_provider=custom_llm_provider,
                service_tier=service_tier,
                data_residency=data_residency,
            )

        input_cost_per_second: Final = model_info.get("input_cost_per_second")
        if input_cost_per_second is not None and response_time_ms is not None:
            verbose_logger.debug(
                "For model=%s - input_cost_per_second: %s; response time: %s",
                model,
                input_cost_per_second,
                response_time_ms,
            )
            ## COST PER SECOND ##
            prompt_tokens_cost_usd_dollar = input_cost_per_second * response_time_ms / 1000

        output_cost_per_second: Final = model_info.get("output_cost_per_second")
        if output_cost_per_second is not None and response_time_ms is not None:
            verbose_logger.debug(
                "For model=%s - output_cost_per_second: %s; response time: %s",
                model,
                output_cost_per_second,
                response_time_ms,
            )
            ## COST PER SECOND ##
            completion_tokens_cost_usd_dollar = output_cost_per_second * response_time_ms / 1000

        verbose_logger.debug(
            "Returned custom cost for model=%s - prompt_tokens_cost_usd_dollar: %s, completion_tokens_cost_usd_dollar: %s",
            model,
            prompt_tokens_cost_usd_dollar,
            completion_tokens_cost_usd_dollar,
        )
        return prompt_tokens_cost_usd_dollar, completion_tokens_cost_usd_dollar


def get_replicate_completion_pricing(completion_response: dict, total_time=0.0):
    # see https://replicate.com/pricing
    # for all litellm currently supported LLMs, almost all requests go to a100_80gb
    a100_80gb_price_per_second_public: Final = (
        DEFAULT_REPLICATE_GPU_PRICE_PER_SECOND  # assume all calls sent to A100 80GB for now
    )
    if total_time == 0.0:  # total time is in ms
        start_time: Final = completion_response.get("created", time.time())
        end_time: Final = getattr(completion_response, "ended", time.time())
        total_time = end_time - start_time

    return a100_80gb_price_per_second_public * total_time / 1000


def has_hidden_params(obj: object) -> bool:
    return hasattr(obj, "_hidden_params")


def _get_provider_for_cost_calc(
    model: str | None,
    custom_llm_provider: str | None = None,
) -> str | None:
    if custom_llm_provider is not None:
        return custom_llm_provider
    if model is None:
        return None
    try:
        _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model)
    except Exception as e:
        verbose_logger.debug(
            "litellm.cost_calculator.py::_get_provider_for_cost_calc() - Error inferring custom_llm_provider - %s", e
        )
        return None

    return custom_llm_provider


def _get_hidden_str_for_cost_calc(hidden_params: object, key: str) -> str | None:
    if not isinstance(hidden_params, Mapping):
        return None
    value: Final[object] = hidden_params.get(key)
    return value if isinstance(value, str) and value else None


def _select_model_name_for_cost_calc(
    model: str | None,
    completion_response: object | None,
    base_model: str | None = None,
    custom_pricing: bool | None = None,
    custom_llm_provider: str | None = None,
    router_model_id: str | None = None,
) -> str | None:
    """
    1. If custom pricing is true, return received model name
    2. If base_model is set (e.g. for azure models), return that
    3. If completion response has model set return that
    4. Check if model is passed in return that
    """

    return_model: str | None = None
    custom_llm_provider = _get_provider_for_cost_calc(model=model, custom_llm_provider=custom_llm_provider)

    completion_response_model: str | None = None
    if completion_response is not None:
        if isinstance(completion_response, BaseModel):
            completion_response_model = getattr(completion_response, "model", None)
        elif isinstance(completion_response, dict):
            completion_response_model = completion_response.get("model", None)
    hidden_params: Final[dict | None] = getattr(completion_response, "_hidden_params", None)
    provider_response_model: Final = _get_hidden_str_for_cost_calc(hidden_params, "provider_response_model")
    explicit_pricing: Final = custom_pricing is True or base_model is not None
    priced_from_response: Final = provider_response_model is not None or completion_response_model is not None
    region_name: Final = (
        _get_hidden_str_for_cost_calc(hidden_params, "region_name")
        if not explicit_pricing and priced_from_response
        else None
    )

    if custom_pricing is True:
        if router_model_id is not None and router_model_id in litellm.model_cost:
            entry: Final = litellm.model_cost[router_model_id]
            if (
                entry.get("input_cost_per_token") is not None
                or entry.get("input_cost_per_second") is not None
                or entry.get("tiered_pricing") is not None
            ):
                return_model = router_model_id
            else:
                return_model = model
        else:
            return_model = model

    elif base_model is not None or provider_response_model is not None:
        return_model = base_model if base_model is not None else provider_response_model

    elif completion_response_model is None and hidden_params is not None:
        if hidden_params.get("model", None) is not None and len(hidden_params["model"]) > 0:
            return_model = hidden_params.get("model", model)

    if return_model is None and completion_response_model is not None:
        return_model = completion_response_model

    if return_model is None and model is not None:
        return_model = model

    if (
        return_model is not None
        and custom_llm_provider is not None
        and not _model_contains_known_llm_provider(return_model)
    ):  # add provider prefix if not already present, to match model_cost
        provider_prefix: Final = custom_llm_provider if region_name is None else f"{custom_llm_provider}/{region_name}"
        return_model = _strip_unregistered_leading_segments(f"{provider_prefix}/{return_model}", region_name)

    return return_model


def _strip_unregistered_leading_segments(model: str, region_name: str | None) -> str:
    """Resolve a provider-prefixed slash alias like "vertex_ai/vertex/claude-opus-5" to the
    registered cost key ("vertex_ai/claude-opus-5"), keeping the model unchanged when it already
    resolves downstream (custom-priced router ids) or no stripped candidate is registered (#38069)."""
    segments: Final = model.split("/")
    if "/".join(segments[1:]) in litellm.model_cost:
        return model
    head_len: Final = 2 if region_name is not None and len(segments) > 2 and segments[1] == region_name else 1
    head: Final = "/".join(segments[:head_len])
    tail: Final = segments[head_len:]
    strippable: Final = next((index for index, segment in enumerate(tail) if segment in LlmProvidersSet), len(tail))
    candidates: Final = (f"{head}/{'/'.join(tail[start:])}" for start in range(min(strippable, len(tail) - 1) + 1))
    return next((candidate for candidate in candidates if candidate in litellm.model_cost), model)


@lru_cache(maxsize=DEFAULT_MAX_LRU_CACHE_SIZE)
def _model_contains_known_llm_provider(model: str) -> bool:
    """
    Check if the model contains a known llm provider
    """
    _provider_prefix: Final = model.split("/")[0]
    return _provider_prefix in LlmProvidersSet


def _get_response_model(completion_response: object) -> str | None:
    """
    Extract the model name from a completion response object.

    Used as a fallback for cost calculation when the input model name
    doesn't exist in model_cost (e.g., Azure Model Router).
    """
    if completion_response is None:
        return None

    if isinstance(completion_response, BaseModel):
        return getattr(completion_response, "model", None)
    elif isinstance(completion_response, dict):
        return completion_response.get("model", None)

    return None


_GEMINI_TRAFFIC_TYPE_TO_SERVICE_TIER: Final[dict] = {
    # ON_DEMAND_PRIORITY maps to "priority" — selects input_cost_per_token_priority, etc.
    "ON_DEMAND_PRIORITY": "priority",
    # FLEX / BATCH / ON_DEMAND_FLEX maps to "flex" — selects input_cost_per_token_flex, etc.
    # Vertex AI reports flex/shared-capacity traffic as ON_DEMAND_FLEX, not FLEX.
    "FLEX": "flex",
    "BATCH": "flex",
    "ON_DEMAND_FLEX": "flex",
    # ON_DEMAND is standard pricing — no service_tier suffix applied
    "ON_DEMAND": None,
}


def _map_traffic_type_to_service_tier(traffic_type: str | None) -> str | None:
    """
    Map a Gemini usageMetadata.trafficType value to a LiteLLM service_tier string.

    This allows the same `_priority` / `_flex` cost-key suffix logic used for
    OpenAI/Azure to work for Gemini and Vertex AI models.

    trafficType values seen in practice
    ------------------------------------
    ON_DEMAND               -> standard pricing  (service_tier = None)
    ON_DEMAND_PRIORITY      -> priority pricing  (service_tier = "priority")
    FLEX / BATCH / ON_DEMAND_FLEX -> batch/flex pricing (service_tier = "flex")
    """
    if traffic_type is None:
        return None
    service_tier: Final = _GEMINI_TRAFFIC_TYPE_TO_SERVICE_TIER.get(str(traffic_type).upper())
    return service_tier


def _normalize_service_tier(service_tier: object) -> str | None:
    """
    Reduce a service_tier value to a concrete billable tier string or None.

    "auto" is a routing preference and any non-string value is not a billable
    tier, so both defer to standard pricing (or to the tier the provider reports
    on the response usage) instead of crashing the downstream cost-key lookup,
    which calls service_tier.lower()
    """
    if not isinstance(service_tier, str) or service_tier.lower() == ServiceTier.AUTO.value:
        return None
    return service_tier


def _extract_service_tier(source: object) -> str | None:
    """Read a raw ``service_tier`` off a response body or usage object, dict or pydantic model alike."""
    if isinstance(source, BaseModel):
        return getattr(source, "service_tier", None)
    elif isinstance(source, dict):
        return source.get("service_tier")

    return None


def _get_usage_object(
    completion_response: object,
) -> Usage | None:
    usage_obj: Final = cast(
        Usage | ResponseAPIUsage | dict | BaseModel,
        (
            completion_response.get("usage")
            if isinstance(completion_response, dict)
            else getattr(completion_response, "get", lambda x: None)("usage")
        ),
    )

    if usage_obj is None:
        return None
    if isinstance(usage_obj, Usage):
        return usage_obj
    elif isinstance(usage_obj, dict) and litellm.AnthropicConfig.is_anthropic_usage_object(usage_obj):
        return litellm.AnthropicConfig().calculate_usage(usage_object=usage_obj, reasoning_content=None)
    elif (
        usage_obj is not None
        and (isinstance(usage_obj, dict) or isinstance(usage_obj, ResponseAPIUsage))
        and ResponseAPILoggingUtils._is_response_api_usage(usage_obj)
    ):
        return ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(usage_obj)
    elif TranscriptionUsageObjectTransformation.is_transcription_usage_object(usage_obj):
        return TranscriptionUsageObjectTransformation.transform_transcription_usage_object(
            cast(
                TranscriptionUsageDurationObject | TranscriptionUsageTokensObject,
                usage_obj,
            )
        )
    elif isinstance(usage_obj, dict) and InteractionsUsageObjectTransformation.is_interactions_usage_object(usage_obj):
        return InteractionsUsageObjectTransformation.transform_interactions_usage_object(usage_obj)
    elif isinstance(usage_obj, dict):
        return Usage(**usage_obj)
    elif isinstance(usage_obj, BaseModel):
        return Usage(**usage_obj.model_dump())
    else:
        verbose_logger.debug("Unknown usage object type: %s, usage_obj: %s", type(usage_obj), usage_obj)
        return None


def _is_known_usage_objects(usage_obj):
    """Returns True if the usage obj is a known Usage type"""
    return (
        isinstance(usage_obj, litellm.Usage)
        or isinstance(usage_obj, ResponseAPIUsage)
        or TranscriptionUsageObjectTransformation.is_transcription_usage_object(usage_obj)
    )


def _infer_call_type(call_type: CallTypesLiteral | None, completion_response: Any) -> CallTypesLiteral | None:
    if call_type is not None:
        return call_type

    if completion_response is None:
        return None

    if isinstance(completion_response, ModelResponse) or isinstance(completion_response, ModelResponseStream):
        return "completion"
    elif isinstance(completion_response, EmbeddingResponse):
        return "embedding"
    elif isinstance(completion_response, TranscriptionResponse):
        return "transcription"
    elif isinstance(completion_response, HttpxBinaryResponseContent):
        return "speech"
    elif isinstance(completion_response, RerankResponse):
        return "rerank"
    elif isinstance(completion_response, ImageResponse):
        return "image_generation"
    elif isinstance(completion_response, TextCompletionResponse):
        return "text_completion"
    elif isinstance(completion_response, LiteLLMSendMessageResponse):
        return "send_message"

    return call_type


def _apply_cost_discount(
    base_cost: float,
    custom_llm_provider: str | None,
) -> tuple[float, float, float]:
    """
    Apply provider-specific cost discount from module-level config.

    Args:
        base_cost: The base cost before discount
        custom_llm_provider: The LLM provider name

    Returns:
        Tuple of (final_cost, discount_percent, discount_amount)
    """
    original_cost: Final = base_cost
    discount_percent = 0.0
    discount_amount = 0.0

    if custom_llm_provider and custom_llm_provider in litellm.cost_discount_config:
        discount_percent = litellm.cost_discount_config[custom_llm_provider]
        discount_amount = original_cost * discount_percent
        final_cost: Final = original_cost - discount_amount

        if verbose_logger.isEnabledFor(logging.DEBUG):
            verbose_logger.debug(
                f"Applied {discount_percent * 100}% discount to {custom_llm_provider}: "
                f"${original_cost:.6f} -> ${final_cost:.6f} (saved ${discount_amount:.6f})"
            )

        return final_cost, discount_percent, discount_amount

    return base_cost, discount_percent, discount_amount


def _apply_cost_margin(
    base_cost: float,
    custom_llm_provider: str | None,
) -> tuple[float, float, float, float]:
    """
    Apply provider-specific or global cost margin from module-level config.

    Args:
        base_cost: The base cost before margin (after discount if applicable)
        custom_llm_provider: The LLM provider name

    Returns:
        Tuple of (final_cost, margin_percent, margin_fixed_amount, margin_total_amount)
    """
    original_cost: Final = base_cost
    margin_percent = 0.0
    margin_fixed_amount = 0.0
    margin_total_amount = 0.0

    # Get margin config - check provider-specific first, then global
    margin_config = None
    if custom_llm_provider and custom_llm_provider in litellm.cost_margin_config:
        margin_config = litellm.cost_margin_config[custom_llm_provider]
        if verbose_logger.isEnabledFor(logging.DEBUG):
            verbose_logger.debug("Found provider-specific margin config for %s: %s", custom_llm_provider, margin_config)
    elif "global" in litellm.cost_margin_config:
        margin_config = litellm.cost_margin_config["global"]
        if verbose_logger.isEnabledFor(logging.DEBUG):
            verbose_logger.debug("Using global margin config: %s", margin_config)
    else:
        if verbose_logger.isEnabledFor(logging.DEBUG):
            verbose_logger.debug(
                "No margin config found. Provider: %s, Available configs: %s",
                custom_llm_provider,
                list(litellm.cost_margin_config.keys()),
            )

    if margin_config is not None:
        # Handle different margin config formats
        if isinstance(margin_config, (int, float)):
            # Simple percentage: {"openai": 0.10}
            margin_percent = float(margin_config)
            margin_total_amount = original_cost * margin_percent
        elif isinstance(margin_config, dict):
            # Complex config: {"percentage": 0.08, "fixed_amount": 0.0005}
            if "percentage" in margin_config:
                margin_percent = float(margin_config["percentage"])
                margin_total_amount += original_cost * margin_percent
            if "fixed_amount" in margin_config:
                margin_fixed_amount = float(margin_config["fixed_amount"])
                margin_total_amount += margin_fixed_amount

        final_cost: Final = original_cost + margin_total_amount

        if verbose_logger.isEnabledFor(logging.DEBUG):
            verbose_logger.debug(
                f"Applied margin to {custom_llm_provider or 'global'}: "
                f"${original_cost:.6f} -> ${final_cost:.6f} "
                f"(margin: {margin_percent * 100 if margin_percent > 0 else 0}% + ${margin_fixed_amount:.6f} = ${margin_total_amount:.6f})"
            )

        return final_cost, margin_percent, margin_fixed_amount, margin_total_amount

    return base_cost, margin_percent, margin_fixed_amount, margin_total_amount


def _store_cost_breakdown_in_logging_obj(
    litellm_logging_obj: LitellmLoggingObject | None,
    prompt_tokens_cost_usd_dollar: float,
    completion_tokens_cost_usd_dollar: float,
    cost_for_built_in_tools_cost_usd_dollar: float,
    total_cost_usd_dollar: float,
    additional_costs: dict | None = None,
    original_cost: float | None = None,
    discount_percent: float | None = None,
    discount_amount: float | None = None,
    margin_percent: float | None = None,
    margin_fixed_amount: float | None = None,
    margin_total_amount: float | None = None,
    cache_read_cost: float | None = None,
    cache_creation_cost: float | None = None,
    reasoning_cost: float | None = None,
    service_tier: str | None = None,
    data_residency: str | None = None,
    vertex_location: str | None = None,
) -> None:
    """
    Helper function to store cost breakdown in the logging object.

    Args:
        litellm_logging_obj: The logging object to store breakdown in
        prompt_tokens_cost_usd_dollar: Cost of input tokens
        completion_tokens_cost_usd_dollar: Cost of completion tokens (includes reasoning if applicable)
        cost_for_built_in_tools_cost_usd_dollar: Cost of built-in tools
        total_cost_usd_dollar: Total cost of request
        additional_costs: Free-form additional costs dict (e.g., {"azure_model_router_flat_cost": 0.00014})
        original_cost: Cost before discount
        discount_percent: Discount percentage applied (0.05 = 5%)
        discount_amount: Discount amount in USD
        margin_percent: Margin percentage applied (0.10 = 10%)
        margin_fixed_amount: Fixed margin amount in USD
        margin_total_amount: Total margin added in USD
        service_tier: Tier the costs above were priced on, already resolved
        data_residency: Region uplift the costs above were priced on, already resolved
        vertex_location: Vertex AI location the costs above were priced on, already resolved
    """
    if litellm_logging_obj is None:
        return

    try:
        # Store the cost breakdown
        litellm_logging_obj.set_cost_breakdown(
            input_cost=prompt_tokens_cost_usd_dollar,
            output_cost=completion_tokens_cost_usd_dollar,
            total_cost=total_cost_usd_dollar,
            cost_for_built_in_tools_cost_usd_dollar=cost_for_built_in_tools_cost_usd_dollar,
            additional_costs=additional_costs,
            original_cost=original_cost,
            discount_percent=discount_percent,
            discount_amount=discount_amount,
            margin_percent=margin_percent,
            margin_fixed_amount=margin_fixed_amount,
            margin_total_amount=margin_total_amount,
            cache_read_cost=cache_read_cost,
            cache_creation_cost=cache_creation_cost,
            reasoning_cost=reasoning_cost,
            service_tier=service_tier,
            data_residency=data_residency,
            vertex_location=vertex_location,
        )

    except Exception as breakdown_error:
        verbose_logger.debug("Error storing cost breakdown: %s", breakdown_error)
        # Don't fail the main cost calculation if breakdown storage fails


def completion_cost(
    completion_response: object | None = None,
    model: str | None = None,
    prompt="",
    messages: list = [],
    completion="",
    total_time: float | None = 0.0,  # used for replicate, sagemaker
    call_type: CallTypesLiteral | None = None,
    ### REGION ###
    custom_llm_provider=None,
    region_name=None,  # used for bedrock pricing
    ### IMAGE GEN ###
    size: str | None = None,
    quality: str | None = None,
    n: int | None = None,  # number of images
    ### CUSTOM PRICING ###
    custom_cost_per_token: CostPerToken | None = None,
    custom_cost_per_second: float | None = None,
    optional_params: dict | None = None,
    custom_pricing: bool | None = None,
    base_model: str | None = None,
    standard_built_in_tools_params: StandardBuiltInToolsParams | None = None,
    litellm_model_name: str | None = None,
    router_model_id: str | None = None,
    litellm_logging_obj: LitellmLoggingObject | None = None,
    ### SERVICE TIER ###
    service_tier: str | None = None,  # for OpenAI service tier pricing
    ### DATA RESIDENCY ###
    data_residency: str | None = None,  # for OpenAI regional-processing uplift (e.g. "eu", "us")
    ### VERTEX LOCATION ###
    vertex_location: str | None = None,  # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global")
) -> float:
    """
    Calculate the cost of a given completion call fot GPT-3.5-turbo, llama2, any litellm supported llm.

    Parameters:
        completion_response (litellm.ModelResponses): [Required] The response received from a LiteLLM completion request.

        [OPTIONAL PARAMS]
        model (str): Optional. The name of the language model used in the completion calls
        prompt (str): Optional. The input prompt passed to the llm
        completion (str): Optional. The output completion text from the llm
        total_time (float, int): Optional. (Only used for Replicate LLMs) The total time used for the request in seconds
        custom_cost_per_token: Optional[CostPerToken]: the cost per input + output token for the llm api call.
        custom_cost_per_second: Optional[float]: the cost per second for the llm api call.

    Returns:
        float: The cost in USD dollars for the completion based on the provided parameters.

    Exceptions:
        Raises exception if model not in the litellm model cost map. Register model, via custom pricing or PR - https://github.com/BerriAI/litellm/blob/main/model_prices_and_context_window.json


    Note:
        - If completion_response is provided, the function extracts token information and the model name from it.
        - If completion_response is not provided, the function calculates token counts based on the model and input text.
        - The cost is calculated based on the model, prompt tokens, and completion tokens.
        - For certain models containing "togethercomputer" in the name, prices are based on the model size.
        - For un-mapped Replicate models, the cost is calculated based on the total time used for the request.
    """
    try:
        call_type = _infer_call_type(call_type, completion_response) or "completion"

        if (
            (call_type == "aimage_generation" or call_type == "image_generation")
            and model is not None
            and isinstance(model, str)
            and len(model) == 0
            and custom_llm_provider == "azure"
        ):
            model = "dall-e-2"  # for dall-e-2, azure expects an empty model name
        # Handle Inputs to completion_cost
        prompt_tokens = 0
        prompt_characters: int | None = None
        completion_tokens = 0
        completion_characters: int | None = None
        cache_creation_input_tokens: int | None = None
        cache_read_input_tokens: int | None = None
        audio_transcription_file_duration: float = 0.0
        cost_per_token_usage_object: Final[Usage | None] = _get_usage_object(completion_response=completion_response)
        rerank_billed_units: RerankBilledUnits | None = None

        # Extract service_tier from optional_params if not provided directly
        if service_tier is None and optional_params is not None:
            service_tier = optional_params.get("service_tier")

        service_tier = _normalize_service_tier(service_tier)

        # Extract service_tier from completion_response if not provided
        if service_tier is None and completion_response is not None:
            service_tier = _extract_service_tier(completion_response)

        service_tier = _normalize_service_tier(service_tier)

        # Extract service_tier from usage object if not provided
        if service_tier is None and cost_per_token_usage_object is not None:
            service_tier = _extract_service_tier(cost_per_token_usage_object)

        service_tier = _normalize_service_tier(service_tier)

        selected_model: Final = _select_model_name_for_cost_calc(
            model=model,
            completion_response=completion_response,
            custom_llm_provider=custom_llm_provider,
            custom_pricing=custom_pricing,
            base_model=base_model,
            router_model_id=router_model_id,
        )

        potential_model_names: Final = [
            selected_model,
            _get_response_model(completion_response),
        ]
        if model is not None:
            potential_model_names.append(model)

        for idx, model in enumerate(potential_model_names):
            try:
                if verbose_logger.isEnabledFor(logging.DEBUG):
                    verbose_logger.debug("selected model name for cost calculation: %s", model)

                if completion_response is not None and (
                    isinstance(completion_response, BaseModel) or isinstance(completion_response, dict)
                ):  # tts returns a custom class
                    if isinstance(completion_response, dict):
                        usage_obj: dict | Usage | None = completion_response.get("usage", {})
                    else:
                        usage_obj = getattr(completion_response, "usage", {})
                    if isinstance(usage_obj, BaseModel) and not _is_known_usage_objects(usage_obj=usage_obj):
                        _usage_for_dump = cast(BaseModel, usage_obj)
                        setattr(
                            completion_response,
                            "usage",
                            litellm.Usage(**_usage_for_dump.model_dump()),
                        )
                    if usage_obj is None:
                        _usage = {}
                    elif isinstance(usage_obj, BaseModel):
                        _usage = cast(BaseModel, usage_obj).model_dump()
                    else:
                        _usage = usage_obj

                    if litellm.AnthropicConfig.is_anthropic_usage_object(_usage):
                        _usage = (
                            litellm.AnthropicConfig()
                            .calculate_usage(usage_object=_usage, reasoning_content=None)
                            .model_dump()
                        )
                    elif ResponseAPILoggingUtils._is_response_api_usage(_usage):
                        _usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(
                            _usage
                        ).model_dump()
                    elif TranscriptionUsageObjectTransformation.is_transcription_usage_object(_usage):
                        tr_usage = TranscriptionUsageObjectTransformation.transform_transcription_usage_object(
                            cast(
                                TranscriptionUsageDurationObject | TranscriptionUsageTokensObject,
                                _usage,
                            )
                        )
                        if tr_usage is not None:
                            _usage = tr_usage.model_dump()
                    elif InteractionsUsageObjectTransformation.is_interactions_usage_object(_usage):
                        _usage = InteractionsUsageObjectTransformation.transform_interactions_usage_object(
                            _usage
                        ).model_dump()
                    else:
                        _usage = _usage

                    # get input/output tokens from completion_response
                    prompt_tokens = _usage.get("prompt_tokens", 0)
                    completion_tokens = _usage.get("completion_tokens", 0)
                    cache_creation_input_tokens = _usage.get("cache_creation_input_tokens", 0)
                    cache_read_input_tokens = _usage.get("cache_read_input_tokens", 0)
                    if (
                        "prompt_tokens_details" in _usage
                        and _usage["prompt_tokens_details"] != {}
                        and _usage["prompt_tokens_details"]
                    ):
                        prompt_tokens_details = _usage.get("prompt_tokens_details") or {}
                        cache_read_input_tokens = prompt_tokens_details.get("cached_tokens", 0)

                    total_time = getattr(completion_response, "_response_ms", 0)

                    hidden_params = getattr(completion_response, "_hidden_params", None)
                    if hidden_params is not None:
                        custom_llm_provider = hidden_params.get("custom_llm_provider", custom_llm_provider or None)
                        region_name = hidden_params.get("region_name", region_name)

                        # For Gemini/Vertex AI responses, trafficType is stored in
                        # provider_specific_fields.  Map it to the service_tier used
                        # by the cost key lookup (_priority / _flex suffixes) so that
                        # ON_DEMAND_PRIORITY requests are billed at priority prices.
                        if service_tier is None:
                            provider_specific = hidden_params.get("provider_specific_fields") or {}
                            raw_traffic_type = provider_specific.get("traffic_type")
                            if raw_traffic_type:
                                service_tier = _map_traffic_type_to_service_tier(raw_traffic_type)
                else:
                    if model is None:
                        raise ValueError(
                            f"Model is None and does not exist in passed completion_response. Passed completion_response={completion_response}, model={model}"
                        )
                    if len(messages) > 0:
                        prompt_tokens = token_counter(model=model, messages=messages)
                    elif len(prompt) > 0:
                        prompt_tokens = token_counter(model=model, text=prompt)
                    completion_tokens = token_counter(model=model, text=completion)

                # Handle A2A calls before model check - A2A doesn't require a model
                if call_type in _A2A_CALL_TYPES:
                    from litellm.a2a_protocol.cost_calculator import A2ACostCalculator

                    return A2ACostCalculator.calculate_a2a_cost(litellm_logging_obj=litellm_logging_obj)

                if model is None:
                    raise ValueError(
                        f"Model is None and does not exist in passed completion_response. Passed completion_response={completion_response}, model={model}"
                    )
                if custom_llm_provider is None:
                    try:
                        model, custom_llm_provider, _, _ = litellm.get_llm_provider(
                            model=model
                        )  # strip the llm provider from the model name -> for image gen cost calculation
                    except Exception as e:
                        verbose_logger.debug(
                            "litellm.cost_calculator.py::completion_cost() - Error inferring custom_llm_provider - %s",
                            e,
                        )
                if CostCalculatorUtils._call_type_has_image_response(call_type) and isinstance(
                    completion_response, ImageResponse
                ):
                    ### IMAGE GENERATION COST CALCULATION ###
                    return CostCalculatorUtils.route_image_generation_cost_calculator(
                        model=model,
                        custom_llm_provider=custom_llm_provider,
                        completion_response=completion_response,
                        quality=quality,
                        n=n,
                        size=size,
                        optional_params=optional_params,
                        call_type=call_type,
                    )
                elif call_type in _VIDEO_CALL_TYPES:
                    ### VIDEO GENERATION COST CALCULATION ###
                    # Extract custom model_info for deployment-specific pricing
                    _video_model_info: ModelInfo | None = None
                    if custom_pricing and litellm_logging_obj is not None:
                        _litellm_params = getattr(litellm_logging_obj, "litellm_params", None)
                        if _litellm_params is not None:
                            _video_model_info = next(
                                (
                                    model_info
                                    for _metadata_key in ("metadata", "litellm_metadata")
                                    if (model_info := (_litellm_params.get(_metadata_key) or {}).get("model_info"))
                                    is not None
                                ),
                                None,
                            )

                    usage_obj = getattr(completion_response, "usage", None)
                    duration_seconds: float | None = None
                    video_resolution: str | None = None
                    provider_reported_cost: float | None = None
                    if completion_response is not None and usage_obj:
                        # Handle both dict and Pydantic Usage object
                        if isinstance(usage_obj, dict):
                            duration_seconds = usage_obj.get("duration_seconds", None)
                            _vr = usage_obj.get("video_resolution", None)
                            provider_reported_cost = usage_obj.get("provider_reported_cost_usd", None)
                        else:
                            duration_seconds = getattr(usage_obj, "duration_seconds", None)
                            _vr = getattr(usage_obj, "video_resolution", None)
                            provider_reported_cost = getattr(usage_obj, "provider_reported_cost_usd", None)
                        if _vr is not None:
                            video_resolution = str(_vr).strip().lower()

                        if _video_model_info is None and provider_reported_cost is not None:
                            return float(provider_reported_cost)

                        if duration_seconds is not None:
                            # Calculate cost based on video duration using video-specific cost calculation
                            from litellm.llms.openai.cost_calculation import (
                                video_generation_cost,
                            )

                            return video_generation_cost(
                                model=model,
                                duration_seconds=duration_seconds,
                                custom_llm_provider=custom_llm_provider,
                                model_info=_video_model_info,
                                video_resolution=video_resolution,
                            )
                    # Fallback to default video cost calculation if no duration available
                    return default_video_cost_calculator(
                        model=model,
                        duration_seconds=0.0,  # Default to 0 if no duration available
                        custom_llm_provider=custom_llm_provider,
                        model_info=_video_model_info,
                        video_resolution=video_resolution,
                    )
                elif call_type in _SPEECH_CALL_TYPES:
                    prompt_characters = litellm.utils._count_characters(text=prompt)
                elif call_type in _TRANSCRIPTION_CALL_TYPES:
                    # Check _hidden_params first (duration stored there to
                    # avoid polluting the response body), then fall back to
                    # the response attribute (for verbose_json responses that
                    # naturally include duration from the provider).
                    _hidden = getattr(completion_response, "_hidden_params", {}) or {}
                    audio_transcription_file_duration = _hidden.get(
                        "audio_transcription_duration",
                        getattr(completion_response, "duration", 0.0),
                    )
                elif call_type in _RERANK_CALL_TYPES:
                    if completion_response is not None and isinstance(completion_response, RerankResponse):
                        meta_obj = completion_response.meta
                        if meta_obj is not None:
                            billed_units: RerankBilledUnits = meta_obj.get("billed_units") or {}
                        else:
                            billed_units = {}

                        rerank_billed_units = RerankBilledUnits(
                            search_units=billed_units.get("search_units"),
                            total_tokens=billed_units.get("total_tokens"),
                        )

                        search_units = billed_units.get("search_units") or 1  # cohere charges per request by default.
                        completion_tokens = search_units
                elif call_type in _SEARCH_CALL_TYPES:
                    from litellm.search import search_provider_cost_per_query

                    # Extract number_of_queries from optional_params or default to 1
                    number_of_queries = 1
                    if optional_params is not None:
                        # Check if query is a list (multiple queries)
                        query = optional_params.get("query")
                        if isinstance(query, list):
                            number_of_queries = len(query)
                        elif query is not None:
                            number_of_queries = 1

                    search_model = model or ""
                    if custom_llm_provider and "/" not in search_model:
                        # If model is like "tavily-search", construct "tavily/search" for cost lookup
                        search_model = f"{custom_llm_provider}/search"

                    (
                        prompt_cost,
                        completion_cost_result,
                    ) = search_provider_cost_per_query(
                        model=search_model,
                        custom_llm_provider=custom_llm_provider,
                        number_of_queries=number_of_queries,
                        optional_params=optional_params,
                    )

                    # Return the total cost (prompt_cost + completion_cost, but for search it's just prompt_cost)
                    _final_cost = prompt_cost + completion_cost_result

                    # Apply discount
                    original_cost = _final_cost
                    (
                        _final_cost,
                        discount_percent,
                        discount_amount,
                    ) = _apply_cost_discount(
                        base_cost=_final_cost,
                        custom_llm_provider=custom_llm_provider,
                    )

                    # Apply margin from module-level config if configured
                    (
                        _final_cost,
                        margin_percent,
                        margin_fixed_amount,
                        margin_total_amount,
                    ) = _apply_cost_margin(
                        base_cost=_final_cost,
                        custom_llm_provider=custom_llm_provider,
                    )

                    # Store cost breakdown in logging object if available
                    _store_cost_breakdown_in_logging_obj(
                        litellm_logging_obj=litellm_logging_obj,
                        prompt_tokens_cost_usd_dollar=prompt_cost,
                        completion_tokens_cost_usd_dollar=completion_cost_result,
                        cost_for_built_in_tools_cost_usd_dollar=0.0,
                        total_cost_usd_dollar=_final_cost,
                        original_cost=original_cost,
                        discount_percent=discount_percent,
                        discount_amount=discount_amount,
                        margin_percent=margin_percent,
                        margin_fixed_amount=margin_fixed_amount,
                        margin_total_amount=margin_total_amount,
                        service_tier=service_tier,
                        data_residency=data_residency,
                    )

                    return _final_cost
                elif call_type == _AREALTIME_CALL_TYPE and isinstance(
                    completion_response, LiteLLMRealtimeStreamLoggingObject
                ):
                    if cost_per_token_usage_object is None or custom_llm_provider is None:
                        raise ValueError(
                            f"usage object and custom_llm_provider must be provided for realtime stream cost calculation. Got cost_per_token_usage_object={cost_per_token_usage_object}, custom_llm_provider={custom_llm_provider}"
                        )
                    return handle_realtime_stream_cost_calculation(
                        results=completion_response.results,
                        combined_usage_object=cost_per_token_usage_object,
                        custom_llm_provider=custom_llm_provider,
                        litellm_model_name=model,
                        data_residency=data_residency,
                        litellm_logging_obj=litellm_logging_obj,
                    )
                elif call_type == _MCP_CALL_TYPE:
                    from litellm.proxy._experimental.mcp_server.cost_calculator import (
                        MCPCostCalculator,
                    )

                    return MCPCostCalculator.calculate_mcp_tool_call_cost(litellm_logging_obj=litellm_logging_obj)
                # Calculate cost based on prompt_tokens, completion_tokens
                if (
                    "togethercomputer" in model or "together_ai" in model or custom_llm_provider == "together_ai"
                ) and not has_together_registry_pricing(model, litellm.model_cost):
                    model = get_model_params_and_category(model, call_type=CallTypes(call_type))

                # replicate llms are calculate based on time for request running
                # see https://replicate.com/pricing
                elif (model in litellm.replicate_models or "replicate" in model) and model not in litellm.model_cost:
                    # for unmapped replicate model, default to replicate's time tracking logic
                    return get_replicate_completion_pricing(completion_response, total_time)

                if model is None:
                    raise ValueError(
                        f"Model is None and does not exist in passed completion_response. Passed completion_response={completion_response}, model={model}"
                    )

                if custom_llm_provider is not None and custom_llm_provider == "vertex_ai":
                    # Calculate the prompt characters + response characters
                    if len(messages) > 0:
                        prompt_string = litellm.utils.get_formatted_prompt(
                            data={"messages": messages}, call_type="completion"
                        )

                        prompt_characters = litellm.utils._count_characters(text=prompt_string)
                    if completion_response is not None and isinstance(completion_response, ModelResponse):
                        completion_string = litellm.utils.get_response_string(response_obj=completion_response)
                        completion_characters = litellm.utils._count_characters(text=completion_string)

                # Get the original request model for router detection
                request_model_for_cost = None
                if litellm_logging_obj is not None:
                    request_model_for_cost = litellm_logging_obj.model

                (
                    prompt_tokens_cost_usd_dollar,
                    completion_tokens_cost_usd_dollar,
                ) = cost_per_token(
                    model=model,
                    prompt_tokens=prompt_tokens or 0,
                    completion_tokens=completion_tokens or 0,
                    custom_llm_provider=custom_llm_provider,
                    response_time_ms=total_time,
                    region_name=region_name,
                    custom_cost_per_second=custom_cost_per_second,
                    custom_cost_per_token=custom_cost_per_token,
                    prompt_characters=prompt_characters,
                    completion_characters=completion_characters,
                    cache_creation_input_tokens=cache_creation_input_tokens,
                    cache_read_input_tokens=cache_read_input_tokens,
                    usage_object=cost_per_token_usage_object,
                    call_type=call_type,
                    audio_transcription_file_duration=audio_transcription_file_duration,
                    rerank_billed_units=rerank_billed_units,
                    service_tier=service_tier,
                    data_residency=data_residency,
                    vertex_location=vertex_location,
                    response=completion_response,
                    request_model=request_model_for_cost,
                )

                # Get additional costs from provider (e.g., routing fees, infrastructure costs)
                if custom_llm_provider == "azure_ai":
                    model_for_additional_costs = request_model_for_cost
                    if completion_response is not None:
                        hidden_params = getattr(completion_response, "_hidden_params", None) or {}
                        hidden_model = hidden_params.get("model") or hidden_params.get("litellm_model_name")
                        if (
                            hidden_model
                            and (
                                "model_router" in (hidden_model or "").lower()
                                or "model-router" in (hidden_model or "").lower()
                            )
                            or model_for_additional_costs is None
                        ):
                            model_for_additional_costs = hidden_model
                    if model_for_additional_costs is None:
                        model_for_additional_costs = model
                    additional_costs = _get_additional_costs(
                        model=model_for_additional_costs,
                        custom_llm_provider=custom_llm_provider,
                        prompt_tokens=prompt_tokens or 0,
                        completion_tokens=completion_tokens or 0,
                    )
                else:
                    additional_costs = None

                _final_cost = prompt_tokens_cost_usd_dollar + completion_tokens_cost_usd_dollar
                cost_for_built_in_tools = StandardBuiltInToolCostTracking.get_cost_for_built_in_tools(
                    model=model,
                    response_object=completion_response,
                    usage=cost_per_token_usage_object,
                    standard_built_in_tools_params=standard_built_in_tools_params,
                    custom_llm_provider=custom_llm_provider,
                )
                _final_cost += cost_for_built_in_tools
                if additional_costs:
                    _final_cost += sum(additional_costs.values())

                original_cost = _final_cost
                if litellm.cost_discount_config:
                    (
                        _final_cost,
                        discount_percent,
                        discount_amount,
                    ) = _apply_cost_discount(
                        base_cost=_final_cost,
                        custom_llm_provider=custom_llm_provider,
                    )
                else:
                    discount_percent = 0.0
                    discount_amount = 0.0

                # Apply margin from module-level config if configured
                if litellm.cost_margin_config:
                    (
                        _final_cost,
                        margin_percent,
                        margin_fixed_amount,
                        margin_total_amount,
                    ) = _apply_cost_margin(
                        base_cost=_final_cost,
                        custom_llm_provider=custom_llm_provider,
                    )
                else:
                    margin_percent = 0.0
                    margin_fixed_amount = 0.0
                    margin_total_amount = 0.0

                # Store cost breakdown in logging object if available
                if litellm_logging_obj is not None:
                    _reasoning_cost: float | None = None
                    _cache_read_cost: float | None = None
                    _cache_creation_cost: float | None = None
                    if cost_per_token_usage_object is not None and model:
                        _breakdown_provider: str | None = (
                            custom_llm_provider if isinstance(custom_llm_provider, str) else None
                        )
                        _token_type_breakdown = get_token_type_cost_breakdown(
                            model=model,
                            custom_llm_provider=_breakdown_provider,
                            usage=cost_per_token_usage_object,
                            service_tier=service_tier,
                            data_residency=data_residency,
                            vertex_location=vertex_location,
                        )
                        _reasoning_cost = _token_type_breakdown.reasoning_cost
                        _cache_read_cost = _token_type_breakdown.cache_read_cost
                        _cache_creation_cost = _token_type_breakdown.cache_creation_cost
                    _store_cost_breakdown_in_logging_obj(
                        litellm_logging_obj=litellm_logging_obj,
                        prompt_tokens_cost_usd_dollar=prompt_tokens_cost_usd_dollar,
                        completion_tokens_cost_usd_dollar=completion_tokens_cost_usd_dollar,
                        cost_for_built_in_tools_cost_usd_dollar=cost_for_built_in_tools,
                        total_cost_usd_dollar=_final_cost,
                        original_cost=original_cost,
                        additional_costs=additional_costs,
                        discount_percent=discount_percent,
                        discount_amount=discount_amount,
                        margin_percent=margin_percent,
                        margin_fixed_amount=margin_fixed_amount,
                        margin_total_amount=margin_total_amount,
                        cache_read_cost=_cache_read_cost,
                        cache_creation_cost=_cache_creation_cost,
                        reasoning_cost=_reasoning_cost,
                        service_tier=service_tier,
                        data_residency=data_residency,
                        vertex_location=vertex_location,
                    )

                return _final_cost
            except Exception as e:
                verbose_logger.debug(
                    "litellm.cost_calculator.py::completion_cost() - Error calculating cost for model=%s - %s", model, e
                )
                if idx == len(potential_model_names) - 1:
                    raise e
        raise Exception(f"Unable to calculat cost for received potential model names - {potential_model_names}")
    except Exception as e:
        raise e


def get_response_cost_from_hidden_params(
    hidden_params: dict | BaseModel,
) -> float | None:
    if isinstance(hidden_params, BaseModel):
        _hidden_params_dict = cast(BaseModel, hidden_params).model_dump()
    else:
        _hidden_params_dict = hidden_params

    additional_headers: Final = _hidden_params_dict.get("additional_headers", {})
    if additional_headers and "llm_provider-x-litellm-response-cost" in additional_headers:
        response_cost: Final = additional_headers["llm_provider-x-litellm-response-cost"]
        if response_cost is None:
            return None
        return float(additional_headers["llm_provider-x-litellm-response-cost"])
    return None


def response_cost_calculator(
    response_object: ModelResponse
    | EmbeddingResponse
    | ImageResponse
    | TranscriptionResponse
    | TextCompletionResponse
    | HttpxBinaryResponseContent
    | RerankResponse
    | ResponsesAPIResponse
    | LiteLLMRealtimeStreamLoggingObject
    | OpenAIModerationResponse
    | Response
    | SearchResponse,
    model: str,
    custom_llm_provider: str | None,
    call_type: Literal[
        "embedding",
        "aembedding",
        "completion",
        "acompletion",
        "atext_completion",
        "text_completion",
        "image_generation",
        "aimage_generation",
        "moderation",
        "amoderation",
        "atranscription",
        "transcription",
        "aspeech",
        "speech",
        "rerank",
        "arerank",
        "search",
        "asearch",
    ],
    optional_params: dict,
    cache_hit: bool | None = None,
    base_model: str | None = None,
    custom_pricing: bool | None = None,
    prompt: str = "",
    standard_built_in_tools_params: StandardBuiltInToolsParams | None = None,
    litellm_model_name: str | None = None,
    router_model_id: str | None = None,
    litellm_logging_obj: LitellmLoggingObject | None = None,
    ### SERVICE TIER ###
    service_tier: str | None = None,  # for OpenAI service tier pricing
    ### DATA RESIDENCY ###
    data_residency: str | None = None,  # for OpenAI regional-processing uplift (e.g. "eu", "us")
    ### VERTEX LOCATION ###
    vertex_location: str | None = None,  # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global")
) -> float:
    """
    Returns
    - float or None: cost of response
    """
    try:
        response_cost: float = 0.0
        if cache_hit is not None and cache_hit is True:
            response_cost = 0.0
        else:
            if isinstance(response_object, BaseModel):
                if hasattr(response_object, "_hidden_params"):
                    response_object._hidden_params["optional_params"] = optional_params
                    provider_response_cost: Final = get_response_cost_from_hidden_params(response_object._hidden_params)
                    if provider_response_cost is not None:
                        return provider_response_cost

            response_cost = completion_cost(
                completion_response=response_object,
                model=model,
                call_type=call_type,
                custom_llm_provider=custom_llm_provider,
                optional_params=optional_params,
                custom_pricing=custom_pricing,
                base_model=base_model,
                prompt=prompt,
                standard_built_in_tools_params=standard_built_in_tools_params,
                litellm_model_name=litellm_model_name,
                router_model_id=router_model_id,
                litellm_logging_obj=litellm_logging_obj,
                service_tier=service_tier,
                data_residency=data_residency,
                vertex_location=vertex_location,
            )
        return response_cost
    except Exception as e:
        raise e


def ocr_cost(
    model: str,
    custom_llm_provider: str | None,
    response: object | None = None,
) -> tuple[float, float]:
    """
    Args:
        model: str - model name
        custom_llm_provider: Optional[str] - custom LLM provider
        response: Optional[Any] - response object

    Returns:
        Tuple[float, float]: cost of OCR processing

        (Parent function requires a tuple, so we return a tuple. Cost is only in the first element.)
    """
    from litellm.llms.base_llm.ocr.transformation import OCRResponse

    #########################################################
    # validate it's an OCR response
    #########################################################
    if response is None or not isinstance(response, OCRResponse):
        raise ValueError(f"response must be of type OCRResponse got type={type(response)}")

    if response.usage_info is None:
        raise ValueError("OCR response usage_info is None")

    try:
        model_info: ModelInfo | None = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
    except Exception:
        model_info = None

    credits: Final = getattr(response.usage_info, "credits", None)
    cost_per_credit = None
    if model_info is not None:
        cost_per_credit = model_info.get("ocr_cost_per_credit")
    if credits is not None and cost_per_credit is not None:
        return cost_per_credit * credits, 0.0

    ocr_cost_per_page: Final = model_info.get("ocr_cost_per_page") if model_info is not None else None
    annotation_cost_per_page: Final = model_info.get("annotation_cost_per_page") if model_info is not None else None
    annotation_rate: Final = annotation_cost_per_page if annotation_cost_per_page is not None else ocr_cost_per_page

    pages_processed: Final = response.usage_info.pages_processed
    annotation_pages: Final = response.usage_info.pages_processed_annotation or 0
    has_billable_annotation_pages: Final = annotation_rate is not None and annotation_pages > 0

    if pages_processed is None and not has_billable_annotation_pages:
        if cost_per_credit is not None or ocr_cost_per_page is None:
            # Surface missing usage data instead of silently under-reporting
            # cost. The previous behavior raised ValueError; we now return 0.0
            # for credit-priced or unpriced models, so log a warning to keep
            # the regression visible to operators.
            verbose_logger.warning(
                "OCR cost: model=%s custom_llm_provider=%s response.usage_info."
                "pages_processed is None and credits=%s; returning 0.0 cost.",
                model,
                custom_llm_provider,
                credits,
            )
            return 0.0, 0.0
        raise ValueError("OCR response pages_processed is None")

    if ocr_cost_per_page is None and not has_billable_annotation_pages:
        # No per-page pricing configured. Either the model is on credit-based
        # pricing (and credits weren't returned, so the credit branch above did
        # not match) or the model has no OCR pricing entry at all. Surface a
        # warning so that missing pricing entries are visible rather than
        # silently producing zero cost for billable usage.
        verbose_logger.warning(
            "OCR cost: model=%s custom_llm_provider=%s reported "
            "pages_processed=%s but no ocr_cost_per_page is configured; "
            "returning 0.0 cost.",
            model,
            custom_llm_provider,
            pages_processed,
        )
        return 0.0, 0.0

    ocr_pages_cost: Final = (ocr_cost_per_page or 0.0) * (pages_processed or 0)
    annotation_pages_cost: Final = (annotation_rate or 0.0) * annotation_pages
    return ocr_pages_cost + annotation_pages_cost, 0.0


def vector_store_search_cost(
    model: str | None,
    custom_llm_provider: str,
    response: VectorStoreSearchResponse,
) -> tuple[float, float]:
    """
    Returns
    - float or None: cost of vector store search
    """
    api_type: str | None = None
    if custom_llm_provider is None:
        custom_llm_provider = "openai"

    if model is not None and "/" in model:
        api_type, custom_llm_provider, _, _ = litellm.get_llm_provider(
            model=model,
        )

    config: Final = ProviderConfigManager.get_provider_vector_stores_config(
        provider=LlmProviders(custom_llm_provider),
        api_type=api_type,
    )

    if config is None:
        verbose_logger.debug("Vector store search is not supported for %s", custom_llm_provider)
        return 0.0, 0.0

    return config.calculate_vector_store_cost(
        response=response,
    )


def rerank_cost(
    model: str,
    custom_llm_provider: str | None,
    billed_units: RerankBilledUnits | None = None,
) -> tuple[float, float]:
    """
    Returns
    - float or None: cost of response OR none if error.
    """
    _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model, custom_llm_provider=custom_llm_provider)

    try:
        config: Final = ProviderConfigManager.get_provider_rerank_config(
            model=model,
            api_base=None,
            present_version_params=[],
            provider=LlmProviders(custom_llm_provider),
        )

        try:
            model_info: ModelInfo | None = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
        except Exception:
            model_info = None

        return config.calculate_rerank_cost(
            model=model,
            custom_llm_provider=custom_llm_provider,
            billed_units=billed_units,
            model_info=model_info,
        )
    except Exception as e:
        raise e


def transcription_cost(model: str, custom_llm_provider: str | None, duration: float) -> tuple[float, float]:
    return openai_cost_per_second(model=model, custom_llm_provider=custom_llm_provider, duration=duration)


def default_image_cost_calculator(
    model: str,
    custom_llm_provider: str | None = None,
    quality: str | None = None,
    n: int | None = 1,  # Default to 1 image
    size: str | None = "1024-x-1024",  # OpenAI default
    optional_params: dict | None = None,
) -> float:
    """
    Default image cost calculator for image generation

    Args:
        model (str): Model name
        image_response (ImageResponse): Response from image generation
        quality (Optional[str]): Image quality setting
        n (Optional[int]): Number of images generated
        size (Optional[str]): Image size (e.g. "1024x1024" or "1024-x-1024")

    Returns:
        float: Cost in USD for the image generation

    Raises:
        Exception: If model pricing not found in cost map
    """
    # Standardize size format to use "-x-"
    size_str: str = size or "1024-x-1024"
    size_str = size_str.replace("x", "-x-") if "x" in size_str and "-x-" not in size_str else size_str

    # Parse dimensions
    height, width = map(int, size_str.split("-x-"))

    # Build model names for cost lookup
    base_model_name = f"{size_str}/{model}"
    model_name_without_custom_llm_provider: str | None = None
    if custom_llm_provider and model.startswith(f"{custom_llm_provider}/"):
        model_name_without_custom_llm_provider = model.replace(f"{custom_llm_provider}/", "")
        base_model_name = f"{custom_llm_provider}/{size_str}/{model_name_without_custom_llm_provider}"
    model_name_with_quality: Final = f"{quality}/{base_model_name}" if quality else base_model_name

    # gpt-image-1 models use low, medium, high quality. If user did not specify quality, use medium fot gpt-image-1 model family
    model_name_with_v2_quality: Final = f"{ImageGenerationRequestQuality.HIGH.value}/{base_model_name}"

    verbose_logger.debug("Looking up cost for models: %s, %s", model_name_with_quality, base_model_name)

    model_without_provider: Final = f"{size_str}/{model.split('/')[-1]}"
    model_with_quality_without_provider = f"{quality}/{model_without_provider}" if quality else model_without_provider

    # Try model with quality first, fall back to base model name
    cost_info: dict | None = None
    models_to_check: Final[list[str | None]] = [
        model_name_with_quality,
        base_model_name,
        model_name_with_v2_quality,
        model_with_quality_without_provider,
        model_without_provider,
        model,
        model_name_without_custom_llm_provider,
    ]
    for _model in models_to_check:
        if _model is not None and _model in litellm.model_cost:
            cost_info = litellm.model_cost[_model]
            break
    if cost_info is None:
        raise Exception(f"Model not found in cost map. Tried checking {models_to_check}")

    # Priority 1: Use per-image pricing if available (for gpt-image-1 and similar models)
    if "input_cost_per_image" in cost_info and cost_info["input_cost_per_image"] is not None:
        return cost_info["input_cost_per_image"] * n
    # Priority 2: Fall back to per-pixel pricing for backward compatibility
    elif "input_cost_per_pixel" in cost_info and cost_info["input_cost_per_pixel"] is not None:
        return cost_info["input_cost_per_pixel"] * height * width * n
    else:
        raise Exception(f"No pricing information found for model {model}. Tried checking {models_to_check}")


def default_video_cost_calculator(
    model: str,
    duration_seconds: float,
    custom_llm_provider: str | None = None,
    model_info: ModelInfo | None = None,
    video_resolution: str | None = None,
) -> float:
    """
    Default video cost calculator for video generation

    Args:
        model (str): Model name
        duration_seconds (float): Duration of the generated video in seconds
        custom_llm_provider (Optional[str]): Custom LLM provider
        model_info (Optional[ModelInfo]): Deployment-level model info containing
            custom video pricing. When provided, used before falling back to
            the global litellm.model_cost lookup.
        video_resolution (Optional[str]): From usage (e.g. ``720p``, ``1080p``) for tiered per-second pricing.

    Returns:
        float: Cost in USD for the video generation

    Raises:
        Exception: If model pricing not found in cost map
    """
    # Use custom model_info pricing if provided (deployment-specific pricing)
    cost_info: dict | None = None
    if model_info is not None:
        cost_info = dict(model_info)
    else:
        # Build model names for cost lookup
        base_model_name = model
        model_name_without_custom_llm_provider: str | None = None
        if custom_llm_provider and model.startswith(f"{custom_llm_provider}/"):
            model_name_without_custom_llm_provider = model.replace(f"{custom_llm_provider}/", "")
            base_model_name = f"{custom_llm_provider}/{model_name_without_custom_llm_provider}"

        verbose_logger.debug("Looking up cost for video model: %s", base_model_name)

        model_without_provider: Final = model.split("/")[-1]

        # Try model with provider first, fall back to base model name
        models_to_check: Final[list[str | None]] = [
            base_model_name,
            model,
            model_without_provider,
            model_name_without_custom_llm_provider,
        ]
        for _model in models_to_check:
            if _model is not None and _model in litellm.model_cost:
                cost_info = litellm.model_cost[_model]
                break

        # If still not found, try with custom_llm_provider prefix
        if cost_info is None and custom_llm_provider:
            prefixed_model: Final = f"{custom_llm_provider}/{model}"
            if prefixed_model in litellm.model_cost:
                cost_info = litellm.model_cost[prefixed_model]

    if cost_info is None:
        raise Exception(f"Model not found in cost map for model={model}")

    # Check for video-specific cost per second first
    video_cost_per_second: Final = cost_info.get("output_cost_per_video_per_second")
    if video_cost_per_second is not None:
        return video_cost_per_second * duration_seconds

    output_cost_per_second: Final = _video_output_cost_per_second(cost_info, video_resolution)
    if output_cost_per_second is not None:
        return output_cost_per_second * duration_seconds

    # If no cost information found, return 0
    verbose_logger.info(
        "No cost information found for video model %s. Please add pricing to model_prices_and_context_window.json",
        model,
    )
    return 0.0


def batch_cost_calculator(
    usage: Usage,
    model: str,
    custom_llm_provider: str | None = None,
    model_info: ModelInfo | None = None,
    data_residency: str | None = None,
) -> tuple[float, float]:
    """
    Calculate the cost of a batch job.

    Args:
        model_info: Optional deployment-level model info containing custom
            batch pricing (e.g. input_cost_per_token_batches). When provided,
            skips the global litellm.get_model_info() lookup so that
            deployment-specific pricing is used.
    """

    _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model, custom_llm_provider=custom_llm_provider)

    verbose_logger.debug(
        "Calculating batch cost per token. model=%s, custom_llm_provider=%s",
        model,
        custom_llm_provider,
    )

    if model_info is None:
        try:
            model_info = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
        except Exception:
            model_info = None
    elif not any(
        model_info.get(k) is not None
        for k in (
            "input_cost_per_token_batches",
            "input_cost_per_token",
            "output_cost_per_token_batches",
            "output_cost_per_token",
        )
    ):
        # model_info was provided (e.g. deployment metadata with only id/db_model)
        # but carries no pricing fields. Fall back to the global pricing table so
        # that standard model pricing is used instead of silently returning $0.
        try:
            global_info: Final = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
            if global_info:
                model_info = global_info
        except Exception:
            pass

    if not model_info:
        return 0.0, 0.0

    input_cost_per_token_batches: Final = model_info.get("input_cost_per_token_batches")
    input_cost_per_token: Final = model_info.get("input_cost_per_token")
    output_cost_per_token_batches: Final = model_info.get("output_cost_per_token_batches")
    output_cost_per_token: Final = model_info.get("output_cost_per_token")
    total_prompt_cost = 0.0
    total_completion_cost = 0.0
    if input_cost_per_token_batches is not None:
        total_prompt_cost = usage.prompt_tokens * input_cost_per_token_batches
    elif input_cost_per_token:
        details: Final = parse_prompt_tokens_details(usage)
        cache_read_tokens: Final = details["cache_hit_tokens"]
        cache_creation_tokens: Final = details["cache_creation_tokens"]

        # Subtract cached tokens from prompt_tokens before calculating cost
        # Fixes issue where cached tokens are being charged again
        base_input_tokens: Final = get_billable_input_tokens(usage) - cache_creation_tokens
        total_prompt_cost = (
            base_input_tokens * (input_cost_per_token) / 2
        )  # batch cost is usually half of the regular token cost

        # Add cache read cost if applicable
        cache_read_cost_key: Final = _get_service_tier_cost_key("cache_read_input_token_cost", None)
        total_prompt_cost += calculate_cost_component(model_info, cache_read_cost_key, cache_read_tokens) / 2

        cache_creation_cost: Final = model_info.get("cache_creation_input_token_cost") or input_cost_per_token
        total_prompt_cost += cache_creation_tokens * cache_creation_cost / 2
    if output_cost_per_token_batches is not None:
        total_completion_cost = usage.completion_tokens * output_cost_per_token_batches
    elif output_cost_per_token:
        total_completion_cost = (
            usage.completion_tokens * (output_cost_per_token) / 2
        )  # batch cost is usually half of the regular token cost

    uplift: Final = _get_regional_uplift_multiplier(model_info, data_residency)
    if uplift != 1.0:
        total_prompt_cost *= uplift
        total_completion_cost *= uplift

    return total_prompt_cost, total_completion_cost


def _attribute_value(obj: object, name: str) -> object:
    return getattr(obj, name)


def _summable_prompt_token_fields(prompt_tokens_details: BaseModel) -> list[str]:
    field_names: Final = list(type(prompt_tokens_details).model_fields)
    if getattr(prompt_tokens_details, "cache_write_tokens", None) is None:
        return field_names
    return [attr for attr in field_names if attr != "cache_creation_tokens"]


class BaseTokenUsageProcessor:
    @staticmethod
    def combine_usage_objects(usage_objects: list[Usage]) -> Usage:
        """
        Combine multiple Usage objects into a single Usage object, checking model keys for nested values.
        """
        from litellm.types.utils import (
            CompletionTokensDetailsWrapper,
            PromptTokensDetailsWrapper,
            Usage,
        )

        combined: Final = Usage()

        # Sum basic token counts
        for usage in usage_objects:
            # Handle direct attributes by checking what exists in the model
            for attr in dir(usage):
                if not attr.startswith("_") and not callable(_attribute_value(usage, attr)):
                    current_val = getattr(combined, attr, 0)
                    new_val = getattr(usage, attr, 0)
                    if (
                        new_val is not None
                        and isinstance(new_val, (int, float))
                        and isinstance(current_val, (int, float))
                    ):
                        setattr(combined, attr, current_val + new_val)
            # Handle nested prompt_tokens_details
            if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details:
                if not hasattr(combined, "prompt_tokens_details") or not combined.prompt_tokens_details:
                    combined.prompt_tokens_details = PromptTokensDetailsWrapper()

                # Check what keys exist in the model's prompt_tokens_details
                # Access model_fields on the class, not the instance, to avoid Pydantic 2.11+ deprecation warnings
                for attr in _summable_prompt_token_fields(usage.prompt_tokens_details):
                    if (
                        hasattr(usage.prompt_tokens_details, attr)
                        and not attr.startswith("_")
                        and not callable(_attribute_value(usage.prompt_tokens_details, attr))
                    ):
                        current_val = getattr(combined.prompt_tokens_details, attr, 0) or 0
                        new_val = getattr(usage.prompt_tokens_details, attr, 0) or 0
                        if new_val is not None and isinstance(new_val, (int, float)):
                            setattr(
                                combined.prompt_tokens_details,
                                attr,
                                current_val + new_val,
                            )

            # Handle nested completion_tokens_details
            if hasattr(usage, "completion_tokens_details") and usage.completion_tokens_details:
                if not hasattr(combined, "completion_tokens_details") or not combined.completion_tokens_details:
                    combined.completion_tokens_details = CompletionTokensDetailsWrapper()

                # Check what keys exist in the model's completion_tokens_details
                # Access model_fields on the class, not the instance, to avoid Pydantic 2.11+ deprecation warnings
                for attr in type(usage.completion_tokens_details).model_fields:
                    if not attr.startswith("_") and not callable(
                        _attribute_value(usage.completion_tokens_details, attr)
                    ):
                        current_val = getattr(combined.completion_tokens_details, attr, 0) or 0
                        new_val = getattr(usage.completion_tokens_details, attr, 0) or 0
                        if isinstance(new_val, (int, float)):
                            setattr(
                                combined.completion_tokens_details,
                                attr,
                                current_val + new_val,
                            )

        return combined


class RealtimeAPITokenUsageProcessor(BaseTokenUsageProcessor):
    @staticmethod
    def collect_usage_from_realtime_stream_results(
        results: OpenAIRealtimeStreamList,
    ) -> list[Usage]:
        """
        Collect usage from realtime stream results
        """
        response_done_events: Final[list[OpenAIRealtimeStreamResponseBaseObject]] = cast(
            list[OpenAIRealtimeStreamResponseBaseObject],
            [result for result in results if result["type"] == "response.done"],
        )
        usage_objects: Final[list[Usage]] = []
        for result in response_done_events:
            usage_object = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(
                result["response"].get("usage", {})
            )
            usage_objects.append(usage_object)
        return usage_objects

    @staticmethod
    def collect_and_combine_usage_from_realtime_stream_results(
        results: OpenAIRealtimeStreamList,
    ) -> Usage:
        """
        Collect and combine usage from realtime stream results
        """
        collected_usage_objects = RealtimeAPITokenUsageProcessor.collect_usage_from_realtime_stream_results(results)
        combined_usage_object: Final = RealtimeAPITokenUsageProcessor.combine_usage_objects(collected_usage_objects)
        return combined_usage_object

    @staticmethod
    def create_logging_realtime_object(
        usage: Usage, results: OpenAIRealtimeStreamList
    ) -> LiteLLMRealtimeStreamLoggingObject:
        return LiteLLMRealtimeStreamLoggingObject(
            usage=usage,
            results=results,
        )


_TRANSCRIPTION_COMPLETED_EVENT_TYPE: Final = "conversation.item.input_audio_transcription.completed"


def _candidate_realtime_token_costs(
    model_name: str,
    combined_usage_object: Usage,
    custom_llm_provider: str,
    data_residency: str | None,
) -> tuple[float, float] | None:
    try:
        return generic_cost_per_token(
            model=model_name,
            usage=combined_usage_object,
            custom_llm_provider=custom_llm_provider,
            data_residency=data_residency,
        )
    except Exception:
        return None


def _cost_map_entry_declares_pricing(model_name: str, custom_llm_provider: str) -> bool:
    entries: Final = (
        litellm.model_cost.get(model_name),
        litellm.model_cost.get(f"{custom_llm_provider}/{model_name}"),
    )
    return any(
        entry is not None and any("cost_per" in field and value is not None for field, value in entry.items())
        for entry in entries
    )


def _first_priced_realtime_token_costs(
    potential_model_names: Sequence[str | None],
    combined_usage_object: Usage,
    custom_llm_provider: str,
    data_residency: str | None,
) -> tuple[float, float]:
    candidate_costs: Final = (
        (model_name, costs)
        for model_name in potential_model_names
        if model_name is not None
        and (
            costs := _candidate_realtime_token_costs(
                model_name=model_name,
                combined_usage_object=combined_usage_object,
                custom_llm_provider=custom_llm_provider,
                data_residency=data_residency,
            )
        )
        is not None
    )
    return next(
        (
            costs
            for model_name, costs in candidate_costs
            if sum(costs) > 0 or _cost_map_entry_declares_pricing(model_name, custom_llm_provider)
        ),
        (0.0, 0.0),
    )


def handle_realtime_stream_cost_calculation(
    results: OpenAIRealtimeStreamList,
    combined_usage_object: Usage,
    custom_llm_provider: str,
    litellm_model_name: str,
    data_residency: str | None = None,
    litellm_logging_obj: LitellmLoggingObject | None = None,
) -> float:
    """
    Handles the cost calculation for realtime stream responses.

    Pick the 'response.done' events. Calculate total cost across all 'response.done' events.

    Args:
        results: A list of OpenAIRealtimeStreamBaseObject objects
    """
    received_model = None
    potential_model_names: Final = []
    for result in results:
        if result["type"] == "session.created":
            received_model = cast(OpenAIRealtimeStreamSessionEvents, result)["session"].get("model", None)
            potential_model_names.append(received_model)

    potential_model_names.append(litellm_model_name)
    input_cost_per_token, output_cost_per_token = _first_priced_realtime_token_costs(
        potential_model_names=potential_model_names,
        combined_usage_object=combined_usage_object,
        custom_llm_provider=custom_llm_provider,
        data_residency=data_residency,
    )
    transcription_cost: Final = (
        handle_realtime_transcription_cost_calculation(
            results=results,
            custom_llm_provider=custom_llm_provider,
            litellm_model_name=litellm_model_name,
        )
        if any(r.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE for r in results)
        else 0.0
    )
    total_cost: Final = input_cost_per_token + output_cost_per_token + transcription_cost

    _store_cost_breakdown_in_logging_obj(
        litellm_logging_obj=litellm_logging_obj,
        prompt_tokens_cost_usd_dollar=input_cost_per_token,
        completion_tokens_cost_usd_dollar=output_cost_per_token,
        cost_for_built_in_tools_cost_usd_dollar=0.0,
        total_cost_usd_dollar=total_cost,
        additional_costs={"transcription_cost": transcription_cost} if transcription_cost > 0 else None,
        data_residency=data_residency,
    )

    return total_cost


def handle_realtime_transcription_cost_calculation(
    results: OpenAIRealtimeStreamList,
    custom_llm_provider: str,
    litellm_model_name: str,
) -> float:
    """
    Cost for realtime transcription sessions (e.g. gpt-realtime-whisper).

    Transcription sessions emit no `response.done` events; instead each
    `conversation.item.input_audio_transcription.completed` event carries a
    `usage` object billed by the ASR model. The usage is one of:
      - {"type": "duration", "seconds": <float>}  → priced via input_cost_per_second
      - {"type": "tokens", "input_tokens": ...}    → priced via input/audio token cost
    """
    completed_events: Final = [
        cast(dict, result) for result in results if result.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE
    ]
    if not completed_events:
        return 0.0

    model_name: Final = _get_transcription_model_name_from_results(results) or litellm_model_name
    try:
        model_info = litellm.get_model_info(model=model_name, custom_llm_provider=custom_llm_provider)
    except Exception:
        model_info = None

    total_cost = 0.0
    for event in completed_events:
        usage = event.get("usage") or {}
        total_cost += _transcription_usage_cost(usage, model_info)
    return total_cost


def _get_transcription_model_name_from_results(
    results: OpenAIRealtimeStreamList,
) -> str | None:
    """Resolve the ASR model from a transcription_session.* / session.* event."""
    for result in results:
        if result.get("type") in (
            "transcription_session.created",
            "transcription_session.updated",
            "session.created",
            "session.updated",
        ):
            session = cast(dict, result).get("session", {}) or {}
            transcription = ((session.get("audio", {}) or {}).get("input", {}) or {}).get(
                "transcription", {}
            ) or session.get("input_audio_transcription", {})
            model = (transcription or {}).get("model") or session.get("model")
            if model:
                return model
    return None


def _transcription_usage_cost(usage: dict, model_info: ModelInfo | None) -> float:
    if model_info is None:
        return 0.0
    usage_type: Final = usage.get("type")
    if usage_type == "duration":
        seconds: Final = usage.get("seconds") or 0.0
        per_second: Final = model_info.get("input_cost_per_second") or 0.0
        return float(seconds) * float(per_second)
    if usage_type == "tokens":
        input_token_details: Final = usage.get("input_token_details") or {}
        audio_tokens: Final = input_token_details.get("audio_tokens") or 0
        text_tokens: Final = input_token_details.get("text_tokens") or 0
        output_tokens: Final = usage.get("output_tokens") or 0
        audio_cost: Final = float(audio_tokens) * float(
            model_info.get("input_cost_per_audio_token") or model_info.get("input_cost_per_token") or 0.0
        )
        text_cost: Final = float(text_tokens) * float(model_info.get("input_cost_per_token") or 0.0)
        output_cost: Final = float(output_tokens) * float(model_info.get("output_cost_per_token") or 0.0)
        return audio_cost + text_cost + output_cost
    return 0.0
