feat: add OTEL GenAI latest-experimental semantic convention support (#27418)

- Introduce `OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental` opt-in that switches OTEL traces to conform with the OpenTelemetry GenAI semantic conventions specification
- Extract all semconv behavior into a new `OTELGenAISemconvMixin` class in `gen_ai_semconv.py`, mixed into `OpenTelemetry` to keep concerns separated
- In semconv mode, span name follows `{operation} {model}` pattern (e.g. `chat gpt-4`) and span kind is set to `CLIENT` instead of legacy `litellm_request`
- Replace `gen_ai.system` with `gen_ai.provider.name` and drop `llm.is_streaming` in semconv mode; add `gen_ai.request.{frequency_penalty,presence_penalty,top_k,seed,stop_sequences,stream,choice.count}` and `gen_ai.usage.cache_{creation,read}.input_tokens` attributes
- Replace per-message `gen_ai.content.prompt` / per-choice `gen_ai.content.completion` log events with a single consolidated `gen_ai.client.inference.operation.details` event; omit `gen_ai.input/output.messages` when content capture is disabled
- Suppress the non-standard `raw_gen_ai_request` child span entirely in semconv mode
- Support both programmatic (`OpenTelemetryConfig.semconv_stability_opt_in` field) and environment variable activation; the two sources are unioned so either or both can enable the opt-in
- Extract OTEL SDK `LogRecord` / `SeverityNumber` version-compatibility shim into a reusable `_otel_log_types()` static method to deduplicate the `< 1.39.0` / `>= 1.39.0` import branching
- Add 30+ unit tests covering opt-in gating, span naming, attribute emission/omission rules, stop sequence normalization, cache token attributes, and the consolidated event lifecycle

Co-authored-by: Yassin Kortam <yassinkortam@g.ucla.edu>

litellm/integrations/opentelemetry.py

@@ -1,7 +1,7 @@
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union, cast
 
 import litellm
 from litellm._logging import verbose_logger
@@ -10,6 +10,12 @@ from litellm.integrations._types.open_inference import (
     SpanAttributes,
 )
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.opentelemetry_utils.gen_ai_semconv import (
+    OTEL_SEMCONV_STABILITY_OPT_IN_ENV,
+    OTELGenAISemconvMixin,
+    OTELSemconvCategory,
+    parse_semconv_opt_in,
+)
 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
 from litellm.secret_managers.main import get_secret_bool, str_to_bool
 from litellm.types.services import ServiceLoggerPayload
@@ -85,6 +91,7 @@ class OpenTelemetryConfig:
     # Programmatic override for OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT.
     # One of NO_CONTENT, SPAN_ONLY, EVENT_ONLY, SPAN_AND_EVENT (or "true" as legacy alias).
     capture_message_content: Optional[str] = None
+    semconv_stability_opt_in: Set[OTELSemconvCategory] = field(default_factory=set)
 
     def __post_init__(self) -> None:
         # If endpoint is specified but exporter is still the default "console",
@@ -110,6 +117,11 @@ class OpenTelemetryConfig:
             self.ignore_context_propagation = str_to_bool(
                 os.getenv("OTEL_IGNORE_CONTEXT_PROPAGATION")
             )
+        # Resolve the env opt-in once here so self.semconv_stability_opt_in is the
+        # single source of truth: the union of programmatic and env categories.
+        self.semconv_stability_opt_in |= parse_semconv_opt_in(
+            os.getenv(OTEL_SEMCONV_STABILITY_OPT_IN_ENV)
+        )
 
     @classmethod
     def from_env(cls):
@@ -157,7 +169,7 @@ class OpenTelemetryConfig:
         )
 
 
-class OpenTelemetry(CustomLogger):
+class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
     def __init__(
         self,
         config: Optional[OpenTelemetryConfig] = None,
@@ -979,13 +991,14 @@ class OpenTelemetry(CustomLogger):
 
         otel_tracer: Tracer = self.get_tracer_to_use_for_request(kwargs)
 
-        # Always create a new span
-        # The parent relationship is preserved through the context parameter
-        span = otel_tracer.start_span(
-            name=self._get_span_name(kwargs),
-            start_time=self._to_ns(start_time),
-            context=context,
-        )
+        span_kwargs: Dict[str, Any] = {
+            "name": self._get_span_name(kwargs),
+            "start_time": self._to_ns(start_time),
+            "context": context,
+        }
+        if self._gen_ai_semconv_latest_experimental:
+            span_kwargs["kind"] = self.span_kind.CLIENT
+        span = otel_tracer.start_span(**span_kwargs)
 
         span.set_status(Status(StatusCode.OK))
         self.set_attributes(span, kwargs, response_obj)
@@ -998,6 +1011,10 @@ class OpenTelemetry(CustomLogger):
         from opentelemetry import trace
         from opentelemetry.trace import Status, StatusCode
 
+        # raw_gen_ai_request is non-standard in semconv mode.
+        if self._gen_ai_semconv_latest_experimental:
+            return
+
         if not self._capture_in_span():
             return
 
@@ -1023,7 +1040,11 @@ class OpenTelemetry(CustomLogger):
         provider = params.get("custom_llm_provider", "Unknown")
 
         common_attrs = {
-            "gen_ai.operation.name": "chat",
+            "gen_ai.operation.name": (
+                self._gen_ai_operation_name(kwargs)
+                if self._gen_ai_semconv_latest_experimental
+                else "chat"
+            ),
             "gen_ai.system": provider,
             "gen_ai.request.model": kwargs.get("model"),
             "gen_ai.framework": "litellm",
@@ -1246,6 +1267,24 @@ class OpenTelemetry(CustomLogger):
                 response_duration_seconds, attributes=common_attrs
             )
 
+    @staticmethod
+    def _otel_log_types():
+        """Resolve ``(LogRecord, SeverityNumber)`` across OTEL SDK versions.
+
+        ``LogRecord`` moved out of ``opentelemetry.sdk._logs`` in OTEL >= 1.39.0
+        (open-telemetry/opentelemetry-python#4676). Imports stay function-local
+        because the SDK is an optional dependency.
+        """
+        from opentelemetry._logs import SeverityNumber
+
+        try:
+            from opentelemetry.sdk._logs import LogRecord  # OTEL < 1.39.0
+        except ImportError:
+            from opentelemetry.sdk._logs._internal import (  # OTEL >= 1.39.0
+                LogRecord,
+            )
+        return LogRecord, SeverityNumber
+
     def _emit_semantic_logs(self, kwargs, response_obj, span: Span):
         if not self.config.enable_events:
             return
@@ -1259,16 +1298,7 @@ class OpenTelemetry(CustomLogger):
         # See: https://github.com/open-telemetry/opentelemetry-python/pull/4676
         # TODO: Refactor to use the proper OTEL Logs API instead of directly creating SDK LogRecords
 
-        from opentelemetry._logs import SeverityNumber
-
-        try:
-            from opentelemetry.sdk._logs import (  # type: ignore[attr-defined]  # OTEL < 1.39.0
-                LogRecord as SdkLogRecord,
-            )
-        except ImportError:
-            from opentelemetry.sdk._logs._internal import (
-                LogRecord as SdkLogRecord,  # type: ignore[attr-defined]  # OTEL >= 1.39.0
-            )
+        SdkLogRecord, SeverityNumber = self._otel_log_types()
 
         # Resolve through the handler's own LoggerProvider (which may be a
         # private one when skip_set_global=True) rather than the module-level
@@ -1280,6 +1310,16 @@ class OpenTelemetry(CustomLogger):
             "custom_llm_provider", "Unknown"
         )
 
+        if self._gen_ai_semconv_latest_experimental:
+            self._emit_inference_details_event(
+                kwargs=kwargs,
+                response_obj=response_obj,
+                provider=provider,
+                otel_logger=otel_logger,
+                parent_ctx=parent_ctx,
+            )
+            return
+
         # per-message events
         for msg in kwargs.get("messages", []):
             role = msg.get("role", "user")
@@ -1496,11 +1536,14 @@ class OpenTelemetry(CustomLogger):
         if should_create_primary_span:
             # Span 1: Request sent to litellm SDK
             otel_tracer: Tracer = self.get_tracer_to_use_for_request(kwargs)
-            span = otel_tracer.start_span(
-                name=self._get_span_name(kwargs),
-                start_time=self._to_ns(start_time),
-                context=_parent_context,
-            )
+            span_kwargs: Dict[str, Any] = {
+                "name": self._get_span_name(kwargs),
+                "start_time": self._to_ns(start_time),
+                "context": _parent_context,
+            }
+            if self._gen_ai_semconv_latest_experimental:
+                span_kwargs["kind"] = self.span_kind.CLIENT
+            span = otel_tracer.start_span(**span_kwargs)
             span.set_status(Status(StatusCode.ERROR))
             self.set_attributes(span, kwargs, response_obj)
 
@@ -1782,11 +1825,21 @@ class OpenTelemetry(CustomLogger):
             )
 
             # The Generative AI Provider: Azure, OpenAI, etc.
-            self.safe_set_attribute(
-                span=span,
-                key=SpanAttributes.LLM_SYSTEM.value,
-                value=litellm_params.get("custom_llm_provider", "Unknown"),
-            )
+            provider_name = litellm_params.get("custom_llm_provider", "Unknown")
+            # Latest-experimental semconv replaced gen_ai.system with
+            # gen_ai.provider.name; emit only the conformant key in that mode.
+            if self._gen_ai_semconv_latest_experimental:
+                self.safe_set_attribute(
+                    span=span,
+                    key="gen_ai.provider.name",
+                    value=provider_name,
+                )
+            else:
+                self.safe_set_attribute(
+                    span=span,
+                    key=SpanAttributes.LLM_SYSTEM.value,
+                    value=provider_name,
+                )
 
             # The maximum number of tokens the LLM generates for a request.
             if optional_params.get("max_tokens"):
@@ -1812,11 +1865,17 @@ class OpenTelemetry(CustomLogger):
                     value=optional_params.get("top_p"),
                 )
 
-            self.safe_set_attribute(
-                span=span,
-                key=SpanAttributes.LLM_IS_STREAMING.value,
-                value=str(optional_params.get("stream", False)),
-            )
+            if self._gen_ai_semconv_latest_experimental:
+                # Semconv emits gen_ai.request.stream (only when streaming) via
+                # _set_semconv_request_attributes; skip the legacy llm.is_streaming.
+                self._set_semconv_request_attributes(span, optional_params)
+                self._set_semconv_cache_token_attributes(span, standard_logging_payload)
+            else:
+                self.safe_set_attribute(
+                    span=span,
+                    key=SpanAttributes.LLM_IS_STREAMING.value,
+                    value=str(optional_params.get("stream", False)),
+                )
 
             if optional_params.get("user"):
                 self.safe_set_attribute(
@@ -1937,14 +1996,18 @@ class OpenTelemetry(CustomLogger):
                         value=safe_dumps(transformed_system_instructions),
                     )
 
-            self.safe_set_attribute(
-                span=span,
-                key=SpanAttributes.GEN_AI_OPERATION_NAME.value,
-                value=(
+            if self._gen_ai_semconv_latest_experimental:
+                operation_name = self._gen_ai_operation_name(kwargs)
+            else:
+                operation_name = (
                     "chat"
                     if standard_logging_payload.get("call_type") == "completion"
                     else standard_logging_payload.get("call_type") or "chat"
-                ),
+                )
+            self.safe_set_attribute(
+                span=span,
+                key=SpanAttributes.GEN_AI_OPERATION_NAME.value,
+                value=operation_name,
             )
 
             if standard_logging_payload.get("request_id"):
@@ -2281,6 +2344,10 @@ class OpenTelemetry(CustomLogger):
         if generation_name:
             return generation_name
 
+        if self._gen_ai_semconv_latest_experimental:
+            model = kwargs.get("model") or "unknown"
+            return f"{self._gen_ai_operation_name(kwargs)} {model}"
+
         return LITELLM_REQUEST_SPAN_NAME
 
     def get_traceparent_from_header(self, headers):

litellm/integrations/opentelemetry_utils/gen_ai_semconv.py

@@ -0,0 +1,271 @@
+"""OTEL GenAI ``gen_ai_latest_experimental`` semantic conventions.
+
+Setting ``OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental`` switches the
+emitted traces to the experimental OTEL GenAI conventions
+(https://opentelemetry.io/docs/specs/semconv/gen-ai/). Concretely, versus the
+default LiteLLM output:
+
+Request span:
+
+- name is ``{operation} {model}`` (e.g. ``chat gpt-4``) instead of
+  ``litellm_request``; span kind is ``CLIENT``.
+- ``gen_ai.operation.name`` is the actual operation (``chat`` /
+  ``text_completion`` / ``embeddings``) instead of always ``chat``.
+- the provider is reported as ``gen_ai.provider.name``; the superseded
+  ``gen_ai.system`` and the legacy ``llm.is_streaming`` are dropped.
+- adds ``gen_ai.request.{frequency_penalty,presence_penalty,top_k,seed}``,
+  ``gen_ai.request.stop_sequences`` (a string array),
+  ``gen_ai.request.stream`` (only when streaming),
+  ``gen_ai.request.choice.count`` (only when n > 1), and
+  ``gen_ai.usage.cache_{creation,read}.input_tokens``.
+- the non-standard ``raw_gen_ai_request`` child span is no longer created.
+
+Events:
+
+- the per-message ``gen_ai.content.prompt`` / per-choice
+  ``gen_ai.content.completion`` log events are replaced by a single
+  ``gen_ai.client.inference.operation.details`` log event carrying
+  ``gen_ai.input.messages`` / ``gen_ai.output.messages`` (message content
+  included only when content capture is enabled).
+"""
+
+from datetime import datetime
+from enum import Enum
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple, Union
+
+from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
+
+if TYPE_CHECKING:
+    from opentelemetry.trace import Span as _Span
+
+    from litellm.integrations.opentelemetry import OpenTelemetryConfig
+
+    Span = Union[_Span, Any]
+else:
+    Span = Any
+
+
+# OTEL_SEMCONV_STABILITY_OPT_IN is a comma-separated list of category-specific
+# opt-in values. See https://opentelemetry.io/docs/specs/semconv/gen-ai/
+OTEL_SEMCONV_STABILITY_OPT_IN_ENV = "OTEL_SEMCONV_STABILITY_OPT_IN"
+
+
+class OTELSemconvCategory(Enum):
+    GEN_AI_LATEST_EXPERIMENTAL = "gen_ai_latest_experimental"
+
+
+# Reverse lookup: opt-in token string -> OTELSemconvCategory.
+_SEMCONV_CATEGORY_BY_VALUE = {
+    category.value: category for category in OTELSemconvCategory
+}
+
+
+# LiteLLM optional_params key -> OTEL gen_ai semconv span attribute.
+_SEMCONV_REQUEST_ATTRIBUTES = {
+    "frequency_penalty": "gen_ai.request.frequency_penalty",
+    "presence_penalty": "gen_ai.request.presence_penalty",
+    "top_k": "gen_ai.request.top_k",
+    "seed": "gen_ai.request.seed",
+}
+
+# usage_object key -> OTEL gen_ai semconv cache-token span attribute.
+_SEMCONV_CACHE_TOKEN_ATTRIBUTES = {
+    "cache_creation_input_tokens": "gen_ai.usage.cache_creation.input_tokens",
+    "cache_read_input_tokens": "gen_ai.usage.cache_read.input_tokens",
+}
+
+# Name of the consolidated GenAI inference event (replaces the legacy
+# per-message gen_ai.content.prompt / per-choice gen_ai.content.completion).
+_INFERENCE_DETAILS_EVENT_NAME = "gen_ai.client.inference.operation.details"
+
+
+def parse_semconv_opt_in(raw: Optional[str]) -> Set[OTELSemconvCategory]:
+    """Parse the comma-separated OTEL_SEMCONV_STABILITY_OPT_IN value into the
+    set of recognized categories. Unknown tokens are ignored per the spec."""
+    if not raw:
+        return set()
+    return {
+        _SEMCONV_CATEGORY_BY_VALUE[token]
+        for token in (part.strip() for part in raw.split(","))
+        if token in _SEMCONV_CATEGORY_BY_VALUE
+    }
+
+
+class OTELGenAISemconvMixin:
+    """OTEL GenAI ``gen_ai_latest_experimental`` semantic-convention behavior.
+
+    Mixed into ``OpenTelemetry`` (its only host). Every member is internal to
+    the OTEL integration; the leading underscore marks "subsystem-internal",
+    not "class-private" (the host lives in a sibling module).
+
+    Members the host calls (the mixin -> host contract):
+
+    - ``_gen_ai_semconv_latest_experimental`` -- opt-in gate; guards every
+      semconv code path in ``opentelemetry.py``.
+    - ``_gen_ai_operation_name`` -- LiteLLM ``call_type`` -> spec
+      ``gen_ai.operation.name``.
+    - ``_set_semconv_request_attributes`` /
+      ``_set_semconv_cache_token_attributes`` -- add the ``gen_ai.request.*``
+      / ``gen_ai.usage.cache_*`` span attributes.
+    - ``_emit_inference_details_event`` -- emit the consolidated event.
+
+    Helpers the host must provide (declared under ``TYPE_CHECKING`` below):
+    ``config``, ``safe_set_attribute``, ``_capture_in_event``,
+    ``_transform_messages_to_otel_semantic_conventions``,
+    ``_transform_choices_to_otel_semantic_conventions``, ``_to_ns``,
+    ``_otel_log_types``.
+    """
+
+    if TYPE_CHECKING:
+        config: "OpenTelemetryConfig"
+
+        def safe_set_attribute(self, span: Span, key: str, value: Any) -> None: ...
+
+        def _capture_in_event(self) -> bool: ...
+
+        def _transform_messages_to_otel_semantic_conventions(
+            self, messages: Union[List[dict], str]
+        ) -> List[dict]: ...
+
+        def _transform_choices_to_otel_semantic_conventions(
+            self, choices: List[dict]
+        ) -> List[dict]: ...
+
+        def _to_ns(self, dt: datetime) -> int: ...
+
+        def _otel_log_types(self) -> Tuple[Any, Any]: ...
+
+    @property
+    def _gen_ai_semconv_latest_experimental(self) -> bool:
+        """Whether the ``gen_ai_latest_experimental`` opt-in is active.
+
+        Every semconv behavior is gated on this; ``False`` => legacy output.
+        """
+        return (
+            OTELSemconvCategory.GEN_AI_LATEST_EXPERIMENTAL
+            in self.config.semconv_stability_opt_in
+        )
+
+    @staticmethod
+    def _gen_ai_operation_name(kwargs: dict) -> str:
+        """Map a LiteLLM ``call_type`` to spec ``gen_ai.operation.name``.
+
+        Substring match (e.g. ``aembedding`` -> ``embeddings``); defaults to
+        ``chat``.
+        """
+        call_type = kwargs.get("call_type", "") or ""
+        match call_type:
+            case s if "embedding" in s:
+                return "embeddings"
+            case s if "text_completion" in s:
+                return "text_completion"
+            case _:
+                return "chat"
+
+    def _set_semconv_request_attributes(
+        self, span: Span, optional_params: dict
+    ) -> None:
+        """Add ``gen_ai.request.*`` span attributes from ``optional_params``.
+
+        Covers the sampling params plus the conditionally-required
+        ``stop_sequences`` / ``stream`` / ``choice.count`` per the spec.
+        """
+        for source_key, semconv_key in _SEMCONV_REQUEST_ATTRIBUTES.items():
+            value = optional_params.get(source_key)
+            if value is not None:
+                self.safe_set_attribute(span=span, key=semconv_key, value=value)
+
+        stop = optional_params.get("stop")
+        if stop is not None:
+            # Spec types this as string[]. safe_set_attribute coerces to a
+            # primitive, so set the array directly via the span API.
+            stop_list = stop if isinstance(stop, list) else [stop]
+            span.set_attribute(
+                "gen_ai.request.stop_sequences", [str(s) for s in stop_list]
+            )
+
+        # Conditionally required: set only when the request is streaming.
+        if optional_params.get("stream"):
+            self.safe_set_attribute(span=span, key="gen_ai.request.stream", value=True)
+
+        # Conditionally required per spec ("if available and != 1"). Valid n is
+        # an int >= 1, so n > 1 is equivalent for conformant input while
+        # suppressing nonsensical values (0, negative, non-int).
+        n = optional_params.get("n")
+        if isinstance(n, int) and n > 1:
+            self.safe_set_attribute(
+                span=span, key="gen_ai.request.choice.count", value=n
+            )
+
+    def _set_semconv_cache_token_attributes(
+        self, span: Span, standard_logging_payload
+    ) -> None:
+        """Add ``gen_ai.usage.cache_*.input_tokens`` from the usage object.
+
+        No-op when the payload or the usage values are missing/zero.
+        """
+        if not standard_logging_payload:
+            return
+        usage = (standard_logging_payload.get("metadata") or {}).get(
+            "usage_object"
+        ) or {}
+        for source_key, semconv_key in _SEMCONV_CACHE_TOKEN_ATTRIBUTES.items():
+            value = usage.get(source_key)
+            if value:
+                self.safe_set_attribute(span=span, key=semconv_key, value=value)
+
+    def _build_inference_details_attrs(
+        self, kwargs: dict, response_obj: dict, provider: str
+    ) -> Dict[str, Any]:
+        """Build the attribute payload for the inference-details event.
+
+        Always includes provider/operation; input/output messages are added
+        only when content capture is enabled and non-empty. Mixin-internal.
+        """
+        attrs: Dict[str, Any] = {
+            "event_name": _INFERENCE_DETAILS_EVENT_NAME,
+            "gen_ai.provider.name": provider,
+            "gen_ai.operation.name": self._gen_ai_operation_name(kwargs),
+        }
+        if not self._capture_in_event():
+            return attrs
+
+        input_messages = self._transform_messages_to_otel_semantic_conventions(
+            kwargs.get("messages") or []
+        )
+        output_messages = self._transform_choices_to_otel_semantic_conventions(
+            response_obj.get("choices", [])
+        )
+        if input_messages:
+            attrs["gen_ai.input.messages"] = safe_dumps(input_messages)
+        if output_messages:
+            attrs["gen_ai.output.messages"] = safe_dumps(output_messages)
+        return attrs
+
+    def _emit_inference_details_event(
+        self,
+        kwargs: dict,
+        response_obj: dict,
+        provider: str,
+        otel_logger,
+        parent_ctx,
+    ) -> None:
+        """Emit the consolidated ``gen_ai.client.inference.operation.details``
+        log event, correlated to the request span via ``parent_ctx``.
+
+        Replaces the legacy per-message / per-choice content events.
+        """
+        LogRecord, SeverityNumber = self._otel_log_types()
+        log_record = LogRecord(
+            timestamp=self._to_ns(datetime.now()),
+            trace_id=parent_ctx.trace_id,
+            span_id=parent_ctx.span_id,
+            trace_flags=parent_ctx.trace_flags,
+            severity_number=SeverityNumber.INFO,
+            severity_text="INFO",
+            body=None,
+            attributes=self._build_inference_details_attrs(
+                kwargs, response_obj, provider
+            ),
+        )
+        otel_logger.emit(log_record)

tests/test_litellm/integrations/test_opentelemetry.py

@@ -18,7 +18,11 @@ from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 
-from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig
+from litellm.integrations.opentelemetry import (
+    OpenTelemetry,
+    OpenTelemetryConfig,
+    OTELSemconvCategory,
+)
 from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
 
 
@@ -545,6 +549,348 @@ class TestOpenTelemetryCaptureMessageContent(unittest.TestCase):
         self.assertTrue(kept._capture_in_event())
 
 
+class TestOpenTelemetrySemconvStability(unittest.TestCase):
+    """OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental opts into
+    semconv-conformant span shape (name, kind, no raw_gen_ai_request child)."""
+
+    @staticmethod
+    def _make(env=None, config_value=None):
+        env_value = env if env is not None else ""
+        with patch.dict(os.environ, {"OTEL_SEMCONV_STABILITY_OPT_IN": env_value}):
+            return OpenTelemetry(
+                config=OpenTelemetryConfig(
+                    exporter="console",
+                    semconv_stability_opt_in=config_value or set(),
+                )
+            )
+
+    def test_default_unset_keeps_legacy_span_name(self):
+        h = self._make()
+        self.assertFalse(h._gen_ai_semconv_latest_experimental)
+        kwargs = {"model": "gpt-4", "call_type": "acompletion"}
+        self.assertEqual(h._get_span_name(kwargs), "litellm_request")
+
+    def test_opt_in_emits_semconv_span_name(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        self.assertTrue(h._gen_ai_semconv_latest_experimental)
+        kwargs = {"model": "gpt-4", "call_type": "acompletion"}
+        self.assertEqual(h._get_span_name(kwargs), "chat gpt-4")
+
+    def test_opt_in_supports_comma_separated_categories(self):
+        h = self._make(env="other_category,gen_ai_latest_experimental")
+        self.assertTrue(h._gen_ai_semconv_latest_experimental)
+
+    def test_opt_in_ignores_unrelated_category(self):
+        h = self._make(env="some_other_category")
+        self.assertFalse(h._gen_ai_semconv_latest_experimental)
+
+    def test_config_field_enables_without_env(self):
+        h = self._make(
+            env="", config_value={OTELSemconvCategory.GEN_AI_LATEST_EXPERIMENTAL}
+        )
+        self.assertTrue(h._gen_ai_semconv_latest_experimental)
+
+    def test_config_field_unions_with_env(self):
+        h = self._make(
+            env="gen_ai_latest_experimental",
+            config_value={OTELSemconvCategory.GEN_AI_LATEST_EXPERIMENTAL},
+        )
+        self.assertTrue(h._gen_ai_semconv_latest_experimental)
+
+    def test_operation_name_for_embeddings(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        kwargs = {
+            "model": "text-embedding-3-small",
+            "call_type": "aembedding",
+        }
+        self.assertEqual(h._get_span_name(kwargs), "embeddings text-embedding-3-small")
+
+    def test_operation_name_for_text_completion(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        kwargs = {"model": "babbage-002", "call_type": "atext_completion"}
+        self.assertEqual(h._get_span_name(kwargs), "text_completion babbage-002")
+
+    def test_operation_name_defaults_to_chat(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        kwargs = {"model": "claude-sonnet-4-5", "call_type": "unknown"}
+        self.assertEqual(h._get_span_name(kwargs), "chat claude-sonnet-4-5")
+
+    def test_generation_name_metadata_overrides_semconv_name(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        kwargs = {
+            "model": "gpt-4",
+            "call_type": "acompletion",
+            "litellm_params": {"metadata": {"generation_name": "user-named-span"}},
+        }
+        self.assertEqual(h._get_span_name(kwargs), "user-named-span")
+
+    def test_opt_in_skips_raw_gen_ai_request_span(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        h._maybe_log_raw_request = OpenTelemetry._maybe_log_raw_request.__get__(h)
+        h.tracer = MagicMock()
+        h.set_raw_request_attributes = MagicMock()
+        kwargs = {"litellm_params": {"metadata": {}}}
+        h._maybe_log_raw_request(kwargs, {}, None, None, MagicMock())
+        h.tracer.start_span.assert_not_called()
+
+    def test_semconv_request_attributes_emit_when_present(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        optional_params = {
+            "frequency_penalty": 0.5,
+            "presence_penalty": 0.2,
+            "top_k": 40,
+            "seed": 42,
+            "stop": ["\n\n"],
+            "stream": True,
+            "n": 3,
+        }
+        h._set_semconv_request_attributes(span, optional_params)
+        calls = {
+            c.args[0] if c.args else c.kwargs.get("key"): c
+            for c in span.set_attribute.call_args_list
+        }
+        self.assertIn("gen_ai.request.frequency_penalty", calls)
+        self.assertIn("gen_ai.request.presence_penalty", calls)
+        self.assertIn("gen_ai.request.top_k", calls)
+        self.assertIn("gen_ai.request.seed", calls)
+        self.assertIn("gen_ai.request.stop_sequences", calls)
+        self.assertIn("gen_ai.request.stream", calls)
+        self.assertIn("gen_ai.request.choice.count", calls)
+
+    def test_semconv_request_choice_count_omitted_when_one(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        h._set_semconv_request_attributes(span, {"n": 1})
+        keys = {c.args[0] for c in span.set_attribute.call_args_list if c.args}
+        self.assertNotIn("gen_ai.request.choice.count", keys)
+
+    def test_semconv_request_choice_count_omitted_for_invalid_n(self):
+        # n must be a valid count (int > 1); 0/negative/non-int are suppressed.
+        h = self._make(env="gen_ai_latest_experimental")
+        for bad_n in (0, -1, "2", 2.0):
+            span = MagicMock()
+            h._set_semconv_request_attributes(span, {"n": bad_n})
+            keys = {c.args[0] for c in span.set_attribute.call_args_list if c.args}
+            self.assertNotIn(
+                "gen_ai.request.choice.count", keys, f"n={bad_n!r} should be omitted"
+            )
+
+    def _stream_calls(self, span):
+        return [
+            c
+            for c in span.set_attribute.call_args_list
+            if c.args and c.args[0] == "gen_ai.request.stream"
+        ]
+
+    def test_semconv_request_stream_emitted_as_bool_when_streaming(self):
+        # Conditionally required per spec: present (as bool True) only when streaming.
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        h._set_semconv_request_attributes(span, {"stream": True})
+        stream_calls = self._stream_calls(span)
+        self.assertEqual(len(stream_calls), 1)
+        self.assertIs(stream_calls[0].args[1], True)
+
+    def test_semconv_request_stream_omitted_when_not_streaming(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        h._set_semconv_request_attributes(span, {"stream": False})
+        self.assertEqual(self._stream_calls(span), [])
+
+    def test_semconv_request_stop_sequences_normalizes_string_to_list(self):
+        # Spec types gen_ai.request.stop_sequences as string[]; a scalar stop
+        # is wrapped, and the value is a real list (not a JSON-encoded string).
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        h._set_semconv_request_attributes(span, {"stop": "STOP_TOKEN"})
+        stop_calls = [
+            c
+            for c in span.set_attribute.call_args_list
+            if c.args and c.args[0] == "gen_ai.request.stop_sequences"
+        ]
+        self.assertEqual(len(stop_calls), 1)
+        self.assertEqual(stop_calls[0].args[1], ["STOP_TOKEN"])
+
+    def test_semconv_cache_token_attributes(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        std_log = {
+            "metadata": {
+                "usage_object": {
+                    "cache_creation_input_tokens": 12,
+                    "cache_read_input_tokens": 34,
+                }
+            }
+        }
+        h._set_semconv_cache_token_attributes(span, std_log)
+        keys = {
+            c.args[0]: c.args[1] for c in span.set_attribute.call_args_list if c.args
+        }
+        self.assertEqual(keys.get("gen_ai.usage.cache_creation.input_tokens"), 12)
+        self.assertEqual(keys.get("gen_ai.usage.cache_read.input_tokens"), 34)
+
+    def test_semconv_cache_token_attributes_handles_none_metadata(self):
+        # standard_logging_payload["metadata"] = None should not crash.
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        h._set_semconv_cache_token_attributes(span, {"metadata": None})
+        span.set_attribute.assert_not_called()
+
+    def test_semconv_cache_token_attributes_omitted_when_zero(self):
+        h = self._make(env="gen_ai_latest_experimental")
+        span = MagicMock()
+        std_log = {
+            "metadata": {
+                "usage_object": {
+                    "cache_creation_input_tokens": 0,
+                    "cache_read_input_tokens": 0,
+                }
+            }
+        }
+        h._set_semconv_cache_token_attributes(span, std_log)
+        keys = {c.args[0] for c in span.set_attribute.call_args_list if c.args}
+        self.assertNotIn("gen_ai.usage.cache_creation.input_tokens", keys)
+        self.assertNotIn("gen_ai.usage.cache_read.input_tokens", keys)
+
+    def _set_attributes_keys(self, h):
+        """Run set_attributes with a minimal chat payload; return {key: value}."""
+        span = MagicMock()
+        kwargs = {
+            "model": "gpt-4",
+            "messages": [{"role": "user", "content": "hi"}],
+            "optional_params": {},
+            "litellm_params": {"custom_llm_provider": "openai"},
+            "standard_logging_object": {
+                "id": "test-id",
+                "call_type": "completion",
+                "metadata": {},
+            },
+        }
+        response_obj = {"id": "r", "model": "gpt-4", "choices": []}
+        h.set_attributes(span=span, kwargs=kwargs, response_obj=response_obj)
+        return {
+            c.args[0]: c.args[1] for c in span.set_attribute.call_args_list if c.args
+        }
+
+    def test_semconv_mode_emits_provider_name_not_system(self):
+        # Latest-experimental semconv replaced gen_ai.system with
+        # gen_ai.provider.name; only the conformant key is emitted.
+        keys = self._set_attributes_keys(self._make(env="gen_ai_latest_experimental"))
+        self.assertEqual(keys.get("gen_ai.provider.name"), "openai")
+        self.assertNotIn("gen_ai.system", keys)
+
+    def test_legacy_mode_emits_system_not_provider_name(self):
+        keys = self._set_attributes_keys(self._make())
+        self.assertEqual(keys.get("gen_ai.system"), "openai")
+        self.assertNotIn("gen_ai.provider.name", keys)
+
+    def test_opt_in_emits_consolidated_inference_details_event(self):
+        from opentelemetry import _logs
+        from opentelemetry._logs._internal import ProxyLoggerProvider
+
+        log_exporter = InMemoryLogExporter()
+        # Make _init_logs see a non-SDK global (the proxy default) so it
+        # falls into the create_new branch and consults _get_log_exporter,
+        # which we patch to return our in-memory exporter.
+        with (
+            patch.dict(
+                os.environ,
+                {"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"},
+            ),
+            patch.object(
+                _logs, "get_logger_provider", return_value=ProxyLoggerProvider()
+            ),
+            patch.object(_logs, "set_logger_provider"),
+            patch.object(OpenTelemetry, "_get_log_exporter", return_value=log_exporter),
+        ):
+            h = OpenTelemetry(
+                config=OpenTelemetryConfig(exporter="console", enable_events=True)
+            )
+        h.message_logging = True
+
+        kwargs = {
+            "model": "gpt-4",
+            "call_type": "acompletion",
+            "messages": [{"role": "user", "content": "hi"}],
+            "litellm_params": {"custom_llm_provider": "openai"},
+        }
+        response_obj = {
+            "choices": [
+                {
+                    "message": {"role": "assistant", "content": "hello"},
+                    "finish_reason": "stop",
+                }
+            ]
+        }
+        span = h.tracer.start_span("test")
+        h._emit_semantic_logs(kwargs, response_obj, span)
+        span.end()
+        h._logger_provider.force_flush(2000)
+
+        records = [r.log_record for r in log_exporter.get_finished_logs()]
+        # Exactly ONE inference details event, not the legacy per-message/choice pair.
+        self.assertEqual(len(records), 1)
+        attrs = dict(records[0].attributes or {})
+        self.assertEqual(
+            attrs["event_name"], "gen_ai.client.inference.operation.details"
+        )
+        self.assertEqual(attrs["gen_ai.provider.name"], "openai")
+        self.assertEqual(attrs["gen_ai.operation.name"], "chat")
+        self.assertIn("gen_ai.input.messages", attrs)
+        self.assertIn("gen_ai.output.messages", attrs)
+
+    def test_opt_in_inference_details_respects_content_kill_switch(self):
+        from opentelemetry import _logs
+        from opentelemetry._logs._internal import ProxyLoggerProvider
+
+        log_exporter = InMemoryLogExporter()
+        with (
+            patch.dict(
+                os.environ,
+                {"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"},
+            ),
+            patch("litellm.turn_off_message_logging", True),
+            patch.object(
+                _logs, "get_logger_provider", return_value=ProxyLoggerProvider()
+            ),
+            patch.object(_logs, "set_logger_provider"),
+            patch.object(OpenTelemetry, "_get_log_exporter", return_value=log_exporter),
+        ):
+            h = OpenTelemetry(
+                config=OpenTelemetryConfig(exporter="console", enable_events=True)
+            )
+            h.message_logging = True
+
+            kwargs = {
+                "model": "gpt-4",
+                "call_type": "acompletion",
+                "messages": [{"role": "user", "content": "private prompt"}],
+                "litellm_params": {"custom_llm_provider": "openai"},
+            }
+            response_obj = {
+                "choices": [
+                    {
+                        "message": {
+                            "role": "assistant",
+                            "content": "private completion",
+                        },
+                        "finish_reason": "stop",
+                    }
+                ]
+            }
+            span = h.tracer.start_span("test")
+            h._emit_semantic_logs(kwargs, response_obj, span)
+            span.end()
+            h._logger_provider.force_flush(2000)
+
+        records = [r.log_record for r in log_exporter.get_finished_logs()]
+        self.assertEqual(len(records), 1)
+        attrs = dict(records[0].attributes or {})
+        self.assertNotIn("gen_ai.input.messages", attrs)
+        self.assertNotIn("gen_ai.output.messages", attrs)
+
+
 class TestOpenTelemetry(unittest.TestCase):
     POLL_INTERVAL = 0.05
     POLL_TIMEOUT = 2.0