test_tracer.py

import json
import os
from datetime import date, datetime, timezone
from unittest import mock

import pytest
from opentelemetry.trace import (
    SpanKind,
    StatusCode,  # type: ignore
)

from strands.telemetry.tracer import JSONEncoder, Tracer, get_tracer, serialize
from strands.types.content import ContentBlock
from strands.types.interrupt import InterruptResponseContent
from strands.types.streaming import Metrics, StopReason, Usage


@pytest.fixture(autouse=True)
def moto_autouse(moto_env, moto_mock_aws):
    _ = moto_env
    _ = moto_mock_aws


@pytest.fixture
def mock_get_tracer_provider():
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer_provider") as mock_get_tracer_provider:
        mock_tracer = mock.MagicMock()
        mock_get_tracer_provider.get_tracer.return_value = mock_tracer
        yield mock_get_tracer_provider


@pytest.fixture
def mock_tracer():
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer") as mock_get_tracer:
        mock_tracer = mock.MagicMock()
        mock_get_tracer.return_value = mock_tracer
        yield mock_tracer


@pytest.fixture
def mock_span():
    mock_span = mock.MagicMock()
    return mock_span


@pytest.fixture
def clean_env():
    """Fixture to provide a clean environment for each test."""
    with mock.patch.dict(os.environ, {}, clear=True):
        yield


def test_init_default():
    """Test initializing the Tracer with default parameters."""
    tracer = Tracer()

    assert tracer.service_name == "strands.telemetry.tracer"
    assert tracer.tracer_provider is not None
    assert tracer.tracer is not None


def test_start_span_no_tracer():
    """Test starting a span when no tracer is configured."""
    tracer = Tracer()
    span = tracer._start_span("test_span")

    assert span is not None


def test_start_span(mock_tracer):
    """Test starting a span with attributes."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        span = tracer._start_span("test_span", attributes={"key": "value"})

        mock_tracer.start_span.assert_called_once_with(name="test_span", context=None, kind=SpanKind.INTERNAL)
        # Check that set_attributes was called with the provided attributes
        mock_span.set_attributes.assert_called_once_with({"key": "value"})
        assert span is not None


def test_end_span_no_span():
    """Test ending a span when span is None."""
    tracer = Tracer()
    # Should not raise an exception
    tracer._end_span(None)


def test_end_span(mock_span):
    """Test ending a span with attributes and no error."""
    tracer = Tracer()
    attributes = {"key": "value"}

    tracer._end_span(mock_span, attributes)

    # Check that set_attributes was called with the provided attributes
    mock_span.set_attributes.assert_called_once_with({"key": "value"})
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_span_with_error(mock_span):
    """Test ending a span with an error."""
    tracer = Tracer()
    error = Exception("Test error")

    tracer._end_span(mock_span, error=error)

    mock_span.set_status.assert_called_once_with(StatusCode.ERROR, str(error))
    mock_span.record_exception.assert_called_once_with(error)
    mock_span.end.assert_called_once()


def test_end_span_with_error_message(mock_span):
    """Test ending a span with an error message."""
    tracer = Tracer()
    error_message = "Test error message"

    tracer.end_span_with_error(mock_span, error_message)

    mock_span.set_status.assert_called_once()
    assert mock_span.set_status.call_args[0][0] == StatusCode.ERROR
    mock_span.end.assert_called_once()


def test_end_span_with_empty_exception_message_uses_exception_name(mock_span):
    """Test that empty exception messages fall back to the exception type name."""
    tracer = Tracer()
    error = Exception()

    tracer.end_span_with_error(mock_span, "", error)

    mock_span.set_status.assert_called_once_with(StatusCode.ERROR, "Exception")
    mock_span.record_exception.assert_called_once_with(error)
    mock_span.end.assert_called_once()


def test_end_span_with_empty_base_exception_message_uses_exception_name(mock_span):
    """Test that empty BaseException messages fall back to the exception type name."""
    tracer = Tracer()
    error = KeyboardInterrupt()

    tracer.end_span_with_error(mock_span, "", error)

    mock_span.set_status.assert_called_once_with(StatusCode.ERROR, "KeyboardInterrupt")
    mock_span.record_exception.assert_called_once_with(error)
    mock_span.end.assert_called_once()


def test_end_span_with_error_prefers_explicit_message(mock_span):
    """Test that an explicit error message takes precedence over the exception text."""
    tracer = Tracer()
    error = Exception()

    tracer.end_span_with_error(mock_span, "Explicit error message", error)

    mock_span.set_status.assert_called_once_with(StatusCode.ERROR, "Explicit error message")
    mock_span.record_exception.assert_called_once_with(error)
    mock_span.end.assert_called_once()


def test_start_model_invoke_span(mock_tracer):
    """Test starting a model invoke span."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        messages = [{"role": "user", "content": [{"text": "Hello"}]}]
        model_id = "test-model"
        custom_attrs = {"custom_key": "custom_value", "user_id": "12345"}
        system_prompt = "You are a helpful assistant"

        span = tracer.start_model_invoke_span(
            messages=messages,
            agent_name="TestAgent",
            model_id=model_id,
            custom_trace_attributes=custom_attrs,
            system_prompt=system_prompt,
        )

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "chat"
        assert mock_tracer.start_span.call_args[1]["kind"] == SpanKind.INTERNAL
        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "chat",
                "gen_ai.system": "strands-agents",
                "custom_key": "custom_value",
                "user_id": "12345",
                "gen_ai.request.model": model_id,
                "agent_name": "TestAgent",
            }
        )

        calls = mock_span.add_event.call_args_list
        assert len(calls) == 2
        assert calls[0] == mock.call(
            "gen_ai.system.message",
            attributes={"content": serialize([{"text": system_prompt}])},
        )
        assert calls[1] == mock.call("gen_ai.user.message", attributes={"content": json.dumps(messages[0]["content"])})
        assert span is not None


def test_start_model_invoke_span_latest_conventions(mock_tracer, monkeypatch):
    """Test starting a model invoke span with the latest semantic conventions."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        messages = [
            {"role": "user", "content": [{"text": "Hello 2025-1993"}]},
            {
                "role": "assistant",
                "content": [
                    {"toolUse": {"input": '"expression": "2025-1993"', "name": "calculator", "toolUseId": "123"}}
                ],
            },
        ]
        model_id = "test-model"
        system_prompt = "You are a calculator assistant"

        span = tracer.start_model_invoke_span(
            messages=messages, agent_name="TestAgent", model_id=model_id, system_prompt=system_prompt
        )

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "chat"
        assert mock_tracer.start_span.call_args[1]["kind"] == SpanKind.INTERNAL

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "chat",
                "gen_ai.provider.name": "strands-agents",
                "gen_ai.request.model": model_id,
                "agent_name": "TestAgent",
            }
        )

        calls = mock_span.add_event.call_args_list
        assert len(calls) == 2
        assert calls[0] == mock.call(
            "gen_ai.client.inference.operation.details",
            attributes={
                "gen_ai.system_instructions": serialize([{"type": "text", "content": system_prompt}]),
            },
        )
        assert calls[1] == mock.call(
            "gen_ai.client.inference.operation.details",
            attributes={
                "gen_ai.input.messages": serialize(
                    [
                        {
                            "role": messages[0]["role"],
                            "parts": [{"type": "text", "content": "Hello 2025-1993"}],
                        },
                        {
                            "role": messages[1]["role"],
                            "parts": [
                                {
                                    "type": "tool_call",
                                    "name": "calculator",
                                    "id": "123",
                                    "arguments": '"expression": "2025-1993"',
                                }
                            ],
                        },
                    ]
                )
            },
        )
        assert span is not None


def test_start_model_invoke_span_without_system_prompt(mock_tracer):
    """Test that no system prompt event is emitted when system_prompt is None."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        messages = [{"role": "user", "content": [{"text": "Hello"}]}]

        span = tracer.start_model_invoke_span(messages=messages, model_id="test-model")

        assert mock_span.add_event.call_count == 1
        mock_span.add_event.assert_called_once_with(
            "gen_ai.user.message", attributes={"content": json.dumps(messages[0]["content"])}
        )
        assert span is not None


def test_start_model_invoke_span_with_system_prompt_content(mock_tracer):
    """Test that system_prompt_content takes priority over system_prompt string."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        messages = [{"role": "user", "content": [{"text": "Hello"}]}]
        system_prompt_content = [{"text": "You are helpful"}, {"text": "Be concise"}]

        span = tracer.start_model_invoke_span(
            messages=messages,
            model_id="test-model",
            system_prompt="ignored string",
            system_prompt_content=system_prompt_content,
        )

        calls = mock_span.add_event.call_args_list
        assert len(calls) == 2
        assert calls[0] == mock.call(
            "gen_ai.system.message",
            attributes={"content": serialize(system_prompt_content)},
        )
        assert span is not None


def test_end_model_invoke_span(mock_span):
    """Test ending a model invoke span."""
    tracer = Tracer()
    message = {"role": "assistant", "content": [{"text": "Response"}]}
    usage = Usage(inputTokens=10, outputTokens=20, totalTokens=30)
    metrics = Metrics(latencyMs=20, timeToFirstByteMs=10)
    stop_reason: StopReason = "end_turn"

    tracer.end_model_invoke_span(mock_span, message, usage, metrics, stop_reason)

    mock_span.set_attributes.assert_called_once_with(
        {
            "gen_ai.usage.prompt_tokens": 10,
            "gen_ai.usage.input_tokens": 10,
            "gen_ai.usage.completion_tokens": 20,
            "gen_ai.usage.output_tokens": 20,
            "gen_ai.usage.total_tokens": 30,
            "gen_ai.server.time_to_first_token": 10,
            "gen_ai.server.request.duration": 20,
        }
    )
    mock_span.add_event.assert_called_with(
        "gen_ai.choice",
        attributes={"message": json.dumps(message["content"]), "finish_reason": "end_turn"},
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_model_invoke_span_latest_conventions(mock_span, monkeypatch):
    """Test ending a model invoke span with the latest semantic conventions."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
        tracer = Tracer()
        message = {"role": "assistant", "content": [{"text": "Response"}]}
        usage = Usage(inputTokens=10, outputTokens=20, totalTokens=30)
        metrics = Metrics(latencyMs=20, timeToFirstByteMs=10)
        stop_reason: StopReason = "end_turn"

        tracer.end_model_invoke_span(mock_span, message, usage, metrics, stop_reason)

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.usage.prompt_tokens": 10,
                "gen_ai.usage.input_tokens": 10,
                "gen_ai.usage.completion_tokens": 20,
                "gen_ai.usage.output_tokens": 20,
                "gen_ai.usage.total_tokens": 30,
                "gen_ai.server.time_to_first_token": 10,
                "gen_ai.server.request.duration": 20,
            }
        )
        mock_span.add_event.assert_called_with(
            "gen_ai.client.inference.operation.details",
            attributes={
                "gen_ai.output.messages": serialize(
                    [
                        {
                            "role": "assistant",
                            "parts": [{"type": "text", "content": "Response"}],
                            "finish_reason": "end_turn",
                        }
                    ]
                ),
            },
        )
        mock_span.set_status.assert_called_once_with(StatusCode.OK)
        mock_span.end.assert_called_once()


def test_start_tool_call_span(mock_tracer):
    """Test starting a tool call span."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        tool = {"name": "test-tool", "toolUseId": "123", "input": {"param": "value"}}
        custom_attrs = {"session_id": "abc123", "environment": "production"}

        span = tracer.start_tool_call_span(tool, custom_trace_attributes=custom_attrs)

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "execute_tool test-tool"

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.tool.name": "test-tool",
                "gen_ai.system": "strands-agents",
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.tool.call.id": "123",
                "session_id": "abc123",
                "environment": "production",
            }
        )
        mock_span.add_event.assert_any_call(
            "gen_ai.tool.message", attributes={"role": "tool", "content": json.dumps({"param": "value"}), "id": "123"}
        )
        assert span is not None


def test_start_tool_call_span_latest_conventions(mock_tracer, monkeypatch):
    """Test starting a tool call span with the latest semantic conventions."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        tool = {"name": "test-tool", "toolUseId": "123", "input": {"param": "value"}}

        span = tracer.start_tool_call_span(tool)

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "execute_tool test-tool"

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.tool.name": "test-tool",
                "gen_ai.provider.name": "strands-agents",
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.tool.call.id": "123",
            }
        )
        mock_span.add_event.assert_called_with(
            "gen_ai.client.inference.operation.details",
            attributes={
                "gen_ai.input.messages": serialize(
                    [
                        {
                            "role": "tool",
                            "parts": [
                                {
                                    "type": "tool_call",
                                    "name": tool["name"],
                                    "id": tool["toolUseId"],
                                    "arguments": tool["input"],
                                }
                            ],
                        }
                    ]
                )
            },
        )
        assert span is not None


def test_start_swarm_call_span_with_string_task(mock_tracer):
    """Test starting a swarm call span with task as string."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        task = "Design foo bar"
        custom_attrs = {"workflow_id": "wf-789", "priority": "high"}

        span = tracer.start_multiagent_span(task, "swarm", custom_trace_attributes=custom_attrs)

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "invoke_swarm"

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "invoke_swarm",
                "gen_ai.system": "strands-agents",
                "gen_ai.agent.name": "swarm",
                "workflow_id": "wf-789",
                "priority": "high",
            }
        )
        mock_span.add_event.assert_any_call("gen_ai.user.message", attributes={"content": "Design foo bar"})
        assert span is not None


def test_start_swarm_span_with_contentblock_task(mock_tracer):
    """Test starting a swarm call span with task as list of contentBlock."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        task = [ContentBlock(text="Original Task: foo bar")]

        span = tracer.start_multiagent_span(task, "swarm")

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "invoke_swarm"

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "invoke_swarm",
                "gen_ai.system": "strands-agents",
                "gen_ai.agent.name": "swarm",
            }
        )
        mock_span.add_event.assert_any_call(
            "gen_ai.user.message", attributes={"content": '[{"text": "Original Task: foo bar"}]'}
        )
        assert span is not None


@pytest.mark.parametrize(
    "task, expected_parts",
    [
        ([ContentBlock(text="Test message")], [{"type": "text", "content": "Test message"}]),
        (
            [InterruptResponseContent(interruptResponse={"interruptId": "test-id", "response": "approved"})],
            [{"type": "interrupt_response", "id": "test-id", "response": "approved"}],
        ),
    ],
)
def test_start_multiagent_span_task_part_conversion(mock_tracer, task, expected_parts, monkeypatch):
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")

    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        tracer.start_multiagent_span(task, "swarm")

        expected_content = json.dumps([{"role": "user", "parts": expected_parts}])
        mock_span.add_event.assert_any_call(
            "gen_ai.client.inference.operation.details", attributes={"gen_ai.input.messages": expected_content}
        )


def test_start_swarm_span_with_contentblock_task_latest_conventions(mock_tracer, monkeypatch):
    """Test starting a swarm call span with task as list of contentBlock with latest semantic conventions."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        task = [ContentBlock(text="Original Task: foo bar")]

        span = tracer.start_multiagent_span(task, "swarm")

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "invoke_swarm"

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "invoke_swarm",
                "gen_ai.provider.name": "strands-agents",
                "gen_ai.agent.name": "swarm",
            }
        )
        mock_span.add_event.assert_any_call(
            "gen_ai.client.inference.operation.details",
            attributes={
                "gen_ai.input.messages": serialize(
                    [{"role": "user", "parts": [{"type": "text", "content": "Original Task: foo bar"}]}]
                )
            },
        )
        assert span is not None


def test_end_swarm_span(mock_span):
    """Test ending a tool call span."""
    tracer = Tracer()
    swarm_final_reuslt = "foo bar bar"

    tracer.end_swarm_span(mock_span, swarm_final_reuslt)

    mock_span.add_event.assert_called_with(
        "gen_ai.choice",
        attributes={"message": "foo bar bar"},
    )


def test_end_swarm_span_latest_conventions(mock_span, monkeypatch):
    """Test ending a tool call span with latest semantic conventions."""
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
    tracer = Tracer()
    swarm_final_reuslt = "foo bar bar"

    tracer.end_swarm_span(mock_span, swarm_final_reuslt)

    mock_span.add_event.assert_called_with(
        "gen_ai.client.inference.operation.details",
        attributes={
            "gen_ai.output.messages": serialize(
                [
                    {
                        "role": "assistant",
                        "parts": [{"type": "text", "content": "foo bar bar"}],
                    }
                ]
            )
        },
    )


def test_start_graph_call_span(mock_tracer):
    """Test starting a graph call span."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        tool = {"name": "test-tool", "toolUseId": "123", "input": {"param": "value"}}

        span = tracer.start_tool_call_span(tool)

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "execute_tool test-tool"

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.system": "strands-agents",
                "gen_ai.tool.name": "test-tool",
                "gen_ai.tool.call.id": "123",
            }
        )
        mock_span.add_event.assert_any_call(
            "gen_ai.tool.message", attributes={"role": "tool", "content": json.dumps({"param": "value"}), "id": "123"}
        )
        assert span is not None


def test_end_tool_call_span(mock_span):
    """Test ending a tool call span."""
    tracer = Tracer()
    tool_result = {"status": "success", "content": [{"text": "Tool result"}]}

    tracer.end_tool_call_span(mock_span, tool_result)

    mock_span.set_attributes.assert_called_once_with({"gen_ai.tool.status": "success"})
    mock_span.add_event.assert_called_with(
        "gen_ai.choice",
        attributes={"message": json.dumps(tool_result.get("content")), "id": ""},
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_tool_call_span_latest_conventions(mock_span, monkeypatch):
    """Test ending a tool call span with the latest semantic conventions."""
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
    tracer = Tracer()
    tool_result = {"status": "success", "content": [{"text": "Tool result"}, {"json": {"foo": "bar"}}]}

    tracer.end_tool_call_span(mock_span, tool_result)

    mock_span.set_attributes.assert_called_once_with({"gen_ai.tool.status": "success"})
    mock_span.add_event.assert_called_with(
        "gen_ai.client.inference.operation.details",
        attributes={
            "gen_ai.output.messages": serialize(
                [
                    {
                        "role": "tool",
                        "parts": [
                            {
                                "type": "tool_call_response",
                                "id": tool_result.get("toolUseId", ""),
                                "response": tool_result.get("content"),
                            }
                        ],
                    }
                ]
            )
        },
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_start_event_loop_cycle_span(mock_tracer):
    """Test starting an event loop cycle span."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        event_loop_kwargs = {"event_loop_cycle_id": "cycle-123"}
        messages = [{"role": "user", "content": [{"text": "Hello"}]}]
        custom_attrs = {"request_id": "req-456", "trace_level": "debug"}

        span = tracer.start_event_loop_cycle_span(
            event_loop_kwargs, messages=messages, custom_trace_attributes=custom_attrs
        )

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "execute_event_loop_cycle"

        mock_span.set_attributes.assert_called_once_with(
            {
                "event_loop.cycle_id": "cycle-123",
                "request_id": "req-456",
                "trace_level": "debug",
            }
        )
        mock_span.add_event.assert_any_call(
            "gen_ai.user.message", attributes={"content": json.dumps([{"text": "Hello"}])}
        )
        assert span is not None


def test_start_event_loop_cycle_span_latest_conventions(mock_tracer, monkeypatch):
    """Test starting an event loop cycle span with the latest semantic conventions."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        event_loop_kwargs = {"event_loop_cycle_id": "cycle-123"}
        messages = [{"role": "user", "content": [{"text": "Hello"}]}]

        span = tracer.start_event_loop_cycle_span(event_loop_kwargs, messages=messages)

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "execute_event_loop_cycle"

        mock_span.set_attributes.assert_called_once_with({"event_loop.cycle_id": "cycle-123"})
        mock_span.add_event.assert_any_call(
            "gen_ai.client.inference.operation.details",
            attributes={
                "gen_ai.input.messages": serialize([{"role": "user", "parts": [{"type": "text", "content": "Hello"}]}])
            },
        )
        assert span is not None


def test_end_event_loop_cycle_span(mock_span):
    """Test ending an event loop cycle span."""
    tracer = Tracer()
    message = {"role": "assistant", "content": [{"text": "Response"}]}
    tool_result_message = {
        "role": "assistant",
        "content": [
            {"toolResult": {"toolUseId": "123", "status": "success", "content": [{"text": "Weather is sunny"}]}}
        ],
    }

    tracer.end_event_loop_cycle_span(mock_span, message, tool_result_message)

    mock_span.add_event.assert_called_with(
        "gen_ai.choice",
        attributes={
            "message": json.dumps(message["content"]),
            "tool.result": json.dumps(tool_result_message["content"]),
        },
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_event_loop_cycle_span_latest_conventions(mock_span, monkeypatch):
    """Test ending an event loop cycle span with the latest semantic conventions."""
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
    tracer = Tracer()
    message = {"role": "assistant", "content": [{"text": "Response"}]}
    tool_result_message = {
        "role": "assistant",
        "content": [
            {"toolResult": {"toolUseId": "123", "status": "success", "content": [{"text": "Weather is sunny"}]}}
        ],
    }

    tracer.end_event_loop_cycle_span(mock_span, message, tool_result_message)

    mock_span.add_event.assert_called_with(
        "gen_ai.client.inference.operation.details",
        attributes={
            "gen_ai.output.messages": serialize(
                [
                    {
                        "role": "assistant",
                        "parts": [
                            {
                                "type": "tool_call_response",
                                "id": "123",
                                "response": [{"text": "Weather is sunny"}],
                            }
                        ],
                    }
                ]
            )
        },
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_start_agent_span(mock_tracer):
    """Test starting an agent span."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        content = [{"text": "test prompt"}]
        model_id = "test-model"
        tools = [{"name": "weather_tool"}]
        custom_attrs = {"custom_attr": "value"}

        span = tracer.start_agent_span(
            custom_trace_attributes=custom_attrs,
            agent_name="WeatherAgent",
            messages=[{"content": content, "role": "user"}],
            model_id=model_id,
            tools=tools,
        )

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "invoke_agent WeatherAgent"
        assert mock_tracer.start_span.call_args[1]["kind"] == SpanKind.INTERNAL

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "invoke_agent",
                "gen_ai.system": "strands-agents",
                "gen_ai.agent.name": "WeatherAgent",
                "gen_ai.request.model": model_id,
                "gen_ai.agent.tools": json.dumps(tools),
                "custom_attr": "value",
            }
        )
        mock_span.add_event.assert_any_call("gen_ai.user.message", attributes={"content": json.dumps(content)})
        assert span is not None


def test_start_agent_span_latest_conventions(mock_tracer, monkeypatch):
    """Test starting an agent span with the latest semantic conventions."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        content = [{"text": "test prompt"}]
        model_id = "test-model"
        tools = [{"name": "weather_tool"}]
        custom_attrs = {"custom_attr": "value"}

        span = tracer.start_agent_span(
            custom_trace_attributes=custom_attrs,
            agent_name="WeatherAgent",
            messages=[{"content": content, "role": "user"}],
            model_id=model_id,
            tools=tools,
        )

        mock_tracer.start_span.assert_called_once()
        assert mock_tracer.start_span.call_args[1]["name"] == "invoke_agent WeatherAgent"

        mock_span.set_attributes.assert_called_once_with(
            {
                "gen_ai.operation.name": "invoke_agent",
                "gen_ai.provider.name": "strands-agents",
                "gen_ai.agent.name": "WeatherAgent",
                "gen_ai.request.model": model_id,
                "gen_ai.agent.tools": json.dumps(tools),
                "custom_attr": "value",
            }
        )
        mock_span.add_event.assert_any_call(
            "gen_ai.client.inference.operation.details",
            attributes={
                "gen_ai.input.messages": serialize(
                    [{"role": "user", "parts": [{"type": "text", "content": "test prompt"}]}]
                )
            },
        )
        assert span is not None


def test_end_agent_span(mock_span):
    """Test ending an agent span."""
    tracer = Tracer()

    # Mock AgentResult with metrics
    mock_metrics = mock.MagicMock()
    mock_metrics.accumulated_usage = {"inputTokens": 50, "outputTokens": 100, "totalTokens": 150}

    mock_response = mock.MagicMock()
    mock_response.metrics = mock_metrics
    mock_response.stop_reason = "end_turn"
    mock_response.__str__ = mock.MagicMock(return_value="Agent response")

    tracer.end_agent_span(mock_span, mock_response)

    mock_span.set_attributes.assert_called_once_with(
        {
            "gen_ai.usage.prompt_tokens": 50,
            "gen_ai.usage.input_tokens": 50,
            "gen_ai.usage.completion_tokens": 100,
            "gen_ai.usage.output_tokens": 100,
            "gen_ai.usage.total_tokens": 150,
            "gen_ai.usage.cache_read_input_tokens": 0,
            "gen_ai.usage.cache_write_input_tokens": 0,
        }
    )
    mock_span.add_event.assert_any_call(
        "gen_ai.choice",
        attributes={"message": "Agent response", "finish_reason": "end_turn"},
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_agent_span_with_langfuse_observation_type(mock_span, monkeypatch):
    """Test ending an agent span with Langfuse observation type to prevent double counting the tokens."""
    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://us.cloud.langfuse.com")
    tracer = Tracer()

    # Mock AgentResult with metrics
    mock_metrics = mock.MagicMock()
    mock_metrics.accumulated_usage = {"inputTokens": 50, "outputTokens": 100, "totalTokens": 150}

    mock_response = mock.MagicMock()
    mock_response.metrics = mock_metrics
    mock_response.stop_reason = "end_turn"
    mock_response.__str__ = mock.MagicMock(return_value="Agent response")

    tracer.end_agent_span(mock_span, mock_response)

    mock_span.set_attributes.assert_called_once_with(
        {
            "langfuse.observation.type": "span",
            "gen_ai.usage.prompt_tokens": 50,
            "gen_ai.usage.input_tokens": 50,
            "gen_ai.usage.completion_tokens": 100,
            "gen_ai.usage.output_tokens": 100,
            "gen_ai.usage.total_tokens": 150,
            "gen_ai.usage.cache_read_input_tokens": 0,
            "gen_ai.usage.cache_write_input_tokens": 0,
        }
    )
    mock_span.add_event.assert_any_call(
        "gen_ai.choice",
        attributes={"message": "Agent response", "finish_reason": "end_turn"},
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_agent_span_latest_conventions(mock_span, monkeypatch):
    """Test ending an agent span with the latest semantic conventions."""
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
    tracer = Tracer()

    # Mock AgentResult with metrics
    mock_metrics = mock.MagicMock()
    mock_metrics.accumulated_usage = {"inputTokens": 50, "outputTokens": 100, "totalTokens": 150}

    mock_response = mock.MagicMock()
    mock_response.metrics = mock_metrics
    mock_response.stop_reason = "end_turn"
    mock_response.__str__ = mock.MagicMock(return_value="Agent response")

    tracer.end_agent_span(mock_span, mock_response)

    mock_span.set_attributes.assert_called_once_with(
        {
            "gen_ai.usage.prompt_tokens": 50,
            "gen_ai.usage.input_tokens": 50,
            "gen_ai.usage.completion_tokens": 100,
            "gen_ai.usage.output_tokens": 100,
            "gen_ai.usage.total_tokens": 150,
            "gen_ai.usage.cache_read_input_tokens": 0,
            "gen_ai.usage.cache_write_input_tokens": 0,
        }
    )
    mock_span.add_event.assert_called_with(
        "gen_ai.client.inference.operation.details",
        attributes={
            "gen_ai.output.messages": serialize(
                [
                    {
                        "role": "assistant",
                        "parts": [{"type": "text", "content": "Agent response"}],
                        "finish_reason": "end_turn",
                    }
                ]
            )
        },
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_model_invoke_span_with_cache_metrics(mock_span):
    """Test ending a model invoke span with cache metrics."""
    tracer = Tracer()
    message = {"role": "assistant", "content": [{"text": "Response"}]}
    usage = Usage(
        inputTokens=10,
        outputTokens=20,
        totalTokens=30,
        cacheReadInputTokens=5,
        cacheWriteInputTokens=3,
    )
    stop_reason: StopReason = "end_turn"
    metrics = Metrics(latencyMs=10, timeToFirstByteMs=5)

    tracer.end_model_invoke_span(mock_span, message, usage, metrics, stop_reason)

    mock_span.set_attributes.assert_called_once_with(
        {
            "gen_ai.usage.prompt_tokens": 10,
            "gen_ai.usage.input_tokens": 10,
            "gen_ai.usage.completion_tokens": 20,
            "gen_ai.usage.output_tokens": 20,
            "gen_ai.usage.total_tokens": 30,
            "gen_ai.usage.cache_read_input_tokens": 5,
            "gen_ai.usage.cache_write_input_tokens": 3,
            "gen_ai.server.request.duration": 10,
            "gen_ai.server.time_to_first_token": 5,
        }
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_end_agent_span_with_cache_metrics(mock_span):
    """Test ending an agent span with cache metrics."""
    tracer = Tracer()

    # Mock AgentResult with metrics including cache tokens
    mock_metrics = mock.MagicMock()
    mock_metrics.accumulated_usage = {
        "inputTokens": 50,
        "outputTokens": 100,
        "totalTokens": 150,
        "cacheReadInputTokens": 25,
        "cacheWriteInputTokens": 10,
    }

    mock_response = mock.MagicMock()
    mock_response.metrics = mock_metrics
    mock_response.stop_reason = "end_turn"
    mock_response.__str__ = mock.MagicMock(return_value="Agent response")

    tracer.end_agent_span(mock_span, mock_response)

    mock_span.set_attributes.assert_called_once_with(
        {
            "gen_ai.usage.prompt_tokens": 50,
            "gen_ai.usage.input_tokens": 50,
            "gen_ai.usage.completion_tokens": 100,
            "gen_ai.usage.output_tokens": 100,
            "gen_ai.usage.total_tokens": 150,
            "gen_ai.usage.cache_read_input_tokens": 25,
            "gen_ai.usage.cache_write_input_tokens": 10,
        }
    )
    mock_span.set_status.assert_called_once_with(StatusCode.OK)
    mock_span.end.assert_called_once()


def test_get_tracer_singleton():
    """Test that get_tracer returns a singleton instance."""
    # Reset the singleton first
    with mock.patch("strands.telemetry.tracer._tracer_instance", None):
        tracer1 = get_tracer()
        tracer2 = get_tracer()

        assert tracer1 is tracer2


def test_get_tracer_new_endpoint():
    """Test that get_tracer creates a new instance when endpoint changes."""
    # Reset the singleton first
    with mock.patch("strands.telemetry.tracer._tracer_instance", None):
        tracer1 = get_tracer()
        tracer2 = get_tracer()

        assert tracer1 is tracer2


def test_initialize_tracer_with_custom_tracer_provider(mock_get_tracer_provider):
    """Test initializing the tracer with NoOpTracerProvider."""
    tracer = Tracer()

    mock_get_tracer_provider.assert_called()

    assert tracer.tracer_provider is not None
    assert tracer.tracer is not None


def test_end_span_with_exception_handling(mock_span):
    """Test ending a span with exception handling."""
    tracer = Tracer()

    # Make set_attribute throw an exception
    mock_span.set_attribute.side_effect = Exception("Test error during set_attribute")

    try:
        # Should not raise an exception
        tracer._end_span(mock_span, {"key": "value"})

        # Should still try to end the span
        mock_span.end.assert_called_once()
    except Exception:
        pytest.fail("_end_span should not raise exceptions")


def test_force_flush_with_error(mock_span, mock_get_tracer_provider):
    """Test force flush with error handling."""
    # Setup the tracer with a provider that raises an exception on force_flush
    tracer = Tracer()

    mock_tracer_provider = mock_get_tracer_provider.return_value
    mock_tracer_provider.force_flush.side_effect = Exception("Force flush error")

    # Should not raise an exception
    tracer._end_span(mock_span)

    # Verify force_flush was called
    mock_tracer_provider.force_flush.assert_called_once()


def test_end_agent_span_with_empty_error_message_uses_exception_name(mock_span):
    """Test that agent spans fall back to the exception type name for empty errors."""
    tracer = Tracer()
    error = Exception()

    tracer.end_agent_span(mock_span, error=error)

    mock_span.set_status.assert_called_once_with(StatusCode.ERROR, "Exception")
    mock_span.record_exception.assert_called_once_with(error)
    mock_span.end.assert_called_once()


def test_end_tool_call_span_with_empty_error_message_uses_exception_name(mock_span):
    """Test that tool call spans fall back to the exception type name for empty errors."""
    tracer = Tracer()
    error = Exception()

    tracer.end_tool_call_span(mock_span, None, error=error)

    mock_span.set_status.assert_called_once_with(StatusCode.ERROR, "Exception")
    mock_span.record_exception.assert_called_once_with(error)
    mock_span.end.assert_called_once()


def test_end_tool_call_span_with_none(mock_span):
    """Test ending a tool call span with None result."""
    tracer = Tracer()

    # Should not raise an exception
    tracer.end_tool_call_span(mock_span, None)

    # Should still end the span
    mock_span.end.assert_called_once()


def test_start_model_invoke_span_with_parent(mock_tracer):
    """Test starting a model invoke span with a parent span."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        parent_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        span = tracer.start_model_invoke_span(
            messages=[], parent_span=parent_span, agent_name="TestAgent", model_id="test-model"
        )

        # Verify trace.set_span_in_context was called with parent span
        mock_tracer.start_span.assert_called_once()

        # Verify span was returned
        assert span is mock_span


@pytest.mark.parametrize(
    "input_data, expected_result",
    [
        ("test string", '"test string"'),
        (1234, "1234"),
        (13.37, "13.37"),
        (False, "false"),
        (None, "null"),
    ],
)
def test_json_encoder_serializable(input_data, expected_result):
    """Test encoding of serializable values."""
    encoder = JSONEncoder()

    result = encoder.encode(input_data)
    assert result == expected_result


def test_json_encoder_datetime():
    """Test encoding datetime and date objects."""
    encoder = JSONEncoder()

    dt = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
    result = encoder.encode(dt)
    assert result == f'"{dt.isoformat()}"'

    d = date(2025, 1, 1)
    result = encoder.encode(d)
    assert result == f'"{d.isoformat()}"'


def test_json_encoder_list():
    """Test encoding a list with mixed content."""
    encoder = JSONEncoder()

    non_serializable = lambda x: x  # noqa: E731

    data = ["value", 42, 13.37, non_serializable, None, {"key": True}, ["value here"]]

    result = json.loads(encoder.encode(data))
    assert result == ["value", 42, 13.37, "<replaced>", None, {"key": True}, ["value here"]]


def test_json_encoder_dict():
    """Test encoding a dict with mixed content."""
    encoder = JSONEncoder()

    class UnserializableClass:
        def __str__(self):
            return "Unserializable Object"

    non_serializable = lambda x: x  # noqa: E731

    now = datetime.now(timezone.utc)

    data = {
        "metadata": {
            "timestamp": now,
            "version": "1.0",
            "debug_info": {"object": non_serializable, "callable": lambda x: x + 1},  # noqa: E731
        },
        "content": [
            {"type": "text", "value": "Hello world"},
            {"type": "binary", "value": non_serializable},
            {"type": "mixed", "values": [1, "text", non_serializable, {"nested": non_serializable}]},
        ],
        "statistics": {
            "processed": 100,
            "failed": 5,
            "details": [{"id": 1, "status": "ok"}, {"id": 2, "status": "error", "error_obj": non_serializable}],
        },
        "list": [
            non_serializable,
            1234,
            13.37,
            True,
            None,
            "string here",
        ],
    }

    expected = {
        "metadata": {
            "timestamp": now.isoformat(),
            "version": "1.0",
            "debug_info": {"object": "<replaced>", "callable": "<replaced>"},
        },
        "content": [
            {"type": "text", "value": "Hello world"},
            {"type": "binary", "value": "<replaced>"},
            {"type": "mixed", "values": [1, "text", "<replaced>", {"nested": "<replaced>"}]},
        ],
        "statistics": {
            "processed": 100,
            "failed": 5,
            "details": [{"id": 1, "status": "ok"}, {"id": 2, "status": "error", "error_obj": "<replaced>"}],
        },
        "list": [
            "<replaced>",
            1234,
            13.37,
            True,
            None,
            "string here",
        ],
    }

    result = json.loads(encoder.encode(data))

    assert result == expected


def test_json_encoder_value_error():
    """Test encoding values that cause ValueError."""
    encoder = JSONEncoder()

    # A very large integer that exceeds JSON limits and throws ValueError
    huge_number = 2**100000

    # Test in a dictionary
    dict_data = {"normal": 42, "huge": huge_number}
    result = json.loads(encoder.encode(dict_data))
    assert result == {"normal": 42, "huge": "<replaced>"}

    # Test in a list
    list_data = [42, huge_number]
    result = json.loads(encoder.encode(list_data))
    assert result == [42, "<replaced>"]

    # Test just the value
    result = json.loads(encoder.encode(huge_number))
    assert result == "<replaced>"


def test_serialize_non_ascii_characters():
    """Test that non-ASCII characters are preserved in JSON serialization."""

    # Test with Japanese text
    japanese_text = "こんにちは世界"
    result = serialize({"text": japanese_text})
    assert japanese_text in result
    assert "\\u" not in result

    # Test with emoji
    emoji_text = "Hello 🌍"
    result = serialize({"text": emoji_text})
    assert emoji_text in result
    assert "\\u" not in result

    # Test with Chinese characters
    chinese_text = "你好，世界"
    result = serialize({"text": chinese_text})
    assert chinese_text in result
    assert "\\u" not in result

    # Test with mixed content
    mixed_text = {"ja": "こんにちは", "emoji": "😊", "zh": "你好", "en": "hello"}
    result = serialize(mixed_text)
    assert "こんにちは" in result
    assert "😊" in result
    assert "你好" in result
    assert "\\u" not in result


def test_serialize_vs_json_dumps():
    """Test that serialize behaves differently from default json.dumps for non-ASCII characters."""

    # Test with Japanese text
    japanese_text = "こんにちは世界"

    # Default json.dumps should escape non-ASCII characters
    default_result = json.dumps({"text": japanese_text})
    assert "\\u" in default_result

    # Our serialize function should preserve non-ASCII characters
    custom_result = serialize({"text": japanese_text})
    assert japanese_text in custom_result
    assert "\\u" not in custom_result


@pytest.mark.parametrize(
    "message, expected_event_name, description",
    [
        # Regular role-based messages
        (
            {"role": "user", "content": [{"text": "Hello"}]},
            "gen_ai.user.message",
            "regular user message",
        ),
        (
            {"role": "assistant", "content": [{"text": "Hello"}]},
            "gen_ai.assistant.message",
            "regular assistant message",
        ),
        (
            {"role": "system", "content": [{"text": "You are a helpful assistant"}]},
            "gen_ai.system.message",
            "regular system message",
        ),
        # Messages with tool results should always be labeled as tool messages
        (
            {
                "role": "user",
                "content": [
                    {
                        "toolResult": {
                            "toolUseId": "123",
                            "status": "success",
                            "content": [{"text": "Tool response"}],
                        }
                    }
                ],
            },
            "gen_ai.tool.message",
            "user message containing tool result",
        ),
        (
            {
                "role": "assistant",
                "content": [
                    {
                        "toolResult": {
                            "toolUseId": "123",
                            "status": "success",
                            "content": [{"text": "Tool response"}],
                        }
                    }
                ],
            },
            "gen_ai.tool.message",
            "assistant message containing tool result",
        ),
        # Mixed content with tool results
        (
            {
                "role": "user",
                "content": [
                    {"text": "Here are the results:"},
                    {
                        "toolResult": {
                            "toolUseId": "123",
                            "status": "success",
                            "content": [{"text": "Tool response"}],
                        }
                    },
                ],
            },
            "gen_ai.tool.message",
            "message with both text and tool result",
        ),
        # Multiple tool results
        (
            {
                "role": "user",
                "content": [
                    {
                        "toolResult": {
                            "toolUseId": "123",
                            "status": "success",
                            "content": [{"text": "First tool"}],
                        }
                    },
                    {
                        "toolResult": {
                            "toolUseId": "456",
                            "status": "success",
                            "content": [{"text": "Second tool"}],
                        }
                    },
                ],
            },
            "gen_ai.tool.message",
            "message with multiple tool results",
        ),
        # Edge cases
        (
            {"role": "user", "content": []},
            "gen_ai.user.message",
            "message with empty content",
        ),
        (
            {"role": "assistant"},
            "gen_ai.assistant.message",
            "message with no content key",
        ),
    ],
)
def test_get_event_name_for_message(message, expected_event_name, description):
    """Test getting event name for various message types using data-driven approach."""
    tracer = Tracer()

    event_name = tracer._get_event_name_for_message(message)

    assert event_name == expected_event_name, f"Failed for {description}"


def test_start_model_invoke_span_with_tool_result_message(mock_tracer):
    """Test that start_model_invoke_span correctly labels tool result messages."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        # Message that contains a tool result
        messages = [
            {
                "role": "user",
                "content": [
                    {"toolResult": {"toolUseId": "123", "status": "success", "content": [{"text": "Weather is sunny"}]}}
                ],
            }
        ]

        span = tracer.start_model_invoke_span(messages=messages, model_id="test-model")

        # Should use gen_ai.tool.message event name instead of gen_ai.user.message
        mock_span.add_event.assert_called_with(
            "gen_ai.tool.message", attributes={"content": json.dumps(messages[0]["content"])}
        )
        assert span is not None


def test_start_agent_span_with_tool_result_message(mock_tracer):
    """Test that start_agent_span correctly labels tool result messages."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        # Message that contains a tool result
        messages = [
            {
                "role": "user",
                "content": [
                    {"toolResult": {"toolUseId": "123", "status": "success", "content": [{"text": "Weather is sunny"}]}}
                ],
            }
        ]

        span = tracer.start_agent_span(messages=messages, agent_name="WeatherAgent", model_id="test-model")

        # Should use gen_ai.tool.message event name instead of gen_ai.user.message
        mock_span.add_event.assert_called_with(
            "gen_ai.tool.message", attributes={"content": json.dumps(messages[0]["content"])}
        )
        assert span is not None


def test_start_event_loop_cycle_span_with_tool_result_message(mock_tracer):
    """Test that start_event_loop_cycle_span correctly labels tool result messages."""
    with mock.patch("strands.telemetry.tracer.trace_api.get_tracer", return_value=mock_tracer):
        tracer = Tracer()
        tracer.tracer = mock_tracer

        mock_span = mock.MagicMock()
        mock_tracer.start_span.return_value = mock_span

        # Message that contains a tool result
        messages = [
            {
                "role": "user",
                "content": [
                    {"toolResult": {"toolUseId": "123", "status": "success", "content": [{"text": "Weather is sunny"}]}}
                ],
            }
        ]

        event_loop_kwargs = {"event_loop_cycle_id": "cycle-123"}
        span = tracer.start_event_loop_cycle_span(event_loop_kwargs, messages=messages)

        # Should use gen_ai.tool.message event name instead of gen_ai.user.message
        mock_span.add_event.assert_called_with(
            "gen_ai.tool.message", attributes={"content": json.dumps(messages[0]["content"])}
        )
        assert span is not None


def test_start_agent_span_does_not_include_tool_definitions_by_default():
    """Verify that start_agent_span does not include tool definitions by default."""
    tracer = Tracer()
    tracer._start_span = mock.MagicMock()

    tools_config = {
        "my_tool": {
            "name": "my_tool",
            "description": "A test tool",
            "inputSchema": {"json": {}},
            "outputSchema": {"json": {}},
        }
    }

    tracer.start_agent_span(messages=[], agent_name="TestAgent", tools_config=tools_config)

    tracer._start_span.assert_called_once()
    _, call_kwargs = tracer._start_span.call_args
    attributes = call_kwargs.get("attributes", {})
    assert "gen_ai.tool.definitions" not in attributes


def test_start_agent_span_includes_tool_definitions_when_enabled(monkeypatch):
    """Verify that start_agent_span includes tool definitions when enabled."""
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_tool_definitions")
    tracer = Tracer()
    tracer._start_span = mock.MagicMock()

    tools_config = {
        "my_tool": {
            "name": "my_tool",
            "description": "A test tool",
            "inputSchema": {"json": {"type": "object", "properties": {}}},
            "outputSchema": {"json": {"type": "object", "properties": {}}},
        }
    }

    tracer.start_agent_span(messages=[], agent_name="TestAgent", tools_config=tools_config)

    tracer._start_span.assert_called_once()
    _, call_kwargs = tracer._start_span.call_args
    attributes = call_kwargs.get("attributes", {})

    assert "gen_ai.tool.definitions" in attributes
    expected_tool_details = [
        {
            "name": "my_tool",
            "description": "A test tool",
            "inputSchema": {"json": {"type": "object", "properties": {}}},
            "outputSchema": {"json": {"type": "object", "properties": {}}},
        }
    ]
    expected_json = serialize(expected_tool_details)
    assert attributes["gen_ai.tool.definitions"] == expected_json


def test_end_model_invoke_span_langfuse_adds_attributes(mock_span, monkeypatch):
    """Test that end_model_invoke_span adds attributes via set_attributes for Langfuse."""
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://us.cloud.langfuse.com")

    tracer = Tracer()
    message = {"role": "assistant", "content": [{"text": "Response"}]}
    usage = Usage(inputTokens=10, outputTokens=20, totalTokens=30)
    metrics = Metrics(latencyMs=20, timeToFirstByteMs=10)
    stop_reason: StopReason = "end_turn"

    tracer.end_model_invoke_span(mock_span, message, usage, metrics, stop_reason)

    expected_output = serialize(
        [
            {
                "role": "assistant",
                "parts": [{"type": "text", "content": "Response"}],
                "finish_reason": "end_turn",
            }
        ]
    )

    assert mock_span.set_attributes.call_count == 2
    mock_span.set_attributes.assert_any_call({"gen_ai.output.messages": expected_output})
    mock_span.set_attributes.assert_any_call(
        {
            "gen_ai.usage.prompt_tokens": 10,
            "gen_ai.usage.input_tokens": 10,
            "gen_ai.usage.completion_tokens": 20,
            "gen_ai.usage.output_tokens": 20,
            "gen_ai.usage.total_tokens": 30,
            "gen_ai.server.time_to_first_token": 10,
            "gen_ai.server.request.duration": 20,
        }
    )

    mock_span.add_event.assert_called_with(
        "gen_ai.client.inference.operation.details",
        attributes={"gen_ai.output.messages": expected_output},
    )


def test_end_model_invoke_span_non_langfuse_no_extra_attributes(mock_span, monkeypatch):
    """Test that end_model_invoke_span doesn't add extra attributes for non-Langfuse endpoints."""
    monkeypatch.setenv("OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental")
    monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://api.honeycomb.io")

    tracer = Tracer()
    message = {"role": "assistant", "content": [{"text": "Response"}]}
    usage = Usage(inputTokens=10, outputTokens=20, totalTokens=30)
    metrics = Metrics(latencyMs=20, timeToFirstByteMs=10)
    stop_reason: StopReason = "end_turn"

    tracer.end_model_invoke_span(mock_span, message, usage, metrics, stop_reason)

    # Verify that set_attribute was NOT called with gen_ai.output.messages
    # (it should only be in the event, not as an attribute)
    expected_output = serialize(
        [
            {
                "role": "assistant",
                "parts": [{"type": "text", "content": "Response"}],
                "finish_reason": "end_turn",
            }
        ]
    )

    # Check that gen_ai.output.messages was not set as an attribute
    set_attribute_calls = [call[0][0] for call in mock_span.set_attribute.call_args_list]
    assert "gen_ai.output.messages" not in set_attribute_calls

    # But verify that add_event was still called
    mock_span.add_event.assert_called_with(
        "gen_ai.client.inference.operation.details",
        attributes={"gen_ai.output.messages": expected_output},
    )


class TestIsLangfuse:
    """Tests for the is_langfuse property."""

    def test_is_langfuse_with_otel_exporter_otlp_endpoint(self, monkeypatch):
        """Test is_langfuse returns True when OTEL_EXPORTER_OTLP_ENDPOINT contains langfuse."""
        monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://us.cloud.langfuse.com")
        tracer = Tracer()
        assert tracer.is_langfuse is True

    def test_is_langfuse_with_otel_exporter_otlp_traces_endpoint(self, monkeypatch):
        """Test is_langfuse returns True when OTEL_EXPORTER_OTLP_TRACES_ENDPOINT contains langfuse."""
        monkeypatch.setenv(
            "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "https://us.cloud.langfuse.com/api/public/otel/v1/traces"
        )
        tracer = Tracer()
        assert tracer.is_langfuse is True

    def test_is_langfuse_with_langfuse_base_url(self, monkeypatch):
        """Test is_langfuse returns True when LANGFUSE_BASE_URL contains langfuse."""
        monkeypatch.setenv("LANGFUSE_BASE_URL", "https://us.cloud.langfuse.com")
        tracer = Tracer()
        assert tracer.is_langfuse is True

    def test_is_langfuse_false_when_no_langfuse_env_vars(self, monkeypatch):
        """Test is_langfuse returns False when no Langfuse-related env vars are set."""
        monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
        monkeypatch.delenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", raising=False)
        monkeypatch.delenv("LANGFUSE_BASE_URL", raising=False)
        tracer = Tracer()
        assert tracer.is_langfuse is False

    def test_is_langfuse_false_with_non_langfuse_endpoint(self, monkeypatch):
        """Test is_langfuse returns False when endpoint is not Langfuse."""
        monkeypatch.setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://api.honeycomb.io")
        monkeypatch.delenv("LANGFUSE_BASE_URL", raising=False)
        tracer = Tracer()
        assert tracer.is_langfuse is False

    def test_is_langfuse_false_with_non_langfuse_base_url(self, monkeypatch):
        """Test is_langfuse returns False when LANGFUSE_BASE_URL doesn't contain langfuse."""
        monkeypatch.setenv("LANGFUSE_BASE_URL", "https://some-other-service.com")
        monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False)
        monkeypatch.delenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", raising=False)
        tracer = Tracer()
        assert tracer.is_langfuse is False