"""``CustomLogger`` adapter on the OpenTelemetry span engine."""

from collections import OrderedDict
from collections.abc import Callable, Iterator, Mapping, Sequence
from contextlib import contextmanager
from datetime import datetime
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, cast

from opentelemetry.context import Context, attach, get_current
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.trace import (
    INVALID_SPAN,
    Link,
    Span,
    Tracer,
    get_current_span,
    set_span_in_context,
    use_span,
)

import litellm
from litellm._logging import verbose_logger
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.otel.emitter import SpanEmitter, stamp_error
from litellm.integrations.otel.mappers import resolve_mappers
from litellm.integrations.otel.model.baggage import promoted_baggage
from litellm.integrations.otel.model.config import OpenTelemetryV2Config
from litellm.integrations.otel.model.metadata import (
    LLMCallEvent,
    RequestIdentity,
    auth_metadata,
    model_from_request_data,
)
from litellm.integrations.otel.model.payloads import (
    GuardrailSpanData,
    LLMCallSpanData,
    MCPListToolsSpanData,
    MCPToolCallSpanData,
    ServiceSpanData,
    SpanError,
    is_mcp_list_tools,
    is_mcp_tool_call,
)
from litellm.integrations.otel.model.semconv import Error
from litellm.integrations.otel.model.spans import SpanRole, span_role_for_service
from litellm.integrations.otel.model.utils import to_ns
from litellm.integrations.otel.plumbing.context import (
    is_recordable_span,
    mcp_message_transport_span,
    request_root_span,
    resolve_mcp_span_context,
    resolve_parent_context,
    resolve_request_span_context,
    set_request_baggage,
    set_request_root_span,
)
from litellm.integrations.otel.plumbing.events import GenAIEventRecorder
from litellm.integrations.otel.plumbing.metrics import (
    GenAIMetricRecorder,
    create_genai_metrics,
)
from litellm.integrations.otel.plumbing.providers import (
    build_tracer_provider,
    get_event_logger,
    get_meter,
    get_tracer,
    resolve_logger_provider,
    resolve_meter_provider,
)
from litellm.integrations.otel.plumbing.routing import TenantTracerCache

if TYPE_CHECKING:
    from opentelemetry.metrics import MeterProvider

    from litellm.caching.dual_cache import DualCache
    from litellm.proxy._types import UserAPIKeyAuth
    from litellm.types.services import ServiceLoggerPayload
    from litellm.types.utils import (
        CallTypesLiteral,
        StandardLoggingGuardrailInformation,
        StandardLoggingPayload,
    )

LITELLM_TRACER_NAME: Final = "litellm"


def _span_error_from_exception(
    exception: "Exception | None",
    *,
    status_code: int | None = None,
    traceback_str: str | None = None,
) -> SpanError:
    """A ``SpanError`` for a proxy-level failure that never produced a
    ``StandardLoggingPayload`` (auth / validation / malformed-body rejections),
    mirroring ``_parse_error``'s field mapping so it stamps the same v2 keys a
    failed LLM call does. ``status_code`` pins ``error.code`` to the real response
    status, matching v1's SERVER-span behavior."""
    from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup

    info: Final = StandardLoggingPayloadSetup.get_error_information(
        original_exception=exception,
        traceback_str=traceback_str,
    )
    return SpanError(
        error_type=info.get("error_class") or info.get("error_code") or None,
        message=info.get("error_message") or None,
        code=str(status_code) if status_code is not None else (info.get("error_code") or None),
        stack_trace=info.get("traceback") or None,
        llm_provider=info.get("llm_provider") or None,
    )


# Any callback whose class belongs to one of these modules is "the OTel
# callback" for proxy-global-registration purposes.
_OTEL_MODULES: Final = (
    "litellm.integrations.otel",
    "litellm.integrations.opentelemetry",
)


# Cap on the open-call carrier map. A span opened at ``pre_call`` that never
# reaches a success/failure callback (e.g. a stream that only fires stream
# events) would otherwise linger; bounding the map evicts the oldest so memory
# stays flat on a long-running proxy while covering every concurrent in-flight
# call.
_OPEN_CALLS_MAX: Final = 10_000


def _request_trace_links(context: Context | None) -> tuple[Link, ...] | None:
    """A link back to the request trace, for a span detached into its own trace."""
    anchor: Final = get_current_span(context).get_span_context()
    return (Link(anchor),) if anchor.is_valid else None


class _LLMCallSpan:
    """The state carried from the ``pre_call`` boundary to span close.

    ``span`` is the live span when it could be opened at the boundary (the server
    span was ambient), or ``None`` when creation was deferred because no ambient
    parent was visible — in which case the async callback creates it against its
    own (worker-copied) ambient context using ``start_time_ns``. The presence of
    a carrier for a call at all is the proof that ``pre_call`` ran, i.e. that an
    upstream call was actually attempted.

    ``provider`` is the routed provider the live span was opened on (``None`` on
    the default route or when creation was deferred). It is held in the tenant
    cache while the span is open so LRU eviction can't shut the provider down
    under it, and must be released exactly once when the carrier is removed.
    """

    __slots__ = ("provider", "span", "start_time_ns")

    def __init__(
        self,
        span: "Span | None",
        start_time_ns: int | None,
        provider: "TracerProvider | None" = None,
    ) -> None:
        self.span = span
        self.start_time_ns = start_time_ns
        self.provider = provider


class OpenTelemetryV2(CustomLogger):
    """The ``CustomLogger`` for OpenTelemetry."""

    def __init__(
        self,
        config: OpenTelemetryV2Config | None = None,
        callback_name: str | None = None,
        tracer_provider: TracerProvider | None = None,
        logger_provider: LoggerProvider | None = None,
        meter_provider: "MeterProvider | None" = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(**kwargs)
        self.config: OpenTelemetryV2Config = config or OpenTelemetryV2Config(**kwargs)
        self.callback_name = callback_name
        self._tracer_provider: TracerProvider = (
            tracer_provider if tracer_provider is not None else build_tracer_provider(self.config)
        )
        self.tracer: Tracer = get_tracer(self._tracer_provider, LITELLM_TRACER_NAME)
        self._metrics_recorder = self._init_metrics(meter_provider)
        self._metric_filter_error_logged = False
        self._emitter = SpanEmitter(
            self.tracer,
            self.config,
            mappers=resolve_mappers(self.config.mapper_names),
            event_recorder=self._init_events(logger_provider),
        )
        self._tenant_tracers = TenantTracerCache(self.config, callback_name, LITELLM_TRACER_NAME)
        self._open_llm_calls: OrderedDict[str, _LLMCallSpan] = OrderedDict()
        self._init_otel_logger_on_litellm_proxy()

    def _init_metrics(self, meter_provider: "MeterProvider | None") -> "GenAIMetricRecorder | None":
        """Create the six GenAI histograms when metrics are enabled, else ``None``.

        ``meter_provider`` is an explicit override (tests inject one); otherwise the
        provider is resolved from the OTel global so the operator's configured
        readers/exporters receive the metrics, building and registering one only
        when no global provider is set.
        """
        if not self.config.enable_metrics:
            return None
        provider: Final = resolve_meter_provider(self.config, meter_provider)
        meter: Final = get_meter(provider, LITELLM_TRACER_NAME)
        return GenAIMetricRecorder(create_genai_metrics(meter), self.callback_name)

    def _init_events(self, logger_provider: LoggerProvider | None) -> "GenAIEventRecorder | None":
        """Create the GenAI event recorder when events are enabled, else ``None``.

        ``logger_provider`` is an explicit override (tests inject one); otherwise the
        provider is resolved from the OTel global so an operator-configured logs
        pipeline receives the events, building and registering one only when no
        global provider is set. A ``None`` resolution means the operator opted out
        of the logs signal, so no recorder is built.
        """
        if not self.config.enable_events:
            return None
        provider: Final = resolve_logger_provider(self.config, logger_provider)
        if provider is None:
            return None
        return GenAIEventRecorder(get_event_logger(provider, LITELLM_TRACER_NAME))

    # ====================================================================== #
    #  Proxy global registration
    # ====================================================================== #

    def _register_in_callback_list(self, callbacks: list) -> None:
        already_otel: Final = any(
            cb.__class__.__module__.startswith(_OTEL_MODULES) for cb in callbacks if hasattr(cb, "__class__")
        )
        if not already_otel:
            callbacks.append(self)

    def _init_otel_logger_on_litellm_proxy(self) -> None:
        try:
            from litellm.proxy import proxy_server
        except Exception:
            return
        try:
            self._register_in_callback_list(litellm.service_callback)
            self._register_in_callback_list(litellm.input_callback)
            self._register_in_callback_list(litellm._async_success_callback)
            self._register_in_callback_list(litellm._async_failure_callback)
        except Exception:
            pass
        if getattr(proxy_server, "open_telemetry_logger", None) is None:
            setattr(proxy_server, "open_telemetry_logger", self)

    # ====================================================================== #
    #  LLM-call callbacks — the span is opened at the ``pre_call`` boundary and
    #  closed here. See ``log_pre_api_call``.
    # ====================================================================== #

    def log_pre_api_call(self, model, messages, kwargs):
        """Open the LLM-call span at the call boundary.

        Runs synchronously inside the request task, before the upstream call —
        the one place where the live server span is genuinely the ambient OTel
        context — so the span parents to it natively, with no span threaded
        through a metadata dict. The open span is stashed on the per-request
        ``LiteLLMLoggingObj`` (a typed object) and closed in the async callback.

        When no recordable parent is visible (``pre_call`` was driven from a thread
        pool for a sync-only provider, where contextvars — and so the anchor —
        don't follow), creation is deferred: only the start time is recorded, and
        the async callback — whose worker context was copied from the request task
        and so still carries the anchor — creates the span then.

        Synthetic proxy-gate error logs (auth/rate-limit rejections) also fire this
        hook but never made an upstream call; they are tagged and skipped so no
        phantom LLM-call span is produced.
        """
        call: Final = LLMCallEvent.from_dict(kwargs)
        if call.is_no_upstream_call:
            return
        call_id: Final = call.call_id
        if call_id is None:
            return
        # Idempotent: a retried call may re-enter ``pre_call`` with the same
        # call id; keep the first span so its start time is the true one.
        if call_id in self._open_llm_calls:
            return
        start_time_ns: Final = to_ns(datetime.now())
        # Parent to the request's anchored root span (stable across the request),
        # falling back to ambient on the SDK path. Open the span live only when
        # that resolves to a recordable parent; otherwise defer to the close
        # callback (the thread-pool case, where the anchor isn't visible here).
        # Do not route on the deferred path: creating or LRU-touching a tenant
        # provider here would evict idle ones even though close re-routes.
        parent_context: Final = resolve_request_span_context()
        if not is_recordable_span(get_current_span(parent_context)):
            self._store_open_call(call_id, _LLMCallSpan(span=None, start_time_ns=start_time_ns))
            return
        # A detached route roots its own trace instead (linked to the request
        # trace) — see ``TenantRoute.detached``.
        route: Final = self._tenant_tracers.route_for(self.tracer, call.dynamic_params, call.auth_metadata)
        try:
            span: Final = self._emitter.start_span(
                SpanRole.LLM_CALL,
                call.provisional_span_name,
                parent_context=(
                    set_span_in_context(INVALID_SPAN, parent_context) if route.detached else parent_context
                ),
                start_time_ns=start_time_ns,
                tracer=route.tracer,
                links=_request_trace_links(parent_context) if route.detached else None,
            )
        except BaseException:
            self._tenant_tracers.release(route.provider)
            raise
        self._store_open_call(
            call_id,
            _LLMCallSpan(span=span, start_time_ns=start_time_ns, provider=route.provider),
        )

    async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
        if self._emit_mcp_tool_call(kwargs, start_time, end_time):
            return
        if self._emit_mcp_list_tools(kwargs, start_time, end_time):
            return
        self._close_llm_call(kwargs, start_time, end_time)
        self._record_metrics(kwargs, response_obj, start_time, end_time)

    def _record_metrics(self, kwargs, response_obj, start_time, end_time) -> None:
        """Record the GenAI metrics for a successful LLM call."""
        self._guarded_record(lambda recorder: recorder.record(kwargs, response_obj, start_time, end_time))

    def _record_failure_metrics(self, kwargs, start_time, end_time) -> None:
        """Record the GenAI metrics for a failed LLM call, so the duration
        histogram covers the whole traffic rather than only what survived.

        A synthetic proxy-gate log (auth / rate-limit rejection) is skipped for the
        same reason it gets no span: no upstream call happened, so its duration is
        not a GenAI operation's duration and would pull the histogram down."""
        if LLMCallEvent.from_dict(kwargs).is_no_upstream_call:
            return
        self._guarded_record(lambda recorder: recorder.record_failure(kwargs, start_time, end_time))

    def _guarded_record(self, record: "Callable[[GenAIMetricRecorder], None]") -> None:
        """Run one metric recording. Best-effort: a recording failure (e.g. a
        malformed payload) must never break the span close or the request itself. A
        misconfigured attribute filter is operator-fixable, so it is surfaced once
        at ERROR instead of being swallowed."""
        if self._metrics_recorder is None:
            return
        try:
            record(self._metrics_recorder)
        except ValueError as exc:
            if not self._metric_filter_error_logged:
                verbose_logger.error(
                    "OpenTelemetryV2: invalid otel.attributes metric filter, metrics disabled: %s",
                    exc,
                )
                self._metric_filter_error_logged = True
        except Exception as exc:
            verbose_logger.debug("OpenTelemetryV2: metric recording failed: %s", exc)

    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
        if self._emit_mcp_tool_call(kwargs, start_time, end_time):
            return
        if self._emit_mcp_list_tools(kwargs, start_time, end_time):
            return
        self._close_llm_call(kwargs, start_time, end_time)
        self._record_failure_metrics(kwargs, start_time, end_time)

    def _seed_identity_baggage(self, identity: RequestIdentity, model: str | None, context: Context) -> Context:
        """Seed authenticated request-identity Baggage onto ``context`` so the Baggage
        processor stamps team/key/metadata onto the span. Identity is read from the
        parsed payload, never the client's ``params._meta`` carrier, so it can't be
        spoofed."""
        bag: Final = promoted_baggage(
            identity,
            model,
            promoted_keys=tuple(self.config.baggage_promoted_keys),
            metadata_keys=tuple(self.config.baggage_metadata_keys),
            team_metadata_keys=tuple(self.config.baggage_team_metadata_keys),
        )
        return set_request_baggage(bag, context=context) if bag else context

    def _emit_mcp_tool_call(
        self,
        kwargs: Mapping[str, object],
        start_time: datetime | float | None,
        end_time: datetime | float | None,
    ) -> bool:
        """Emit an MCP tool-call span when the closed request was a tool call.

        MCP tool calls reach the success/failure callbacks like any other request
        (with ``call_type`` ``call_mcp_tool``), but they are not LLM calls and have
        no ``pre_call`` carrier — so they get their own CLIENT span here. It nests
        under the transport span of the request carrying this message, and trace
        context the client propagated in ``params._meta`` is recorded as a span
        link (see ``resolve_mcp_span_context``). Returns whether it handled the
        event, so the caller skips the LLM-call path. The whole span is emitted at
        once (there is no boundary to open it at), deduped on the call id.
        """
        raw_payload: Final = kwargs.get("standard_logging_object")
        if not raw_payload or not is_mcp_tool_call(cast(Mapping[str, object], raw_payload)):
            return False
        payload: Final = cast("StandardLoggingPayload", raw_payload)
        data: Final = MCPToolCallSpanData.from_standard_logging_payload(
            payload, capture_content=self.config.capture_span_content
        )
        # A stray LLM carrier from a ``pre_call`` that mis-fired for this id would
        # otherwise linger until evicted; drop it so it's neither leaked nor closed
        # as a phantom LLM span.
        if data.identity.call_id:
            self._release_carrier(self._open_llm_calls.pop(data.identity.call_id, None))
        route: Final = self._tenant_tracers.route_for(self.tracer, None, auth_metadata(payload, kwargs))
        try:
            parent_context, links = resolve_mcp_span_context()
            seeded: Final = self._seed_identity_baggage(data.identity, None, parent_context)
            self._emitter.emit(
                SpanRole.MCP_TOOL_CALL,
                data,
                parent_context=(set_span_in_context(INVALID_SPAN, seeded) if route.detached else seeded),
                start_time_ns=to_ns(start_time),
                end_time_ns=to_ns(end_time),
                links=((*(links or ()), *(_request_trace_links(seeded) or ())) if route.detached else links),
                tracer=route.tracer,
            )
        finally:
            self._tenant_tracers.release(route.provider)
        return True

    def _emit_mcp_list_tools(
        self,
        kwargs: Mapping[str, object],
        start_time: datetime | float | None,
        end_time: datetime | float | None,
    ) -> bool:
        """Emit an MCP ``tools/list`` span when the closed request was a discovery call.

        Like a tool call, listing reaches the success/failure callbacks (here with
        ``call_type`` ``list_mcp_tools``) with no ``pre_call`` carrier, so it gets its
        own CLIENT span, nested under the transport span of the request carrying
        this message with any ``params._meta`` trace context recorded as a span
        link (see ``resolve_mcp_span_context``). Returns whether it handled the event so
        the caller skips the LLM-call path.
        """
        raw_payload: Final = kwargs.get("standard_logging_object")
        if not raw_payload or not is_mcp_list_tools(cast(Mapping[str, object], raw_payload)):
            return False
        payload: Final = cast("StandardLoggingPayload", raw_payload)
        data: Final = MCPListToolsSpanData.from_standard_logging_payload(
            payload, capture_content=self.config.capture_span_content
        )
        if data.identity.call_id:
            self._release_carrier(self._open_llm_calls.pop(data.identity.call_id, None))
        route: Final = self._tenant_tracers.route_for(self.tracer, None, auth_metadata(payload, kwargs))
        try:
            parent_context, links = resolve_mcp_span_context()
            seeded: Final = self._seed_identity_baggage(data.identity, None, parent_context)
            self._emitter.emit(
                SpanRole.MCP_LIST_TOOLS,
                data,
                parent_context=(set_span_in_context(INVALID_SPAN, seeded) if route.detached else seeded),
                start_time_ns=to_ns(start_time),
                end_time_ns=to_ns(end_time),
                links=((*(links or ()), *(_request_trace_links(seeded) or ())) if route.detached else links),
                tracer=route.tracer,
            )
        finally:
            self._tenant_tracers.release(route.provider)
        return True

    def _close_llm_call(
        self,
        kwargs: Mapping[str, object],
        start_time: datetime | float | None,
        end_time: datetime | float | None,
    ) -> Span | None:
        """Finish the LLM-call span opened at ``pre_call`` (or create it deferred).

        No carrier for this call id means ``pre_call`` never ran — the request was
        rejected at the gate or blocked by a pre-call guardrail before any upstream
        call — so there is nothing to record and no phantom span.
        """
        call: Final = LLMCallEvent.from_dict(kwargs)
        call_id: Final = call.call_id
        # ``pop`` is the dedup: this method runs from both the success and failure
        # paths, and whichever fires first removes the carrier and closes the span.
        carrier: Final = self._open_llm_calls.pop(call_id, None) if call_id else None
        # A missing carrier does not always mean nothing happened: a team/key-scoped
        # logger is a success/failure callback only, so ``pre_call`` never reaches it
        # and no carrier exists. The payload plus the request-level provider-handoff
        # stamp (``upstream_started``) is the affirmative signal of a real call; a
        # gate rejection carries ``is_no_upstream_call`` and gets no span.
        if carrier is None and (call.is_no_upstream_call or not call.upstream_started or call.payload is None):
            return None
        try:
            return self._finish_carrier(carrier, call, start_time, end_time)
        finally:
            # After the span has ended, so a release-triggered provider shutdown
            # force-flushes it out rather than racing its enqueue.
            self._release_carrier(carrier)

    def _store_open_call(self, call_id: str, carrier: _LLMCallSpan) -> None:
        """Remember an in-flight LLM call, evicting the oldest if over budget.

        A call that opens but never closes (a stream that only fires stream
        events) would linger otherwise. Eviction only drops the boundary carrier,
        not the call: if that call later closes as a real completed call, it still
        emits through the deferred branch in ``_close_llm_call`` (the same path a
        team/key-scoped logger uses, since it never opens a carrier), deduplicated
        by call id. Only a call that is evicted and never closes goes unexported.
        """
        self._open_llm_calls[call_id] = carrier
        if len(self._open_llm_calls) > _OPEN_CALLS_MAX:
            _, evicted = self._open_llm_calls.popitem(last=False)
            self._release_carrier(evicted)

    def _release_carrier(self, carrier: "_LLMCallSpan | None") -> None:
        """Release the routed provider a removed carrier was holding open."""
        if carrier is not None:
            self._tenant_tracers.release(carrier.provider)

    def _finish_carrier(
        self,
        carrier: "_LLMCallSpan | None",
        call: LLMCallEvent,
        start_time: datetime | float | None,
        end_time: datetime | float | None,
    ) -> Span | None:
        payload: Final = call.payload
        call_id: Final = call.call_id
        if payload is None:
            if carrier is not None and carrier.span is not None:
                # Opened at the boundary but the payload never materialized — end
                # it (named provisionally) so it isn't leaked as an open span, and
                # register the dedup marker so a later payload-carrying close for
                # the same call id cannot re-emit through the deferred branch.
                self._emitter.mark_emitted(call_id, SpanRole.LLM_CALL)
                carrier.span.end(end_time=to_ns(end_time))
            return None
        data: Final = LLMCallSpanData.from_standard_logging_payload(
            payload,
            capture_content=self.config.capture_span_content,
            time_to_first_chunk_seconds=call.time_to_first_chunk_seconds,
        )
        end_time_ns: Final = to_ns(end_time)
        if carrier is not None and carrier.span is not None:
            # Born at the boundary: stamp attributes from the typed payload, set
            # status, and end it. Its parent (the server span) was captured at
            # creation from real ambient context. Register the dedup marker so a
            # second close for the same call id (success then failure on one
            # logging object) cannot re-emit through the deferred branch.
            self._emitter.mark_emitted(call_id, SpanRole.LLM_CALL)
            self._emitter.finish_span(SpanRole.LLM_CALL, carrier.span, data, end_time_ns=end_time_ns)
            return carrier.span
        # Deferred: ``pre_call`` saw no recordable parent, so create the span now.
        # The worker copied the request task's context, which carries the anchored
        # root span — parent to it (ambient fallback on the SDK path). Seed identity
        # Baggage so the span — and the SDK path, which has none — is labeled
        # consistently. A detached route roots its own trace instead, linked back.
        route: Final = self._tenant_tracers.route_for(self.tracer, call.dynamic_params, call.auth_metadata)
        try:
            parent_ctx: Final = self._seed_identity_baggage(
                data.identity, data.request_model, resolve_request_span_context()
            )
            return self._emitter.emit(
                SpanRole.LLM_CALL,
                data,
                parent_context=(set_span_in_context(INVALID_SPAN, parent_ctx) if route.detached else parent_ctx),
                start_time_ns=(carrier.start_time_ns if carrier is not None else to_ns(start_time)),
                end_time_ns=end_time_ns,
                tracer=route.tracer,
                links=_request_trace_links(parent_ctx) if route.detached else None,
            )
        finally:
            self._tenant_tracers.release(route.provider)

    # ====================================================================== #
    #  Service hooks
    # ====================================================================== #

    async def async_service_success_hook(
        self,
        payload: "ServiceLoggerPayload",
        parent_otel_span: Span | None = None,
        start_time: datetime | float | None = None,
        end_time: datetime | float | None = None,
        event_metadata: dict | None = None,
    ) -> None:
        self._emit_service(
            payload,
            parent_otel_span=parent_otel_span,
            start_time=start_time,
            end_time=end_time,
            event_metadata=event_metadata,
            error_override=None,
        )

    async def async_service_failure_hook(
        self,
        payload: "ServiceLoggerPayload",
        error: str | None = "",
        parent_otel_span: Span | None = None,
        start_time: datetime | float | None = None,
        end_time: datetime | float | None = None,
        event_metadata: dict | None = None,
    ) -> None:
        self._emit_service(
            payload,
            parent_otel_span=parent_otel_span,
            start_time=start_time,
            end_time=end_time,
            event_metadata=event_metadata,
            error_override=error or "error",
        )

    def _emit_service(
        self,
        payload: "ServiceLoggerPayload",
        *,
        parent_otel_span: Span | None,
        start_time: datetime | float | None,
        end_time: datetime | float | None,
        event_metadata: dict | None,
        error_override: str | None,
    ) -> Span | None:
        data = ServiceSpanData.from_payload(payload, event_metadata=event_metadata)
        # Decide whether this service call is a span at all, and of what kind.
        # ``None`` means metrics-only (framework instrumentation that duplicates a
        # gen-AI span — ``self``/``router``/``proxy_pre_call`` — or ``auth``, which
        # gets a live phase span instead). Those still feed Prometheus/Datadog via
        # their own hooks; they just never enter the trace.
        role: Final = span_role_for_service(data.service_name)
        if role is None:
            return None
        # A metrics-only ping with neither timing nor a parent (in-memory queue
        # gauges) is not a traceable operation; a span for it would be a
        # zero-duration root with no context, so skip it. Real background work
        # (budget/reset jobs, spend flush) passes start/end times and still emits
        # as a root; anything with a parent emits regardless.
        if error_override is None and start_time is None and end_time is None and parent_otel_span is None:
            return None
        if error_override is not None and data.error is None:
            data = ServiceSpanData(
                service_name=data.service_name,
                call_type=data.call_type,
                error=SpanError(message=error_override),
                event_metadata=data.event_metadata,
            )
        # Parent like every other span: ambient context first (so identity Baggage
        # rides along and the call nests under whatever request phase is active —
        # e.g. a DB lookup under the live ``auth`` span), falling back to the
        # server span the proxy threaded as ``parent_otel_span``. A background
        # service call has neither, so it starts its own root trace.
        parent_context: Final = resolve_parent_context(threaded=parent_otel_span)
        return self._emitter.emit(
            role,
            data,
            parent_context=parent_context,
            start_time_ns=to_ns(start_time),
            end_time_ns=to_ns(end_time),
        )

    # ====================================================================== #
    #  async_post_call_* hooks — emit guardrail spans. The server span's status
    #  / errors are the FastAPI instrumentor's job, so we don't touch it here.
    # ====================================================================== #

    def seed_request_identity(self, user_api_key_dict: object, model: str | None = None) -> None:
        """Attach request-identity Baggage to the current context + server span.

        Seeding identity into Baggage makes **every** span emitted afterwards for
        this request — LLM call, guardrail, DB call — inherit it via
        ``LiteLLMBaggageSpanProcessor``. Called once at the auth boundary (as soon
        as the key resolves) so post-auth spans are labeled consistently; the
        Baggage rides the request task's contextvar from there on. Auth-internal
        DB lookups that run before the key is known stay unlabeled — identity
        isn't determined yet, which is correct.
        """
        try:
            identity: Final = RequestIdentity.from_user_api_key_auth(user_api_key_dict)
            bag: Final = promoted_baggage(
                identity,
                model,
                promoted_keys=tuple(self.config.baggage_promoted_keys),
                metadata_keys=tuple(self.config.baggage_metadata_keys),
                team_metadata_keys=tuple(self.config.baggage_team_metadata_keys),
            )
            if bag:
                # Attach (no detach): the contextvar is scoped to this request's
                # asyncio task and is reclaimed when the task ends.
                attach(set_request_baggage(bag, context=get_current()))
                # The server span was started by the instrumentor before this ran,
                # so the Baggage processor (which only fires at span start) won't
                # backfill it — stamp identity on it directly. Prefer the anchored
                # root span over the ambient one so identity still lands on the
                # server span when seeding from inside the live ``auth`` phase span
                # (the auth-failure path), where ``get_current_span`` is the phase
                # span, not the request's root.
                server_span: Final = request_root_span() or get_current_span()
                if is_recordable_span(server_span):
                    # Re-capture the anchor here too: this runs post-auth with the
                    # server span active and covers entrypoints that bypass
                    # ``create_litellm_proxy_request_started_span`` (e.g. the SDK
                    # path's ``async_pre_call_hook``). Idempotent.
                    set_request_root_span(server_span)
                    for key, value in bag.items():
                        server_span.set_attribute(key, value)
        except Exception:
            pass

    @contextmanager
    def start_phase_span(self, name: str) -> "Iterator[Span]":
        span: Final = self._emitter.start_span(SpanRole.SERVICE, name)
        with use_span(span, end_on_exit=True):
            try:
                yield span
            except Exception as exc:
                if is_recordable_span(span):
                    stamp_error(span, _span_error_from_exception(exc), record_event=False, set_status=False)
                raise

    async def async_pre_call_hook(
        self,
        user_api_key_dict: "UserAPIKeyAuth",
        cache: "DualCache",
        data: dict,
        call_type: "CallTypesLiteral",
    ) -> dict:
        self.seed_request_identity(
            user_api_key_dict,
            model=model_from_request_data(data),
        )
        return data

    def record_error_attributes_on_span(
        self,
        span: "Span | None",
        exception: "Exception | None",
        status_code: int,
    ) -> None:
        """Stamp the v2 error.* attributes on the FastAPI-owned SERVER span for a
        failure that dies before any LLM-call span exists (malformed body, auth /
        validation rejection). Called from the proxy's global exception handler via
        ``_close_dangling_otel_server_span``, which swallows the exception into a
        ``JSONResponse`` so the instrumentor never sees it and leaves the span
        ``UNSET``; the status is set here instead (v1 did the same from the handler)
        so a failed request reads as failed and not merely as a span carrying an
        error message. The instrumentor still owns the span's lifecycle, so this
        never ends it. The exception event is recorded only when nothing stamped
        this span already — ``async_post_call_failure_hook`` and the ``auth`` phase
        span record their own, and a second event would duplicate it — while the
        attributes are always restamped so ``error.code`` stays pinned to the real
        response status."""
        if span is None or not is_recordable_span(span):
            return
        already_stamped: Final = Error.TYPE in (getattr(span, "attributes", None) or ())
        stamp_error(
            span,
            _span_error_from_exception(exception, status_code=status_code),
            record_event=not already_stamped,
        )

    async def async_post_call_failure_hook(
        self,
        request_data: dict,
        original_exception: Exception,
        user_api_key_dict: "UserAPIKeyAuth",
        traceback_str: "str | None" = None,
    ) -> None:
        """Stamp error.* on the request's root SERVER span for a proxy-level
        failure that never reached an LLM call (empty body rejected in the
        endpoint, auth failure), so the failed request carries the same error keys
        a failed LLM call does. v1's ``OpenTelemetry`` implemented this same hook;
        v2 lost it when it stopped subclassing ``OpenTelemetry``, which is the
        LIT-4179 regression for pre-call failures.

        An MCP message is handled on the session's task, where the request-root
        anchor is whatever request opened the session, so prefer the transport the
        gateway published for this specific message. Without that, a failed tool
        call aimed its error at the ``initialize`` request's finished span and the
        SDK dropped it, leaving the POST that actually failed unmarked."""
        span: Final = mcp_message_transport_span() or request_root_span() or user_api_key_dict.parent_otel_span
        if span is None or not is_recordable_span(span):
            return
        stamp_error(span, _span_error_from_exception(original_exception, traceback_str=traceback_str))
        return

    def emit_guardrail_span(self, entry: "StandardLoggingGuardrailInformation") -> None:
        # Emitted by the guardrail-recording code the moment a guardrail finishes,
        # not from a post-call hook — that hook does not fire on every path (a
        # pass-through request that passes its guardrails never reaches it), which
        # left passing guardrails without a span.
        #
        # A guardrail is a sibling of the LLM call under the request's root span,
        # so parent it to the explicit anchor — never the active span, which during
        # a pre_call guardrail can be the live ``auth`` phase span. Emit with the
        # guardrail's actual execution window so a pre_call guardrail is placed
        # before the LLM call rather than at emission time. One entry in, one span
        # out — the module-level entry point routes each entry to this single
        # registered logger so a guardrail is never emitted more than once.
        data: Final = GuardrailSpanData.from_logging_entry(entry)
        self._emitter.emit(
            SpanRole.GUARDRAIL,
            data,
            parent_context=resolve_request_span_context(),
            start_time_ns=to_ns(data.start_time),
            end_time_ns=to_ns(data.end_time),
        )

    def create_litellm_proxy_request_started_span(
        self, start_time: datetime, headers: Mapping[str, str] | None
    ) -> Span | None:
        span: Final = get_current_span()
        if not is_recordable_span(span):
            return None
        set_request_root_span(span)
        return span


def select_global_otel_v2_logger(
    in_memory_loggers: Sequence[object],
    registered: "OpenTelemetryV2 | None" = None,
) -> "OpenTelemetryV2":
    """The single ``OpenTelemetryV2`` whose provider should become the OTel global.

    The callback factory designates one logger as canonical the moment it builds
    the first one (``_init_otel_logger_on_litellm_proxy`` sets
    ``proxy_server.open_telemetry_logger``), and every other v2 entry point —
    guardrail, identity seeding, phase spans — already routes through that same
    ``registered`` owner. Reuse it here too so the global provider has one source
    of truth instead of a second, independently-derived guess; this is the logger
    a preset (arize, langfuse, …) folds the ``OTEL_*`` base exporter and its own
    exporter into, so the FastAPI server span and the gen-ai spans share one
    provider and one trace.

    Fall back to ``in_memory_loggers`` for the SDK path, where no proxy global is
    set (selecting from there, not ``service_callback``, which a preset logger does
    not always reach), and build a generic logger from ``OTEL_*`` only when none was
    configured at all. Each fallback still avoids the second generic logger that
    orphaned the gen-ai spans onto a different backend than the server span.
    """
    if registered is not None:
        return registered
    existing: Final = next((cb for cb in in_memory_loggers if isinstance(cb, OpenTelemetryV2)), None)
    return existing if existing is not None else OpenTelemetryV2()


def publish_global_otel_v2_provider(
    in_memory_loggers: Sequence[object],
    set_global_provider: Callable[[TracerProvider], None],
    registered: "OpenTelemetryV2 | None" = None,
) -> "OpenTelemetryV2":
    """Select the single v2 logger and publish its provider as the OTel global.

    The proxy calls this once at startup, after callbacks are initialized, so the
    preset logger already exists; it passes ``registered`` (the canonical owner the
    factory designated as ``proxy_server.open_telemetry_logger``) so the global
    provider reuses the same logger the rest of the v2 code emits through (see
    :func:`select_global_otel_v2_logger`). Both ``registered`` and
    ``set_global_provider`` (the proxy passes
    ``opentelemetry.trace.set_tracer_provider``) are injected so the publish step is
    unit-testable without reading or mutating real global OTel state. Returns the
    logger whose provider was published.
    """
    logger: Final = select_global_otel_v2_logger(in_memory_loggers, registered=registered)
    set_global_provider(logger._tracer_provider)
    return logger


def _registered_v2_logger() -> "OpenTelemetryV2 | None":
    try:
        from litellm.proxy import proxy_server
    except Exception:
        return None
    logger: Final = getattr(proxy_server, "open_telemetry_logger", None)
    return logger if isinstance(logger, OpenTelemetryV2) else None


def emit_guardrail_span(entry: "StandardLoggingGuardrailInformation") -> None:
    """Emit a guardrail span on the registered v2 OTel logger.

    Called by the guardrail-recording code the moment a guardrail finishes, so a
    span is produced regardless of whether a post-call hook later runs (it does
    not on the pass-through allow path). Routes through the single canonical
    logger — the same one every other v2 entry point uses — so a guardrail
    recorded once yields exactly one span; fanning out across every reachable
    ``OpenTelemetryV2`` instance double-emits the same entry. Best-effort: span
    emission must never break guardrail evaluation.
    """
    logger: Final = _registered_v2_logger()
    if logger is None:
        return
    try:
        logger.emit_guardrail_span(entry)
    except Exception:
        pass


def seed_request_identity(user_api_key_dict: object, model: str | None = None) -> None:
    logger: Final = _registered_v2_logger()
    if logger is not None:
        logger.seed_request_identity(user_api_key_dict, model=model)


@contextmanager
def phase_span(name: str) -> "Iterator[Span | None]":
    logger: Final = _registered_v2_logger()
    if logger is None:
        yield None
        return
    with logger.start_phase_span(name) as span:
        yield span


def build_otel_v2_logger(
    config: OpenTelemetryV2Config,
    callback_name: str | None = None,
    tracer_provider: TracerProvider | None = None,
    logger_provider: LoggerProvider | None = None,
    meter_provider: "MeterProvider | None" = None,
    settings: Mapping[str, object] = MappingProxyType({}),
) -> OpenTelemetryV2:
    return _logger_class(config)(
        config=config,
        callback_name=callback_name,
        tracer_provider=tracer_provider,
        logger_provider=logger_provider,
        meter_provider=meter_provider,
        **settings,
    )


def _logger_class(config: OpenTelemetryV2Config) -> type[OpenTelemetryV2]:
    if "langfuse" not in config.mapper_names or not config.capture_span_content:
        return OpenTelemetryV2
    from litellm.integrations.otel.langfuse_logger import LangfuseOpenTelemetryV2

    return LangfuseOpenTelemetryV2
