litellm/tests/unit/integrations/test_langfuse_otel.py

import asyncio
import base64
import contextvars
import json
import os
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from types import MappingProxyType
from typing import Final
from unittest.mock import MagicMock, patch

import pytest
from openai.types.responses import ResponseFunctionToolCall
from opentelemetry import trace
from opentelemetry.context import Context
from opentelemetry.sdk.trace import SpanLimits, TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.trace import Span

import litellm
from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL, REDACTED_BY_LITELLM
from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger
from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME, OpenTelemetry, OpenTelemetryConfig
from litellm.integrations.otel.model.destination import OtelDestination
from litellm.integrations.otel.plumbing.context import set_request_destinations
from litellm.proxy._types import UserAPIKeyAuth
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import Choices, ModelResponse, StandardCallbackDynamicParams


class TestLangfuseOtelIntegration:
    def test_get_langfuse_otel_config_with_required_env_vars(self):
        """Test that config is created correctly with required environment variables."""
        # Clean environment of any Langfuse-related variables
        env_vars_to_clean = [
            "LANGFUSE_HOST",
            "OTEL_EXPORTER_OTLP_ENDPOINT",
            "OTEL_EXPORTER_OTLP_HEADERS",
        ]
        with patch.dict(
            os.environ,
            {
                "LANGFUSE_PUBLIC_KEY": "test_public_key",
                "LANGFUSE_SECRET_KEY": "test_secret_key",
            },
            clear=False,
        ):
            # Remove any existing Langfuse variables
            for var in env_vars_to_clean:
                if var in os.environ:
                    del os.environ[var]

            config = LangfuseOtelLogger.get_langfuse_otel_config()

            assert isinstance(config, OpenTelemetryConfig)
            assert config.exporter == "otlp_http"
            assert "Authorization=Basic" in config.headers
            # Note: We no longer set os.environ explicitly to avoid leakage
            # assert os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") == "https://us.cloud.langfuse.com/api/public/otel"
            # assert "Authorization=Basic" in os.environ.get("OTEL_EXPORTER_OTLP_HEADERS", "")

    def test_get_langfuse_otel_config_missing_keys(self):
        """Test that ValueError is raised when required keys are missing."""
        with patch.dict(os.environ, {}, clear=True):
            with pytest.raises(
                ValueError,
                match="LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY must be set",
            ):
                LangfuseOtelLogger.get_langfuse_otel_config()

    def test_get_langfuse_otel_config_with_eu_host(self):
        """Test config with EU host."""
        with patch.dict(
            os.environ,
            {
                "LANGFUSE_PUBLIC_KEY": "test_public_key",
                "LANGFUSE_SECRET_KEY": "test_secret_key",
                "LANGFUSE_HOST": "https://cloud.langfuse.com",
            },
            clear=False,
        ):
            config = LangfuseOtelLogger.get_langfuse_otel_config()
            # Endpoint assertion removed as side effect is gone
            assert isinstance(config, OpenTelemetryConfig)

    def test_get_langfuse_otel_config_with_custom_host(self):
        """Test config with custom host."""
        with patch.dict(
            os.environ,
            {
                "LANGFUSE_PUBLIC_KEY": "test_public_key",
                "LANGFUSE_SECRET_KEY": "test_secret_key",
                "LANGFUSE_HOST": "https://my-langfuse.com",
            },
            clear=False,
        ):
            config = LangfuseOtelLogger.get_langfuse_otel_config()
            # Endpoint assertion removed as side effect is gone
            assert isinstance(config, OpenTelemetryConfig)

    def test_get_langfuse_otel_config_with_host_no_protocol(self):
        """Test config with custom host without protocol."""
        with patch.dict(
            os.environ,
            {
                "LANGFUSE_PUBLIC_KEY": "test_public_key",
                "LANGFUSE_SECRET_KEY": "test_secret_key",
                "LANGFUSE_HOST": "my-langfuse.com",
            },
            clear=False,
        ):
            config = LangfuseOtelLogger.get_langfuse_otel_config()
            # Endpoint assertion removed as side effect is gone
            assert isinstance(config, OpenTelemetryConfig)

    def test_set_langfuse_otel_attributes(self):
        """Test that set_langfuse_otel_attributes calls the Arize utils function."""
        from litellm.integrations.langfuse.langfuse_otel_attributes import (
            LangfuseLLMObsOTELAttributes,
        )

        mock_span = MagicMock()
        mock_kwargs = {"test": "kwargs"}
        mock_response = {"test": "response"}

        with patch(
            "litellm.integrations.arize._utils.set_attributes"
        ) as mock_set_attributes:
            LangfuseOtelLogger.set_langfuse_otel_attributes(
                mock_span, mock_kwargs, mock_response
            )

            mock_set_attributes.assert_called_once_with(
                mock_span, mock_kwargs, mock_response, LangfuseLLMObsOTELAttributes
            )
            mock_span.set_attribute.assert_any_call(
                "langfuse.observation.type", "generation"
            )

    def test_set_langfuse_environment_attribute(self):
        """Test that Langfuse environment is set correctly when environment variable is present."""
        mock_span = MagicMock()
        mock_kwargs = {"test": "kwargs"}
        test_env = "staging"

        with patch.dict(os.environ, {"LANGFUSE_TRACING_ENVIRONMENT": test_env}):
            with patch(
                "litellm.integrations.arize._utils.safe_set_attribute"
            ) as mock_safe_set_attribute:
                LangfuseOtelLogger._set_langfuse_specific_attributes(
                    mock_span, mock_kwargs, {}
                )

                # safe_set_attribute(span, key, value) → positional args
                mock_safe_set_attribute.assert_any_call(
                    mock_span, "langfuse.environment", test_env
                )

    def test_set_langfuse_environment_attribute_prefers_dynamic_param(self):
        """Per-key/team langfuse_environment beats the deployment env var."""

        class _RecordingSpan:
            def __init__(self):
                self.attributes = {}

            def set_attribute(self, key, value):
                self.attributes[key] = value

        span = _RecordingSpan()
        mock_kwargs = {
            "standard_callback_dynamic_params": {
                "langfuse_environment": "team-a-env"
            }
        }

        with patch.dict(
            os.environ, {"LANGFUSE_TRACING_ENVIRONMENT": "deployment-wide"}
        ):
            LangfuseOtelLogger._set_langfuse_specific_attributes(
                span, mock_kwargs, {}
            )

        assert span.attributes["langfuse.environment"] == "team-a-env"

    @pytest.mark.parametrize(
        "completion_start_time",
        [
            datetime(2026, 10, 9, 12, 0, 0, 250000, tzinfo=timezone.utc),
            datetime(2026, 10, 9, 12, 0, 0, 250000),
        ],
        ids=["aware", "naive_local_time"],
    )
    def test_set_langfuse_specific_attributes_completion_start_time(self, completion_start_time: datetime):
        """The first-token time goes out the way the Langfuse SDK sends it: a JSON string, with an explicit offset."""
        span: Final = TracerProvider().get_tracer(__name__).start_span("litellm_request")

        LangfuseOtelLogger._set_langfuse_specific_attributes(span, {"completion_start_time": completion_start_time}, {})

        sent: Final = datetime.fromisoformat(json.loads(span.attributes["langfuse.observation.completion_start_time"]))
        assert sent.tzinfo is not None
        assert sent.timestamp() == completion_start_time.timestamp()

    def test_set_langfuse_specific_attributes_without_completion_start_time(self):
        span: Final = TracerProvider().get_tracer(__name__).start_span("litellm_request")

        LangfuseOtelLogger._set_langfuse_specific_attributes(span, {}, {})

        assert "langfuse.observation.completion_start_time" not in span.attributes

    def test_extract_langfuse_metadata_basic(self):
        """Ensure metadata is correctly pulled from litellm_params."""
        metadata_in = {"generation_name": "my-gen", "custom": "data"}
        kwargs = {"litellm_params": {"metadata": metadata_in}}
        extracted = LangfuseOtelLogger._extract_langfuse_metadata(kwargs)
        assert extracted == metadata_in

    def test_extract_langfuse_metadata_with_header_enrichment(self, monkeypatch):
        """_extract_langfuse_metadata should call LangFuseLogger.add_metadata_from_header when available."""
        import sys
        import types

        # Build a stub module + class on-the-fly
        stub_module = types.ModuleType("litellm.integrations.langfuse.langfuse")

        class StubLFLogger:
            @staticmethod
            def add_metadata_from_header(litellm_params, metadata):
                # Echo back existing metadata plus a marker
                return {**metadata, "enriched": True}

        stub_module.LangFuseLogger = StubLFLogger  # type: ignore

        # Register stub in sys.modules so import inside method succeeds.
        # Use monkeypatch so the real module is restored after the test runs,
        # preventing sys.modules corruption that would break patch() targets in
        # later tests (the patch would hit the stub while the real module's
        # globals remain unpatched).
        monkeypatch.setitem(sys.modules, "litellm.integrations.langfuse.langfuse", stub_module)  # type: ignore

        kwargs = {"litellm_params": {"metadata": {"foo": "bar"}}}
        extracted = LangfuseOtelLogger._extract_langfuse_metadata(kwargs)
        assert extracted.get("foo") == "bar"
        assert extracted.get("enriched") is True

    def test_set_langfuse_specific_attributes_metadata(self):
        """Verify every supported metadata key maps to the correct OTEL attribute and complex types are JSON-serialised."""
        # Build a sample metadata payload covering all mappings
        metadata = {
            "generation_name": "gen-name",
            "generation_id": "gen-id",
            "parent_observation_id": "parent-id",
            "version": "v1",
            "mask_input": True,
            "mask_output": False,
            "trace_user_id": "user-123",
            "session_id": "sess-456",
            "tags": ["tagA", "tagB"],
            "trace_name": "trace-name",
            "trace_id": "trace-id",
            "trace_metadata": {"k": "v"},
            "trace_version": "t-ver",
            "trace_release": "rel-1",
            "existing_trace_id": "existing-id",
            "update_trace_keys": ["key1", "key2"],
            "debug_langfuse": True,
        }
        kwargs = {"litellm_params": {"metadata": metadata}}

        # Capture calls to safe_set_attribute
        with patch(
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(
                MagicMock(), kwargs, None
            )

            # Build expected calls manually for clarity
            from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes

            expected = {
                LangfuseSpanAttributes.GENERATION_NAME.value: "gen-name",
                LangfuseSpanAttributes.GENERATION_ID.value: "gen-id",
                LangfuseSpanAttributes.PARENT_OBSERVATION_ID.value: "parent-id",
                LangfuseSpanAttributes.VERSION.value: "t-ver",
                LangfuseSpanAttributes.MASK_INPUT.value: True,
                LangfuseSpanAttributes.MASK_OUTPUT.value: False,
                LangfuseSpanAttributes.TRACE_USER_ID.value: "user-123",
                LangfuseSpanAttributes.SESSION_ID.value: "sess-456",
                # Lists / dicts should be JSON strings
                LangfuseSpanAttributes.TAGS.value: json.dumps(["tagA", "tagB"]),
                LangfuseSpanAttributes.TRACE_NAME.value: "trace-name",
                LangfuseSpanAttributes.TRACE_ID.value: "traceid",  # stripped dashes
                LangfuseSpanAttributes.TRACE_METADATA.value: json.dumps({"k": "v"}),
                LangfuseSpanAttributes.RELEASE.value: "rel-1",
                LangfuseSpanAttributes.EXISTING_TRACE_ID.value: "existing-id",
                LangfuseSpanAttributes.UPDATE_TRACE_KEYS.value: json.dumps(
                    ["key1", "key2"]
                ),
                LangfuseSpanAttributes.DEBUG_LANGFUSE.value: True,
            }

            # Flatten the actual calls into {key: value}
            actual = {
                call.args[1]: call.args[2]  # (span, key, value)
                for call in mock_safe_set_attribute.call_args_list
            }

            assert (
                actual == expected
            ), "Mismatch between expected and actual OTEL attribute mapping."

    @pytest.mark.parametrize(
        "metadata, expected_version",
        [
            (
                {"version": "v-observation", "trace_version": "v-trace"},
                "v-trace",
            ),
            ({"trace_version": "v-trace"}, "v-trace"),
            ({"version": "v-observation"}, "v-observation"),
            ({"version": "v-observation", "trace_version": ""}, ""),
            ({}, None),
        ],
        ids=[
            "trace-version-wins-as-documented",
            "trace-only",
            "observation-version-is-the-fallback",
            "empty-trace-version-is-not-absent",
            "neither-key-emits-nothing",
        ],
    )
    def test_version_emitted_on_langfuse_v4_key(self, metadata, expected_version):
        kwargs = {"litellm_params": {"metadata": {"trace_release": "rel-9", **metadata}}}

        with patch(
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(
                MagicMock(), kwargs, None
            )

        emitted = {
            call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list
        }

        if expected_version is None:
            assert "langfuse.version" not in emitted
        else:
            assert emitted["langfuse.version"] == expected_version
        assert emitted["langfuse.release"] == "rel-9"
        for retired_key in (
            "langfuse.generation.version",
            "langfuse.trace.version",
            "langfuse.trace.release",
        ):
            assert retired_key not in emitted

    def test_set_langfuse_specific_attributes_with_content(self):
        """Test that _set_langfuse_specific_attributes correctly sets observation.output with regular content response."""
        from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes
        from litellm.types.utils import Choices, ModelResponse

        # Create response with content
        response_obj = ModelResponse(
            id="chatcmpl-test",
            model="gpt-4o",
            choices=[
                Choices(
                    finish_reason="stop",
                    message={
                        "role": "assistant",
                        "content": "The weather in Tokyo is sunny.",
                    },
                )
            ],
        )

        kwargs = {
            "messages": [{"role": "user", "content": "What's the weather in Tokyo?"}],
        }

        with patch(
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(
                MagicMock(), kwargs, response_obj
            )

            expect_output = {
                LangfuseSpanAttributes.OBSERVATION_INPUT.value: [
                    {"role": "user", "content": "What's the weather in Tokyo?"}
                ],
                LangfuseSpanAttributes.OBSERVATION_OUTPUT.value: {
                    "role": "assistant",
                    "content": "The weather in Tokyo is sunny.",
                },
            }

            # Flatten the actual calls into {key: value}
            actual = {
                call.args[1]: json.loads(call.args[2])
                for call in mock_safe_set_attribute.call_args_list
                if call.args[1] in expect_output
            }

            assert (
                actual == expect_output
            ), "Mismatch in observation input/output OTEL attributes."

    def test_set_langfuse_specific_attributes_with_tool_calls(self):
        """Test that _set_langfuse_specific_attributes correctly sets observation.output with tool calls in Langfuse format."""
        from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes
        from litellm.types.utils import (
            ChatCompletionMessageToolCall,
            Choices,
            Function,
            ModelResponse,
        )

        # Create response with tool calls
        response_obj = ModelResponse(
            id="chatcmpl-test",
            model="gpt-4o",
            choices=[
                Choices(
                    finish_reason="tool_calls",
                    message={
                        "role": "assistant",
                        "content": None,
                        "tool_calls": [
                            ChatCompletionMessageToolCall(
                                function=Function(
                                    arguments='{"location":"Tokyo"}', name="get_weather"
                                ),
                                id="call_123",
                                type="function",
                            )
                        ],
                    },
                )
            ],
        )

        with patch(
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(
                MagicMock(), {}, response_obj
            )

            expected = {
                LangfuseSpanAttributes.OBSERVATION_OUTPUT.value: [
                    {
                        "id": "chatcmpl-test",
                        "name": "get_weather",
                        "arguments": {"location": "Tokyo"},
                        "call_id": "call_123",
                        "type": "function_call",
                    }
                ]
            }

            # Flatten the actual calls into {key: value}
            actual = {
                call.args[1]: json.loads(call.args[2])
                for call in mock_safe_set_attribute.call_args_list
                if call.args[1] in expected
            }
            assert (
                actual == expected
            ), "Mismatch in observation output OTEL attribute for tool calls."

    def test_construct_dynamic_otel_headers_with_langfuse_keys(self):
        """Test that construct_dynamic_otel_headers creates proper auth headers when langfuse keys are provided."""
        from litellm.types.utils import StandardCallbackDynamicParams

        # Create dynamic params with langfuse keys
        dynamic_params = StandardCallbackDynamicParams(
            langfuse_public_key="test_public_key", langfuse_secret_key="test_secret_key"
        )

        logger = LangfuseOtelLogger()
        result = logger.construct_dynamic_otel_headers(dynamic_params)

        # Should return a dict with otlp_auth_headers
        assert result is not None
        assert "Authorization" in result

        # The auth header should contain the basic auth format
        auth_header = result["Authorization"]
        assert auth_header.startswith("Basic ")

        # Verify the header format by decoding
        import base64

        # Extract the base64 part from "Authorization=Basic <base64>"
        base64_part = auth_header.replace("Basic ", "")
        decoded = base64.b64decode(base64_part).decode()

        assert decoded == "test_public_key:test_secret_key"

    def test_construct_dynamic_otel_headers_empty_params(self):
        """Test that construct_dynamic_otel_headers returns empty dict when no langfuse keys are provided."""
        from litellm.types.utils import StandardCallbackDynamicParams

        # Create dynamic params without langfuse keys
        dynamic_params = StandardCallbackDynamicParams()

        logger = LangfuseOtelLogger()
        result = logger.construct_dynamic_otel_headers(dynamic_params)

        # Should return an empty dict
        assert result == {}

    def test_get_langfuse_otel_config_with_otel_host_priority(self):
        """LANGFUSE_OTEL_HOST should take priority over LANGFUSE_HOST."""
        with patch.dict(
            os.environ,
            {
                "LANGFUSE_PUBLIC_KEY": "test_public_key",
                "LANGFUSE_SECRET_KEY": "test_secret_key",
                "LANGFUSE_HOST": "https://should-not-be-used.com",
                "LANGFUSE_OTEL_HOST": "https://otel-host.com",
            },
            clear=False,
        ):
            config = LangfuseOtelLogger.get_langfuse_otel_config()
            assert isinstance(config, OpenTelemetryConfig)
            # Endpoint assertion removed as side effect is gone


def _exported_auth_headers(exporter):
    import requests_mock
    from opentelemetry.sdk.trace.export import SpanExportResult

    with requests_mock.Mocker() as collector:
        collector.post(requests_mock.ANY, status_code=200)
        assert exporter.export(()) is SpanExportResult.SUCCESS
    return [
        (
            request.url,
            {name: request.headers[name] for name in ("Authorization", "x-langfuse-ingestion-version")},
        )
        for request in collector.request_history
    ]


class TestLangfuseOtelKeyDynamicConfig:
    """Key/team-scoped Langfuse credentials must define the full export target
    (OTLP endpoint + auth), not just auth headers on the init-time exporter."""

    CLEAN_ENV_VARS = [
        "LANGFUSE_PUBLIC_KEY",
        "LANGFUSE_SECRET_KEY",
        "LANGFUSE_HOST",
        "LANGFUSE_OTEL_HOST",
        "OTEL_EXPORTER",
        "OTEL_EXPORTER_OTLP_PROTOCOL",
        "OTEL_ENDPOINT",
        "OTEL_EXPORTER_OTLP_ENDPOINT",
        "OTEL_HEADERS",
        "OTEL_EXPORTER_OTLP_HEADERS",
    ]

    def _clean_env(self):
        cleaned = {k: v for k, v in os.environ.items() if k not in self.CLEAN_ENV_VARS}
        return patch.dict(os.environ, cleaned, clear=True)

    def _dynamic_params(self, **overrides):
        from litellm.types.utils import StandardCallbackDynamicParams

        params = {
            "langfuse_public_key": "key_public",
            "langfuse_secret_key": "key_secret",
            "langfuse_host": "https://langfuse.example.com",
        }
        params.update(overrides)
        return StandardCallbackDynamicParams(**{k: v for k, v in params.items() if v is not None})

    def test_construct_dynamic_otel_config_with_key_credentials(self):
        with self._clean_env():
            logger = LangfuseOtelLogger()
            config = logger.construct_dynamic_otel_config(self._dynamic_params())

        assert config is not None
        assert config.exporter == "otlp_http"
        assert config.endpoint == "https://langfuse.example.com/api/public/otel"

        import base64

        expected_auth = base64.b64encode(b"key_public:key_secret").decode()
        assert config.headers == f"Authorization=Basic {expected_auth},x-langfuse-ingestion-version=4"

    def test_construct_dynamic_otel_config_host_without_protocol(self):
        with self._clean_env():
            logger = LangfuseOtelLogger()
            config = logger.construct_dynamic_otel_config(self._dynamic_params(langfuse_host="langfuse.example.com"))

        assert config is not None
        assert config.endpoint == "https://langfuse.example.com/api/public/otel"

    def test_construct_dynamic_otel_config_defaults_to_us_cloud(self):
        with self._clean_env():
            logger = LangfuseOtelLogger()
            config = logger.construct_dynamic_otel_config(self._dynamic_params(langfuse_host=None))

        assert config is not None
        assert config.endpoint == "https://us.cloud.langfuse.com/api/public/otel"

    def test_construct_dynamic_otel_config_falls_back_to_env_host(self):
        with self._clean_env():
            with patch.dict(os.environ, {"LANGFUSE_HOST": "https://env-host.example.com"}):
                logger = LangfuseOtelLogger()
                config = logger.construct_dynamic_otel_config(self._dynamic_params(langfuse_host=None))

        assert config is not None
        assert config.endpoint == "https://env-host.example.com/api/public/otel"

    def test_construct_dynamic_otel_config_requires_both_keys(self):
        with self._clean_env():
            logger = LangfuseOtelLogger()

            assert logger.construct_dynamic_otel_config(self._dynamic_params(langfuse_secret_key=None)) is None
            assert logger.construct_dynamic_otel_config(self._dynamic_params(langfuse_public_key=None)) is None

    def test_key_dynamic_params_create_otlp_exporter_without_global_env(self):
        """Without global LANGFUSE_* env vars, a request carrying key-scoped Langfuse
        credentials must get a tracer exporting via OTLP HTTP to that key's host."""
        from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
            OTLPSpanExporter,
        )
        from opentelemetry.sdk.trace.export import BatchSpanProcessor

        with self._clean_env():
            logger = LangfuseOtelLogger()
            assert logger.OTEL_EXPORTER == "console"

            tracer = logger.get_tracer_to_use_for_request(
                {"standard_callback_dynamic_params": self._dynamic_params()}
            )

        assert tracer is not logger.tracer
        assert len(logger._tracer_provider_cache) == 1

        provider = next(iter(logger._tracer_provider_cache.values())).provider
        span_processors = provider._active_span_processor._span_processors
        assert len(span_processors) == 1
        assert isinstance(span_processors[0], BatchSpanProcessor)

        exporter = span_processors[0].span_exporter
        assert isinstance(exporter, OTLPSpanExporter)
        import base64

        expected_auth = base64.b64encode(b"key_public:key_secret").decode()
        assert _exported_auth_headers(exporter) == [
            (
                "https://langfuse.example.com/api/public/otel/v1/traces",
                {"Authorization": f"Basic {expected_auth}", "x-langfuse-ingestion-version": "4"},
            )
        ]

    def test_key_dynamic_params_reuse_cached_provider(self):
        with self._clean_env():
            logger = LangfuseOtelLogger()
            kwargs = {"standard_callback_dynamic_params": self._dynamic_params()}
            logger.get_tracer_to_use_for_request(kwargs)
            logger.get_tracer_to_use_for_request(kwargs)

        assert len(logger._tracer_provider_cache) == 1

    def test_no_dynamic_params_keeps_default_tracer(self):
        with self._clean_env():
            logger = LangfuseOtelLogger()
            tracer = logger.get_tracer_to_use_for_request({})

        assert tracer is logger.tracer
        assert logger._tracer_provider_cache == {}

    def test_key_credentials_never_passed_to_debug_logger(self):
        """The span-processor debug logs must receive a redacted header value, so the
        key-scoped Langfuse secret never enters a log record regardless of downstream
        handler configuration, while the exporter still gets the real header."""
        import base64

        from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
            OTLPSpanExporter,
        )

        from litellm.integrations import opentelemetry as otel_module

        secret = base64.b64encode(b"key_public:key_secret").decode()

        recorded_arguments = []

        def _spy(message, *args, **kwargs):
            recorded_arguments.append(" ".join(str(part) for part in (message, *args)))

        with self._clean_env():
            logger = LangfuseOtelLogger()
            with patch.object(otel_module.verbose_logger, "debug", side_effect=_spy):
                logger.get_tracer_to_use_for_request(
                    {"standard_callback_dynamic_params": self._dynamic_params()}
                )

        logged = "\n".join(recorded_arguments)
        assert "initializing span processor" in logged
        assert secret not in logged
        assert f"Basic {secret}" not in logged

        provider = next(iter(logger._tracer_provider_cache.values())).provider
        exporter = provider._active_span_processor._span_processors[0].span_exporter
        assert isinstance(exporter, OTLPSpanExporter)
        assert _exported_auth_headers(exporter) == [
            (
                "https://langfuse.example.com/api/public/otel/v1/traces",
                {"Authorization": f"Basic {secret}", "x-langfuse-ingestion-version": "4"},
            )
        ]


class TestLangfuseOtelResponsesAPI:
    """Test suite for Langfuse OTEL integration with ResponsesAPI"""

    def test_langfuse_otel_with_responses_api(self):
        """Test that Langfuse OTEL logger works with ResponsesAPI responses and logs metadata."""
        # Create a mock ResponsesAPIResponse
        mock_response = ResponsesAPIResponse(
            id="response-123",
            created_at=1234567890,
            output=[
                {
                    "type": "message",
                    "content": [{"type": "text", "text": "Hello from responses API"}],
                }
            ],
            parallel_tool_calls=False,
            tool_choice="auto",
            tools=[],
            top_p=1.0,
        )

        # Create kwargs with metadata that should be logged
        test_metadata = {
            "user_id": "test123",
            "session_id": "abc456",
            "custom_field": "test_value",
            "generation_name": "responses_test_generation",
            "trace_name": "responses_api_trace",
        }

        kwargs = {
            "call_type": "responses",
            "messages": [{"role": "user", "content": "Hello"}],
            "model": "gpt-4o",
            "optional_params": {},
            "litellm_params": {"metadata": test_metadata},
        }

        mock_span = MagicMock()

        from litellm.integrations.langfuse.langfuse_otel_attributes import (
            LangfuseLLMObsOTELAttributes,
        )

        with patch(
            "litellm.integrations.arize._utils.set_attributes"
        ) as mock_set_attributes:
            with patch(
                "litellm.integrations.arize._utils.safe_set_attribute"
            ) as mock_safe_set_attribute:
                logger = LangfuseOtelLogger()
                logger.set_langfuse_otel_attributes(mock_span, kwargs, mock_response)

                # Verify that set_attributes was called for general attributes
                mock_set_attributes.assert_called_once_with(
                    mock_span, kwargs, mock_response, LangfuseLLMObsOTELAttributes
                )

                # Verify that Langfuse-specific attributes were set
                mock_safe_set_attribute.assert_any_call(
                    mock_span, "langfuse.generation.name", "responses_test_generation"
                )
                mock_safe_set_attribute.assert_any_call(
                    mock_span, "langfuse.trace.name", "responses_api_trace"
                )

    def test_responses_api_metadata_extraction(self):
        """Test that metadata is correctly extracted from ResponsesAPI kwargs."""
        # Clean up any existing module mocks
        import sys

        if "litellm.integrations.langfuse.langfuse" in sys.modules:
            sys.modules["litellm.integrations.langfuse.langfuse"]

        test_metadata = {
            "user_id": "responses_user_123",
            "session_id": "responses_session_456",
            "custom_metadata": {"key": "value"},
            "generation_name": "responses_generation",
            "trace_id": "custom_trace_id",
        }

        kwargs = {
            "call_type": "responses",
            "model": "gpt-4o",
            "litellm_params": {"metadata": test_metadata},
        }

        extracted_metadata = LangfuseOtelLogger._extract_langfuse_metadata(kwargs)

        # Verify all expected metadata was extracted (may have additional fields from header enrichment)
        for key, value in test_metadata.items():
            assert extracted_metadata[key] == value

        assert extracted_metadata["user_id"] == "responses_user_123"
        assert extracted_metadata["generation_name"] == "responses_generation"
        assert extracted_metadata["trace_id"] == "custom_trace_id"

    def test_responses_api_langfuse_specific_attributes(self):
        """Test that ResponsesAPI metadata maps correctly to Langfuse OTEL attributes."""
        metadata = {
            "generation_name": "responses_gen",
            "generation_id": "resp_gen_123",
            "trace_name": "responses_trace",
            "trace_user_id": "resp_user_456",
            "session_id": "resp_session_789",
            "tags": ["responses", "api", "test"],
            "trace_metadata": {"source": "responses_api", "version": "1.0"},
        }

        kwargs = {"call_type": "responses", "litellm_params": {"metadata": metadata}}

        mock_span = MagicMock()

        with patch(
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(mock_span, kwargs, {})

            # Verify specific attributes were set
            from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes

            expected_calls = [
                (
                    mock_span,
                    LangfuseSpanAttributes.GENERATION_NAME.value,
                    "responses_gen",
                ),
                (mock_span, LangfuseSpanAttributes.GENERATION_ID.value, "resp_gen_123"),
                (mock_span, LangfuseSpanAttributes.TRACE_NAME.value, "responses_trace"),
                (
                    mock_span,
                    LangfuseSpanAttributes.TRACE_USER_ID.value,
                    "resp_user_456",
                ),
                (
                    mock_span,
                    LangfuseSpanAttributes.SESSION_ID.value,
                    "resp_session_789",
                ),
                (
                    mock_span,
                    LangfuseSpanAttributes.TAGS.value,
                    json.dumps(["responses", "api", "test"]),
                ),
                (
                    mock_span,
                    LangfuseSpanAttributes.TRACE_METADATA.value,
                    json.dumps({"source": "responses_api", "version": "1.0"}),
                ),
            ]

            for expected_call in expected_calls:
                mock_safe_set_attribute.assert_any_call(*expected_call)

    def test_responses_api_with_output(self):
        """Test Langfuse OTEL logger with Responses API output (reasoning + message)."""
        from openai.types.responses import (
            ResponseReasoningItem,
            ResponseOutputMessage,
            ResponseOutputText,
        )
        from openai.types.responses.response_reasoning_item import Summary
        from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes

        # Create Responses API response with reasoning and message
        response_obj = ResponsesAPIResponse(
            id="response-456",
            created_at=1625247600,
            output=[
                ResponseReasoningItem(
                    id="reasoning-001",
                    type="reasoning",
                    summary=[
                        Summary(
                            text="Let me analyze this problem step by step...",
                            type="summary_text",
                        ),
                        Summary(
                            text="Now checking the forecast data.",
                            type="summary_text",
                        ),
                    ],
                ),
                ResponseOutputMessage(
                    id="msg-001",
                    type="message",
                    role="assistant",
                    status="completed",
                    content=[
                        ResponseOutputText(
                            annotations=[],
                            text="The weather in San Francisco is sunny, 20°C.",
                            type="output_text",
                        )
                    ],
                ),
            ],
        )

        kwargs = {
            "call_type": "responses",
            "messages": [
                {"role": "user", "content": "What's the weather in San Francisco?"}
            ],
            "model": "gpt-4o",
            "optional_params": {},
        }

        mock_span = MagicMock()

        with patch(
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(
                mock_span, kwargs, response_obj
            )

            # Verify observation output was set
            output_calls = [
                call
                for call in mock_safe_set_attribute.call_args_list
                if call.args[1] == LangfuseSpanAttributes.OBSERVATION_OUTPUT.value
            ]

            assert len(output_calls) > 0, "observation.output should be set"
            output_json = output_calls[0].args[2]
            output_data = json.loads(output_json)

            # Verify output contains reasoning and message
            assert isinstance(output_data, list)
            assert len(output_data) == 3

            assert output_data[0]["role"] == "reasoning_summary"
            assert (
                output_data[0]["content"]
                == "Let me analyze this problem step by step..."
            )
            assert output_data[1]["role"] == "reasoning_summary"
            assert output_data[1]["content"] == "Now checking the forecast data."

            # Verify message
            assert output_data[2]["role"] == "assistant"
            assert (
                output_data[2]["content"]
                == "The weather in San Francisco is sunny, 20°C."
            )

            trace_output_calls: Final = [
                call
                for call in mock_safe_set_attribute.call_args_list
                if call.args[1] == LangfuseSpanAttributes.TRACE_OUTPUT.value
            ]
            assert len(trace_output_calls) > 0, "trace.output should be set"
            assert trace_output_calls[0].args[2] == output_json

    def test_responses_api_with_function_calls(self):
        """Test Langfuse OTEL logger with Responses API function_call output."""
        from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes
        from openai.types.responses import ResponseFunctionToolCall

        # Create Responses API response with function call
        response_obj = ResponsesAPIResponse(
            id="response-789",
            created_at=1625247700,
            output=[
                ResponseFunctionToolCall(
                    id="fc-123",
                    type="function_call",
                    name="get_weather",
                    call_id="call-abc",
                    arguments='{"location": "San Francisco", "unit": "celsius"}',
                    status="completed",
                )
            ],
        )

        kwargs = {
            "call_type": "responses",
            "messages": [
                {"role": "user", "content": "What's the weather in San Francisco?"}
            ],
            "model": "gpt-4o",
            "optional_params": {},
        }

        mock_span = MagicMock()

        with patch(
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(
                mock_span, kwargs, response_obj
            )

            # Verify observation output was set
            output_calls = [
                call
                for call in mock_safe_set_attribute.call_args_list
                if call.args[1] == LangfuseSpanAttributes.OBSERVATION_OUTPUT.value
            ]

            assert len(output_calls) > 0, "observation.output should be set"
            output_json = output_calls[0].args[2]
            output_data = json.loads(output_json)

            # Verify output contains function call
            assert isinstance(output_data, list)
            assert len(output_data) == 1

            # Verify function call details
            assert output_data[0]["type"] == "function_call"
            assert output_data[0]["id"] == "fc-123"
            assert output_data[0]["name"] == "get_weather"
            assert output_data[0]["call_id"] == "call-abc"
            assert output_data[0]["arguments"]["location"] == "San Francisco"
            assert output_data[0]["arguments"]["unit"] == "celsius"

    def test_responses_api_function_call_with_redacted_arguments(self):
        """Sentinel arguments (invalid JSON) must not kill the whole observation output."""
        from openai.types.responses import ResponseFunctionToolCall

        from litellm.types.integrations.langfuse_otel import LangfuseSpanAttributes

        response_obj = ResponsesAPIResponse(
            id="response-redacted",
            created_at=1625247700,
            output=[
                ResponseFunctionToolCall(
                    id="fc-redacted",
                    type="function_call",
                    name="get_weather",
                    call_id="call-redacted",
                    arguments="redacted-by-litellm",
                    status="completed",
                )
            ],
        )

        kwargs = {
            "call_type": "responses",
            "messages": [{"role": "user", "content": "What's the weather?"}],
            "model": "gpt-4o",
            "optional_params": {},
        }

        mock_span = MagicMock()

        with patch(  # test-quality-ok: the span attribute sink is the observable boundary; sibling tests in this class stub the same seam
            "litellm.integrations.arize._utils.safe_set_attribute"
        ) as mock_safe_set_attribute:
            LangfuseOtelLogger._set_langfuse_specific_attributes(mock_span, kwargs, response_obj)

            output_calls = [
                call
                for call in mock_safe_set_attribute.call_args_list
                if call.args[1] == LangfuseSpanAttributes.OBSERVATION_OUTPUT.value
            ]

            assert len(output_calls) > 0, "observation.output should still be set"
            output_data = json.loads(output_calls[0].args[2])
            assert output_data[0]["name"] == "get_weather"
            assert output_data[0]["arguments"] == {}


_TRACER: Final = TracerProvider().get_tracer("langfuse-otel-tests")
TRACE_FIELD_KEYS: Final = (
    "langfuse.trace.name",
    "langfuse.trace.input",
    "langfuse.trace.output",
    "langfuse.trace.tags",
)
CALLER: Final = trace.SpanContext(
    trace_id=0x4BF92F3577B34DA6A3CE929D0E0E4736,
    span_id=0x00F067AA0BA902B7,
    is_remote=True,
    trace_flags=trace.TraceFlags(trace.TraceFlags.SAMPLED),
)
CHAT_KWARGS: Final = {
    "call_type": "acompletion",
    "messages": [{"role": "user", "content": "weather?"}],
}


def _chat_response(content: str) -> ModelResponse:
    return ModelResponse(
        id="chatcmpl-derived",
        model="gpt-4o",
        choices=[Choices(finish_reason="stop", message={"role": "assistant", "content": content})],
    )


def _stamped(span: Span, kwargs: Mapping[str, object], response_obj: object = None) -> Mapping[str, object]:
    LangfuseOtelLogger._set_langfuse_specific_attributes(span, kwargs, response_obj)
    span.end()
    return dict(getattr(span, "attributes", None) or {})


def _emitted(kwargs: Mapping[str, object], response_obj: object = None) -> Mapping[str, object]:
    return _stamped(_TRACER.start_span("litellm_request"), kwargs, response_obj)


def _trace_fields(attributes: Mapping[str, object]) -> Mapping[str, object]:
    return {key: attributes[key] for key in TRACE_FIELD_KEYS if key in attributes}


class _PresetCache:
    @staticmethod
    def get_preset_cache_key_from_kwargs(**kwargs: object) -> str:  # kwargs-ok: mirrors Cache's signature
        return "preset-abc"


class TestDerivedTraceFields:
    def test_chat_call_derives_trace_name_input_and_output(self) -> None:
        attributes: Final = _emitted({**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("Sunny."))

        assert attributes["langfuse.trace.name"] == "litellm-acompletion"
        assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"]
        assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"]
        assert json.loads(str(attributes["langfuse.trace.input"])) == [{"role": "user", "content": "weather?"}]
        assert json.loads(str(attributes["langfuse.trace.output"])) == {"role": "assistant", "content": "Sunny."}

    def test_missing_call_type_defaults_to_completion(self) -> None:
        attributes: Final = _emitted(
            {"messages": [{"role": "user", "content": "hi"}], "litellm_params": {"metadata": {}}}
        )

        assert attributes["langfuse.trace.name"] == "litellm-completion"

    def test_caller_trace_name_wins_and_is_the_only_name(self) -> None:
        attributes: Final = _emitted(
            {"call_type": "acompletion", "litellm_params": {"metadata": {"trace_name": "caller-trace"}}}
        )

        assert attributes["langfuse.trace.name"] == "caller-trace"

    def test_existing_trace_id_keeps_caller_tags_and_leaves_derived_trace_fields_off(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_hit"])
        attributes: Final = _emitted(
            {
                **CHAT_KWARGS,
                "cache_hit": True,
                "litellm_params": {"metadata": {"existing_trace_id": "abc123", "tags": ["experiment"]}},
                "standard_logging_object": {"request_tags": ["injected"]},
            },
            _chat_response("hi"),
        )

        assert _trace_fields(attributes) == {"langfuse.trace.tags": json.dumps(["experiment"])}
        assert attributes["langfuse.trace.existing_id"] == "abc123"
        assert "langfuse.observation.input" in attributes
        assert "langfuse.observation.output" in attributes

    @pytest.mark.parametrize(
        ("mask_flags", "masked_trace_keys"),
        [
            ({"mask_input": True}, frozenset({"langfuse.trace.input"})),
            ({"mask_output": True}, frozenset({"langfuse.trace.output"})),
            ({"mask_input": True, "mask_output": True}, frozenset({"langfuse.trace.input", "langfuse.trace.output"})),
            (
                {"mask_input": "true", "mask_output": "TRUE"},
                frozenset({"langfuse.trace.input", "langfuse.trace.output"}),
            ),
            ({"mask_input": "false", "mask_output": "False"}, frozenset()),
            ({"mask_input": False, "mask_output": None}, frozenset()),
        ],
    )
    def test_mask_flags_redact_the_derived_trace_io(
        self, mask_flags: Mapping[str, object], masked_trace_keys: frozenset[str]
    ) -> None:
        attributes: Final = _emitted(
            {
                "call_type": "acompletion",
                "messages": [{"role": "user", "content": "secret question"}],
                "litellm_params": {"metadata": mask_flags},
            },
            _chat_response("secret answer"),
        )
        observation_of: Final = {
            "langfuse.trace.input": attributes["langfuse.observation.input"],
            "langfuse.trace.output": attributes["langfuse.observation.output"],
        }

        assert {key: attributes.get(key) for key in observation_of} == {
            key: REDACTED_BY_LITELLM if key in masked_trace_keys else observation
            for key, observation in observation_of.items()
        }

    def test_responses_api_output_mirrors_observation_output_items(self) -> None:
        response_obj: Final = ResponsesAPIResponse(
            id="response-derived",
            created_at=1625247700,
            output=[
                ResponseFunctionToolCall(
                    id="fc-1",
                    type="function_call",
                    name="get_weather",
                    call_id="call-1",
                    arguments='{"city":"sf"}',
                    status="completed",
                )
            ],
        )
        attributes: Final = _emitted({"call_type": "aresponses", "litellm_params": {"metadata": {}}}, response_obj)

        assert attributes["langfuse.trace.name"] == "litellm-aresponses"
        assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"]
        assert json.loads(str(attributes["langfuse.trace.output"])) == [
            {
                "id": "fc-1",
                "name": "get_weather",
                "call_id": "call-1",
                "type": "function_call",
                "arguments": {"city": "sf"},
            }
        ]

    def test_tags_merge_caller_request_and_default_tags_in_order(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_hit", "user_api_key_alias"])
        attributes: Final = _emitted(
            {
                "call_type": "acompletion",
                "cache_hit": True,
                "litellm_params": {"metadata": {"tags": ["a", "b"], "user_api_key_alias": "k1"}},
                "standard_logging_object": {"request_tags": ["b", "c"]},
            }
        )

        assert json.loads(str(attributes["langfuse.trace.tags"])) == [
            "a",
            "b",
            "c",
            "cache_hit:True",
            "user_api_key_alias:k1",
        ]

    @pytest.mark.parametrize(
        ("metadata", "expected"),
        [
            ({}, ["cache_key:preset-abc"]),
            ({"hidden_params": {"cache_key": "explicit-key"}}, ["cache_key:explicit-key"]),
        ],
        ids=["falls_back_to_the_preset_cache_key", "prefers_hidden_params_over_the_preset"],
    )
    def test_cache_key_default_tag(
        self, monkeypatch: pytest.MonkeyPatch, metadata: Mapping[str, object], expected: Sequence[str]
    ) -> None:
        monkeypatch.setattr(litellm, "langfuse_default_tags", ["cache_key"])
        monkeypatch.setattr(litellm, "cache", _PresetCache())
        attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": metadata}})

        assert json.loads(str(attributes["langfuse.trace.tags"])) == expected

    @pytest.mark.parametrize(
        ("caller_tags", "expected"),
        [
            ("single", ["single", "req"]),
            ("h1, h2", ["h1", "h2", "req"]),
            ('["j1", "j2"]', ["j1", "j2", "req"]),
            (["x", 1], ["x", "1", "req"]),
            ("", ["req"]),
        ],
    )
    def test_caller_tags_are_read_the_way_langfuse_reads_a_raw_tags_value(
        self, caller_tags: object, expected: Sequence[str]
    ) -> None:
        attributes: Final = _emitted(
            {
                "call_type": "acompletion",
                "litellm_params": {"metadata": {"tags": caller_tags}},
                "standard_logging_object": {"request_tags": ["req"]},
            }
        )

        assert json.loads(str(attributes["langfuse.trace.tags"])) == expected

    def test_no_tags_anywhere_leaves_trace_tags_unset(self) -> None:
        attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": {}}})

        assert "langfuse.trace.tags" not in attributes

    def test_empty_string_role_is_omitted_from_output(self) -> None:
        response_obj: Final = {"id": "chatcmpl-norole", "choices": [{"message": {"role": "", "content": "hi"}}]}
        attributes: Final = _emitted({"call_type": "acompletion", "litellm_params": {"metadata": {}}}, response_obj)

        assert json.loads(str(attributes["langfuse.observation.output"])) == {"content": "hi"}

    def test_no_messages_leaves_trace_input_unset_but_emits_output(self) -> None:
        attributes: Final = _emitted(
            {"call_type": "acompletion", "litellm_params": {"metadata": {}}}, _chat_response("hi")
        )

        assert "langfuse.trace.input" not in attributes
        assert "langfuse.observation.input" not in attributes
        assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"]


def _proxy_span(parent: Context | None = None) -> Span:
    return _TRACER.start_span(LITELLM_PROXY_REQUEST_SPAN_NAME, context=parent)


def _child_of(parent: Span) -> Span:
    return _TRACER.start_span("litellm_request", context=trace.set_span_in_context(parent))


def _carrying(parent: Span, key: str = "metadata") -> Mapping[str, object]:
    return {**CHAT_KWARGS, "litellm_params": {key: {"litellm_parent_otel_span": parent}}}


class TestDerivedTraceFieldsOnlyOnTracesLitellmStarts:
    def test_a_span_under_an_inbound_traceparent_joins_the_caller_trace(self) -> None:
        span: Final = _TRACER.start_span(
            "litellm_request", context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER))
        )

        attributes: Final = _stamped(span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok"))

        assert _trace_fields(attributes) == {}
        assert "langfuse.observation.input" in attributes and "langfuse.observation.output" in attributes

    def test_the_active_app_span_litellm_writes_onto_keeps_its_trace_fields(self) -> None:
        with _TRACER.start_as_current_span("checkout") as app_span:
            app_span.set_attribute("langfuse.trace.name", "checkout")

            attributes: Final = _stamped(
                app_span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok")
            )

        assert _trace_fields(attributes) == {"langfuse.trace.name": "checkout"}

    def test_a_litellm_root_span_derives_while_an_app_span_is_current(self) -> None:
        with _TRACER.start_as_current_span("checkout"):
            span: Final = _TRACER.start_span("litellm_request", context=Context())

            attributes: Final = _stamped(
                span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok")
            )

        assert attributes["langfuse.trace.name"] == "litellm-acompletion"

    def test_a_span_under_a_caller_provided_parent_span_joins_the_caller_trace(self) -> None:
        caller_span: Final = _TRACER.start_span("caller-root")

        attributes: Final = _stamped(_child_of(caller_span), _carrying(caller_span), _chat_response("ok"))

        assert _trace_fields(attributes) == {}

    @pytest.mark.parametrize("key", ["metadata", "litellm_metadata"])
    def test_a_span_under_litellms_own_unparented_proxy_span_derives(self, key: str) -> None:
        proxy_span: Final = _proxy_span()

        attributes: Final = _stamped(_child_of(proxy_span), _carrying(proxy_span, key), _chat_response("ok"))

        assert attributes["langfuse.trace.name"] == "litellm-acompletion"
        assert attributes["langfuse.trace.input"] == attributes["langfuse.observation.input"]
        assert attributes["langfuse.trace.output"] == attributes["langfuse.observation.output"]

    def test_a_span_under_a_proxy_span_with_an_inbound_traceparent_joins_the_caller_trace(self) -> None:
        proxy_span: Final = _proxy_span(trace.set_span_in_context(trace.NonRecordingSpan(CALLER)))

        attributes: Final = _stamped(_child_of(proxy_span), _carrying(proxy_span), _chat_response("ok"))

        assert _trace_fields(attributes) == {}

    def test_a_span_whose_parent_is_not_the_carried_proxy_span_joins_that_parent_trace(self) -> None:
        other_parent: Final = _TRACER.start_span("caller-root")

        attributes: Final = _stamped(_child_of(other_parent), _carrying(_proxy_span()), _chat_response("ok"))

        assert _trace_fields(attributes) == {}

    def test_an_inbound_traceparent_still_carries_the_caller_trace_name_and_tags(self) -> None:
        span: Final = _TRACER.start_span(
            "litellm_request", context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER))
        )

        attributes: Final = _stamped(
            span,
            {**CHAT_KWARGS, "litellm_params": {"metadata": {"trace_name": "caller-trace", "tags": ["caller"]}}},
            _chat_response("ok"),
        )

        assert _trace_fields(attributes) == {
            "langfuse.trace.name": "caller-trace",
            "langfuse.trace.tags": json.dumps(["caller"]),
        }


ERROR_KEYS: Final = (
    "error.code",
    "http.response.status_code",
    "error.type",
    "error.message",
    "error.llm_provider",
    "error.stack_trace",
)
SPAN_ATTRIBUTE_LIMIT: Final = 128
_LIMITED_TRACER: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=SPAN_ATTRIBUTE_LIMIT)).get_tracer(
    "langfuse-otel-limit-tests"
)


@dataclass(frozen=True, slots=True)
class ExportedSpan:
    keys: frozenset[str]
    dropped: int


def _exported_full_span(filler: int, *, joins_caller: bool, failed: bool) -> ExportedSpan:
    span: Final = _LIMITED_TRACER.start_span(
        "litellm_request",
        context=trace.set_span_in_context(trace.NonRecordingSpan(CALLER)) if joins_caller else Context(),
    )
    span.set_attributes({f"main.{index}": index for index in range(filler)})
    LangfuseOtelLogger._set_langfuse_specific_attributes(
        span, {**CHAT_KWARGS, "litellm_params": {"metadata": {}}}, _chat_response("ok")
    )
    span.set_attributes(dict.fromkeys(ERROR_KEYS, "scripted failure") if failed else {})
    span.end()
    return ExportedSpan(frozenset(getattr(span, "attributes", None) or {}), getattr(span, "dropped_attributes", 0))


class TestDerivedTraceFieldsAtTheSpanAttributeLimit:
    @pytest.mark.parametrize("failed", [False, True])
    @pytest.mark.parametrize("filler", range(SPAN_ATTRIBUTE_LIMIT - 30, SPAN_ATTRIBUTE_LIMIT + 1))
    def test_trace_fields_never_evict_or_drop_another_attribute(self, filler: int, failed: bool) -> None:
        derived: Final = _exported_full_span(filler, joins_caller=False, failed=failed)
        without_trace_fields: Final = _exported_full_span(filler, joins_caller=True, failed=failed)

        assert without_trace_fields.keys <= derived.keys
        assert derived.dropped == without_trace_fields.dropped

    def test_a_span_with_room_to_spare_carries_every_trace_field(self) -> None:
        derived: Final = _exported_full_span(SPAN_ATTRIBUTE_LIMIT - 30, joins_caller=False, failed=False)

        assert {"langfuse.trace.name", "langfuse.trace.input", "langfuse.trace.output"} <= derived.keys


LANGFUSE_ENV_ONLY: Final = {
    key: value
    for key, value in os.environ.items()
    if not key.startswith(("LANGFUSE_", "OTEL_"))
}


def _decoded_basic_auth(header: str) -> str:
    assert header.startswith("Basic ")
    return base64.b64decode(header.removeprefix("Basic ")).decode()


def test_langfuse_otel_env_config_headers_carry_v4_ingestion_and_basic_auth() -> None:
    with patch.dict(
        os.environ, {**LANGFUSE_ENV_ONLY, "LANGFUSE_PUBLIC_KEY": "pk-lf-123", "LANGFUSE_SECRET_KEY": "sk-lf-123"}, clear=True
    ):
        logger: Final = LangfuseOtelLogger()
    headers: Final = OpenTelemetry._get_headers_dictionary(logger.config.headers)
    assert headers["x-langfuse-ingestion-version"] == "4"
    assert _decoded_basic_auth(headers["Authorization"]) == "pk-lf-123:sk-lf-123"


def test_langfuse_otel_dynamic_headers_carry_v4_ingestion_and_basic_auth() -> None:
    with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True):
        logger: Final = LangfuseOtelLogger()
    headers: Final = logger.construct_dynamic_otel_headers(
        StandardCallbackDynamicParams(langfuse_public_key="pk-lf-dynamic", langfuse_secret_key="sk-lf-dynamic")
    )
    assert headers is not None
    assert headers["x-langfuse-ingestion-version"] == "4"
    assert _decoded_basic_auth(headers["Authorization"]) == "pk-lf-dynamic:sk-lf-dynamic"


def test_langfuse_otel_does_not_start_proxy_request_span() -> None:
    langfuse_provider: Final = TracerProvider()
    generic_provider: Final = TracerProvider()
    with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True):
        langfuse_logger: Final = LangfuseOtelLogger(tracer_provider=langfuse_provider)
        generic_logger: Final = OpenTelemetry(
            config=OpenTelemetryConfig(exporter="console", skip_set_global=True), tracer_provider=generic_provider
        )
    started_at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc)
    request_headers: Final = {"Authorization": "Bearer test"}
    try:
        assert (
            langfuse_logger.create_litellm_proxy_request_started_span(start_time=started_at, headers=request_headers)
            is None
        )
        assert (
            generic_logger.create_litellm_proxy_request_started_span(start_time=started_at, headers=request_headers)
            is not None
        )
    finally:
        langfuse_provider.shutdown()
        generic_provider.shutdown()


_HOOK_CALL: Final[Mapping[str, object]] = MappingProxyType({})


def _spans_the_legacy_logger_exports(
    destinations: tuple[OtelDestination, ...], call: Mapping[str, object] = _HOOK_CALL
) -> tuple[str, ...]:
    exporter: Final = InMemorySpanExporter()
    provider: Final = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True):
        logger: Final = LangfuseOtelLogger(tracer_provider=provider)
    at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc)
    kwargs: Final[dict[str, object]] = {**CHAT_KWARGS, **call, "litellm_params": {"metadata": {}}}
    server_span: Final = provider.get_tracer("proxy").start_span("POST /v1/chat/completions")

    async def log() -> None:
        set_request_destinations(destinations)
        logger.log_success_event(dict(kwargs), _chat_response("ok"), at, at)  # pyright: ignore[reportUnknownMemberType]  # the v1 logger hooks are untyped
        await logger.async_log_failure_event({**kwargs, "exception": ValueError("boom")}, None, at, at)  # pyright: ignore[reportUnknownMemberType]  # the v1 logger hooks are untyped
        await logger.async_post_call_failure_hook({}, ValueError("boom"), UserAPIKeyAuth(parent_otel_span=server_span))

    try:
        contextvars.copy_context().run(asyncio.run, log())
        return tuple(span.name for span in exporter.get_finished_spans())
    finally:
        provider.shutdown()


_LANGFUSE_DESTINATION: Final = OtelDestination(
    endpoint="http://tenant.local/api/public/otel",
    headers={"Authorization": "Basic dGVuYW50"},
    callback_name="langfuse_otel",
)


_PROVIDER_CALL: Final[Mapping[str, object]] = MappingProxyType(
    {"standard_logging_object": {"id": "c1"}, "api_call_start_time": datetime(2026, 1, 1, tzinfo=timezone.utc)}
)


@pytest.mark.parametrize("call", [_HOOK_CALL, _PROVIDER_CALL], ids=["hook", "provider-call"])
def test_a_request_with_a_langfuse_destination_exports_nothing_through_the_legacy_logger(
    call: Mapping[str, object],
) -> None:
    assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,), call) == ()


def test_a_call_rejected_before_any_provider_still_exports_through_the_legacy_logger() -> None:
    rejected: Final = {"standard_logging_object": {"id": "c1"}, LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL: True}

    assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION,), rejected) == (
        "litellm_request",
        "raw_gen_ai_request",
        "litellm_request",
    )


def test_a_request_without_a_langfuse_destination_still_exports_through_the_legacy_logger() -> None:
    logged: Final = (
        "litellm_request",
        "raw_gen_ai_request",
        "litellm_request",
        "Failed Proxy Server Request",
        "POST /v1/chat/completions",
    )

    assert _spans_the_legacy_logger_exports(()) == logged
    assert _spans_the_legacy_logger_exports((_LANGFUSE_DESTINATION.model_copy(update={"callback_name": "arize"}),)) == (
        logged
    )