import json
import re
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Any, Optional

import httpx
import pytest
import respx
from openai import AsyncOpenAI
from openinference.semconv.trace import (
    MessageAttributes,
    SpanAttributes,
    ToolAttributes,
    ToolCallAttributes,
)
from opentelemetry.semconv.attributes.url_attributes import URL_FULL, URL_PATH
from sqlalchemy import select
from strawberry.scalars import JSON

from phoenix.db import models
from phoenix.db.types.annotation_configs import (
    CategoricalAnnotationValue,
    CategoricalOutputConfig,
    OptimizationDirection,
)
from phoenix.db.types.db_helper_types import UNDEFINED
from phoenix.db.types.evaluators import InputMapping
from phoenix.db.types.identifier import Identifier
from phoenix.db.types.model_provider import (
    LLMClientFactory,
    ModelProvider,
    openai_rate_limit_key,
)
from phoenix.db.types.prompts import (
    PromptChatTemplate,
    PromptMessage,
    PromptOpenAIInvocationParameters,
    PromptOpenAIInvocationParametersContent,
    PromptResponseFormatJSONSchema,
    PromptResponseFormatJSONSchemaDefinition,
    PromptTemplateFormat,
    PromptTemplateType,
    PromptToolChoiceNone,
    PromptToolChoiceOneOrMore,
    PromptToolChoiceSpecificFunctionTool,
    PromptToolChoiceZeroOrMore,
    PromptToolFunction,
    PromptToolFunctionDefinition,
    PromptTools,
    TextContentPart,
)
from phoenix.server.api.evaluators import (
    BuiltInEvaluator,
    LLMEvaluator,
    apply_input_mapping,
    cast_template_variable_types,
    get_evaluators,
    json_diff_count,
    levenshtein_distance,
    validate_template_variables,
)
from phoenix.server.api.evaluators import (
    LLMEvaluator as LLMEvaluatorClass,
)
from phoenix.server.api.exceptions import BadRequest, NotFound
from phoenix.server.api.helpers.evaluators import (
    _LLMEvaluatorPromptErrorMessage,
    validate_evaluator_prompt_and_configs,
)
from phoenix.server.api.helpers.playground_clients import OpenAIStreamingClient
from phoenix.server.api.input_types.PlaygroundEvaluatorInput import EvaluatorInputMappingInput
from phoenix.server.daemons.generative_model_store import GenerativeModelStore
from phoenix.server.daemons.span_cost_calculator import SpanCostCalculator
from phoenix.server.sandbox.session_manager import SandboxSessionManager
from phoenix.server.types import DbSessionFactory
from phoenix.trace.attributes import flatten
from phoenix.tracers import Tracer
from tests.unit.vcr import CustomVCR


def validate_consistent_llm_evaluator_and_prompt_version(
    prompt_version: models.PromptVersion,
    output_config: CategoricalOutputConfig,
    description: Optional[str] = None,
) -> None:
    """Test helper that wraps validate_evaluator_prompt_and_configs."""
    validate_evaluator_prompt_and_configs(
        prompt_tools=prompt_version.tools,
        prompt_response_format=prompt_version.response_format,
        evaluator_output_configs=[output_config],
        evaluator_description=description,
    )


class TestValidateConsistentLLMEvaluatorAndPromptVersion:
    def test_both_descriptions_null_does_not_raise(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        validate_consistent_llm_evaluator_and_prompt_version(
            prompt_version, output_config, description=None
        )

    def test_both_descriptions_strings_do_not_raise(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        description = "evaluates the correctness of the output"
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.description = description
        validate_consistent_llm_evaluator_and_prompt_version(
            prompt_version, output_config, description=description
        )

    def test_string_evaluator_description_and_null_function_description_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.EVALUATOR_DESCRIPTION_MUST_MATCH_FUNCTION_DESCRIPTION,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(
                prompt_version,
                output_config,
                description="a string description",
            )

    def test_null_evaluator_description_and_string_function_description_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.description = "a string description"
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.EVALUATOR_DESCRIPTION_MUST_MATCH_FUNCTION_DESCRIPTION,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(
                prompt_version,
                output_config,
                description="my different description",
            )

    def test_non_null_response_format_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        prompt_version.response_format = PromptResponseFormatJSONSchema(
            type="json_schema",
            json_schema=PromptResponseFormatJSONSchemaDefinition(
                name="test",
                schema={"type": "object"},
            ),
        )
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.RESPONSE_FORMAT_NOT_SUPPORTED,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_null_tools_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        prompt_version.tools = None
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.TOOLS_REQUIRED,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_empty_tools_list_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        prompt_version.tools.__dict__[
            "tools"
        ] = []  # skips validation for empty tools list on PromptTools type
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.TOOL_COUNT_MUST_MATCH_CONFIG_COUNT,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_multiple_tools_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        prompt_version.tools.tools.append(
            PromptToolFunction(
                type="function",
                function=PromptToolFunctionDefinition(
                    name="hallucination_evaluator",
                    description="evaluates hallucinations",
                    parameters={
                        "type": "object",
                        "properties": {
                            "label": {
                                "type": "string",
                                "enum": ["hallucinated", "not_hallucinated"],
                                "description": "hallucination",
                            }
                        },
                        "required": ["label"],
                    },
                ),
            )
        )
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.TOOL_COUNT_MUST_MATCH_CONFIG_COUNT,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_null_tool_choice_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        prompt_version.tools.tool_choice = PromptToolChoiceNone(type="none")
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.TOOL_CHOICE_REQUIRED,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_zero_or_more_tool_choice_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        prompt_version.tools.tool_choice = PromptToolChoiceZeroOrMore(type="zero_or_more")
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.TOOL_CHOICE_REQUIRED,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_specific_function_tool_choice_with_matching_name_does_not_raise(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        prompt_version.tools.tool_choice = PromptToolChoiceSpecificFunctionTool(
            type="specific_function", function_name="correctness"
        )
        validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_specific_function_tool_choice_with_mismatched_name_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        prompt_version.tools.tool_choice = PromptToolChoiceSpecificFunctionTool(
            type="specific_function", function_name="different_function_name"
        )
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.TOOL_CHOICE_SPECIFIC_FUNCTION_NAME_MUST_MATCH_DEFINED_FUNCTION_NAME,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_function_parameters_type_not_object_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters["type"] = "array"
        with pytest.raises(
            ValueError,
            match="^'correctness' function has errors. At 'type': Input should be 'object'.$",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_empty_function_parameters_properties_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters["properties"] = {}
        with pytest.raises(
            ValueError,
            match="^'correctness' function has errors. At 'properties.label': Field required.$",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_missing_label_property_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "other_property": {
                    "type": "string",
                    "enum": ["a", "b"],
                    "description": "some description",
                }
            },
            "required": ["other_property"],
        }
        with pytest.raises(
            ValueError,
            match="^'correctness' function has errors. At 'properties.label': Field required.$",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_label_property_type_not_string_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "number",
                    "enum": ["0", "1"],
                    "description": "correctness",
                }
            },
            "required": ["label"],
        }
        with pytest.raises(
            ValueError,
            match="^'correctness' function has errors. At 'properties.label.type': Input should be 'string'.$",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_label_enum_less_than_two_items_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct"],
                    "description": "correctness",
                }
            },
            "required": ["label"],
        }
        with pytest.raises(
            ValueError,
            match="^'correctness' function has errors. At 'properties.label.enum': List should have at least 2 items after validation, not 1.$",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_label_missing_enum_field_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "description": "correctness",
                }
            },
            "required": ["label"],
        }
        with pytest.raises(
            ValueError,
            match="^'correctness' function has errors. At 'properties.label.enum': Field required.$",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_label_missing_description_field_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct", "incorrect"],
                }
            },
            "required": ["label"],
        }
        with pytest.raises(
            ValueError,
            match="^'correctness' function has errors. At 'properties.label.description': Field required.$",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_duplicate_function_parameters_required_values_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters["required"] = [
            "label",
            "label",
        ]
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.REQUIRED_VALUES_MUST_BE_UNIQUE,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_label_defined_but_not_required_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct", "incorrect"],
                    "description": "correctness",
                }
            },
            "required": [],
        }
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.MISSING_REQUIRED_PROPERTIES.format(
                properties="label",
            ),
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_label_and_explanation_defined_but_only_label_required_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct", "incorrect"],
                    "description": "correctness",
                },
                "explanation": {
                    "type": "string",
                    "description": "explanation for the label",
                },
            },
            "required": ["label"],
        }
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.MISSING_REQUIRED_PROPERTIES.format(
                properties="explanation",
            ),
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_unexpected_required_property_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct", "incorrect"],
                    "description": "correctness",
                }
            },
            "required": ["label", "confidence"],
        }
        with pytest.raises(
            ValueError,
            match="Found unexpected required properties: confidence",
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_label_description_mismatch_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct", "incorrect"],
                    "description": "different_name",
                }
            },
            "required": ["label"],
        }
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.EVALUATOR_ANNOTATION_NAME_MUST_MATCH_FUNCTION_LABEL_PROPERTY_DESCRIPTION,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_choices_mismatch_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters["properties"]["label"]["enum"].append(
            "neutral"
        )
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.EVALUATOR_CHOICES_MUST_MATCH_TOOL_FUNCTION_LABELS,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_with_explanation_property_does_not_raise(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct", "incorrect"],
                    "description": "correctness",
                },
                "explanation": {
                    "type": "string",
                    "description": "explanation for the label",
                },
            },
            "required": ["label", "explanation"],
        }
        validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)

    def test_explanation_property_explicitly_set_to_none_raises(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        assert prompt_version.tools is not None
        assert prompt_version.tools.tools[0].type == "function"
        prompt_version.tools.tools[0].function.parameters = {
            "type": "object",
            "properties": {
                "label": {
                    "type": "string",
                    "enum": ["correct", "incorrect"],
                    "description": "correctness",
                },
                "explanation": None,
            },
            "required": ["label"],
        }
        with pytest.raises(
            ValueError,
            match=_LLMEvaluatorPromptErrorMessage.EXPLANATION_PROPERTIES_MUST_BE_STRING_OR_OMITTED,
        ):
            validate_consistent_llm_evaluator_and_prompt_version(prompt_version, output_config)


class TestMultiConfigValidation:
    """Tests for validate_consistent_llm_evaluator_and_prompt_version with multiple configs.

    Only the first (primary) config is validated against the prompt tool definition.
    Additional configs are independent annotation outputs that don't need to match
    the prompt tool structure. These tests document that behaviour.
    """

    def test_multi_config_with_matching_tools_does_not_raise(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        """When each config has a matching tool definition (by name), validation
        should pass. Each config's name must match a tool function name."""
        from phoenix.server.api.helpers.evaluators import (
            validate_consistent_llm_evaluator_and_prompt_version as validate_fn,
        )

        # Build a second config that matches a second tool
        second_config = CategoricalOutputConfig(
            type="CATEGORICAL",
            name="hallucination",
            optimization_direction=OptimizationDirection.MAXIMIZE,
            description="hallucination detection",
            values=[
                CategoricalAnnotationValue(label="hallucinated", score=1.0),
                CategoricalAnnotationValue(label="factual", score=0.0),
            ],
        )
        # Add a second tool matching the second config name
        assert prompt_version.tools is not None
        prompt_version.tools.tools.append(
            PromptToolFunction(
                type="function",
                function=PromptToolFunctionDefinition(
                    name="hallucination",
                    description=UNDEFINED,
                    parameters={
                        "type": "object",
                        "properties": {
                            "label": {
                                "type": "string",
                                "enum": ["hallucinated", "factual"],
                                "description": "hallucination",
                            }
                        },
                        "required": ["label"],
                    },
                ),
            )
        )
        llm_evaluator = models.LLMEvaluator(
            prompt_id=1,
            name="test",
            description=None,
            output_configs=[output_config, second_config],
        )
        # Should not raise because each config has a matching tool
        validate_fn(prompt_version, llm_evaluator)

    def test_multi_config_all_categorical_required(
        self,
        output_config: CategoricalOutputConfig,
        prompt_version: models.PromptVersion,
    ) -> None:
        """All configs must be CategoricalAnnotationConfig, even if only the primary
        config is validated against the prompt tool."""
        from phoenix.db.types.annotation_configs import ContinuousOutputConfig
        from phoenix.server.api.helpers.evaluators import (
            validate_consistent_llm_evaluator_and_prompt_version as validate_fn,
        )

        non_categorical_config = ContinuousOutputConfig(
            type="CONTINUOUS",
            name="score",
            optimization_direction=OptimizationDirection.MAXIMIZE,
            lower_bound=0.0,
            upper_bound=1.0,
        )
        llm_evaluator = models.LLMEvaluator(
            prompt_id=1,
            name="test",
            description=None,
            output_configs=[output_config, non_categorical_config],
        )
        with pytest.raises(ValueError, match="CategoricalAnnotationConfig"):
            validate_fn(prompt_version, llm_evaluator)


class TestApplyInputMapping:
    def test_extracts_value_using_jsonpath_expression(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"output": {"type": "string"}},
            "required": ["output"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"output": "$.response"}),
            literal_mapping=JSON({}),
        )
        context = {"response": "Hello, world!"}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"output": "Hello, world!"}

    def test_extracts_nested_value_using_jsonpath(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"text": {"type": "string"}},
            "required": ["text"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"text": "$.data.nested.value"}),
            literal_mapping=JSON({}),
        )
        context = {"data": {"nested": {"value": "deep content"}}}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"text": "deep content"}

    def test_literal_mapping_overrides_path_mapping(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"key": {"type": "string"}},
            "required": ["key"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"key": "$.from_path"}),
            literal_mapping=JSON({"key": "literal_value"}),
        )
        context = {"from_path": "path_value"}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"key": "literal_value"}

    def test_falls_back_to_context_for_unmapped_schema_keys(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {
                "input": {"type": "string"},
                "output": {"type": "string"},
            },
            "required": ["input", "output"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON({}),
        )
        context = {"input": "user input", "output": "model output"}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"input": "user input", "output": "model output"}

    def test_raises_on_invalid_jsonpath_expression(self) -> None:
        # Invalid JSONPath expressions are now validated at construction time
        with pytest.raises(BadRequest, match=r"Invalid JSONPath expression for key 'key'"):
            EvaluatorInputMappingInput(
                path_mapping=JSON({"key": "[[[invalid jsonpath"}),
                literal_mapping=JSON({}),
            )

    def test_raises_when_jsonpath_has_no_matches(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"key": {"type": "string"}},
            "required": ["key"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"key": "$.nonexistent.path"}),
            literal_mapping=JSON({}),
        )
        context = {"other": "value", "key": "fallback"}
        with pytest.raises(
            ValueError,
            match=r"JSONPath expression '\$\.nonexistent\.path' for key 'key' "
            r"did not match any values",
        ):
            apply_input_mapping(
                input_schema=input_schema,
                input_mapping=input_mapping.to_orm(),
                context=context,
            )

    def test_with_empty_mappings_uses_context_fallback(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"a": {"type": "string"}, "b": {"type": "string"}},
            "required": ["a", "b"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON({}),
        )
        context = {"a": "value_a", "b": "value_b", "c": "value_c"}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        # Only keys in schema are included
        assert result == {"a": "value_a", "b": "value_b"}

    def test_combines_path_literal_and_fallback_sources(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {
                "from_path": {"type": "string"},
                "from_literal": {"type": "string"},
                "from_fallback": {"type": "string"},
            },
            "required": ["from_path", "from_literal", "from_fallback"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"from_path": "$.extracted"}),
            literal_mapping=JSON({"from_literal": "hardcoded"}),
        )
        context = {"extracted": "path_result", "from_fallback": "context_value"}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {
            "from_path": "path_result",
            "from_literal": "hardcoded",
            "from_fallback": "context_value",
        }

    def test_returns_list_for_multi_match_jsonpath(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"item": {"type": "array", "items": {"type": "string"}}},
            "required": ["item"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"item": "$.items[*]"}),
            literal_mapping=JSON({}),
        )
        context = {"items": ["first", "second", "third"]}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"item": ["first", "second", "third"]}

    def test_path_mapping_extracts_array_value(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {
                "list": {
                    "type": "array",
                    "items": {"type": "number"},
                }
            },
            "required": ["list"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"list": "$.data"}),
            literal_mapping=JSON({}),
        )
        context = {"data": [1, 2, 3]}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"list": [1, 2, 3]}

    def test_path_mapping_extracts_object_value(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {
                "obj": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "number"},
                        "b": {"type": "number"},
                    },
                }
            },
            "required": ["obj"],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"obj": "$.nested"}),
            literal_mapping=JSON({}),
        )
        context = {"nested": {"a": 1, "b": 2}}
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"obj": {"a": 1, "b": 2}}

    def test_path_mapping_with_llm_messages_structure(self) -> None:
        """Test deep path access into LLM message structures like those from subscriptions.py.

        The context_dict in chat_completion subscriptions contains input/output keys
        with values that are message arrays. This test verifies JSONPath can traverse
        into these structures (important after removing json.dumps from context values).
        """
        input_schema = {
            "type": "object",
            "properties": {"content": {"type": "string"}},
            "required": ["content"],
        }
        # Simulates the structure of LLM_INPUT_MESSAGES attribute
        context = {
            "input": [
                {"message": {"role": "user", "content": "What is 2+2?"}},
            ],
            "output": [
                {"message": {"role": "assistant", "content": "4"}},
            ],
        }
        # Extract content from first output message
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"content": "$.output[0].message.content"}),
            literal_mapping=JSON({}),
        )
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"content": "4"}

    def test_path_mapping_extracts_all_message_contents(self) -> None:
        """Test extracting all message contents from a multi-message conversation."""
        input_schema = {
            "type": "object",
            "properties": {"messages": {}},
            "required": ["messages"],
        }
        context = {
            "input": [
                {"message": {"role": "user", "content": "Hello"}},
                {"message": {"role": "assistant", "content": "Hi there!"}},
                {"message": {"role": "user", "content": "How are you?"}},
            ],
        }
        # Extract all message contents using wildcard
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"messages": "$.input[*].message.content"}),
            literal_mapping=JSON({}),
        )
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"messages": ["Hello", "Hi there!", "How are you?"]}

    def test_path_mapping_with_dataset_revision_structure(self) -> None:
        """Test path access with dataset revision input/output structures.

        The context_dict in chat_completion_over_dataset uses revision.input,
        revision.output, and run.output which are typically dict objects.
        """
        input_schema = {
            "type": "object",
            "properties": {
                "question": {"type": "string"},
                "expected_answer": {"type": "string"},
                "actual_answer": {"type": "string"},
            },
            "required": ["question", "expected_answer", "actual_answer"],
        }
        # Simulates context from chat_completion_over_dataset
        context = {
            "input": {"question": "What is the capital of France?", "category": "geography"},
            "expected": {"answer": "Paris", "confidence": 0.99},
            "output": {"messages": [{"role": "assistant", "content": "Paris"}]},
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "question": "$.input.question",
                    "expected_answer": "$.expected.answer",
                    "actual_answer": "$.output.messages[0].content",
                }
            ),
            literal_mapping=JSON({}),
        )
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {
            "question": "What is the capital of France?",
            "expected_answer": "Paris",
            "actual_answer": "Paris",
        }

    def test_path_mapping_cannot_traverse_stringified_json(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"content": {"type": "string"}},
            "required": ["content"],
        }
        # If context values were stringified, deep paths wouldn't work
        context = {
            "output": json.dumps([{"message": {"role": "assistant", "content": "4"}}]),
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"content": "$.output[0].message.content"}),
            literal_mapping=JSON({}),
        )
        with pytest.raises(
            ValueError,
            match=r"JSONPath expression '\$\.output\[0\]\.message\.content' for key 'content' "
            r"did not match any values",
        ):
            apply_input_mapping(
                input_schema=input_schema,
                input_mapping=input_mapping.to_orm(),
                context=context,
            )

    def test_path_mapping_with_complex_nested_tool_calls(self) -> None:
        input_schema = {
            "type": "object",
            "properties": {"function_name": {"type": "string"}},
            "required": ["function_name"],
        }
        context = {
            "output": [
                {
                    "message": {
                        "role": "assistant",
                        "tool_calls": [
                            {
                                "tool_call": {
                                    "id": "call_123",
                                    "function": {
                                        "name": "get_weather",
                                        "arguments": '{"city": "NYC"}',
                                    },
                                }
                            }
                        ],
                    }
                }
            ],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {"function_name": "$.output[0].message.tool_calls[0].tool_call.function.name"}
            ),
            literal_mapping=JSON({}),
        )
        result = apply_input_mapping(
            input_schema=input_schema,
            input_mapping=input_mapping.to_orm(),
            context=context,
        )
        assert result == {"function_name": "get_weather"}


class TestCastTemplateVariableTypes:
    def test_converts_int_to_string(self) -> None:
        template_variables = {"count": 42}
        input_schema = {
            "type": "object",
            "properties": {"count": {"type": "string"}},
        }
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"count": "42"}

    def test_converts_list_to_string(self) -> None:
        template_variables = {"items": [1, 2, 3]}
        input_schema = {
            "type": "object",
            "properties": {"items": {"type": "string"}},
        }
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"items": "[1, 2, 3]"}

    def test_converts_dict_to_string(self) -> None:
        template_variables = {"data": {"key": "value"}}
        input_schema = {
            "type": "object",
            "properties": {"data": {"type": "string"}},
        }
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"data": "{'key': 'value'}"}

    def test_converts_none_to_string(self) -> None:
        template_variables = {"value": None}
        input_schema = {
            "type": "object",
            "properties": {"value": {"type": "string"}},
        }
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"value": "None"}

    def test_leaves_existing_string_unchanged(self) -> None:
        template_variables = {"text": "hello world"}
        input_schema = {
            "type": "object",
            "properties": {"text": {"type": "string"}},
        }
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"text": "hello world"}

    def test_ignores_non_string_schema_types(self) -> None:
        template_variables = {"count": 42, "flag": True}
        input_schema = {
            "type": "object",
            "properties": {
                "count": {"type": "number"},
                "flag": {"type": "boolean"},
            },
        }
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"count": 42, "flag": True}

    def test_preserves_keys_not_in_schema(self) -> None:
        template_variables = {"in_schema": 123, "not_in_schema": 456}
        input_schema = {
            "type": "object",
            "properties": {"in_schema": {"type": "string"}},
        }
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"in_schema": "123", "not_in_schema": 456}

    def test_handles_empty_schema(self) -> None:
        template_variables = {"key": 42}
        input_schema: dict[str, Any] = {}
        result = cast_template_variable_types(
            template_variables=template_variables,
            input_schema=input_schema,
        )
        assert result == {"key": 42}


class TestValidateTemplateVariables:
    def test_passes_with_valid_input(self) -> None:
        template_variables = {"name": "Alice", "age": "30"}
        input_schema = {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "string"},
            },
            "required": ["name", "age"],
        }
        validate_template_variables(
            template_variables=template_variables,
            input_schema=input_schema,
        )

    def test_raises_on_missing_required_field(self) -> None:
        template_variables = {"name": "Alice"}
        input_schema = {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "string"},
            },
            "required": ["name", "age"],
        }
        with pytest.raises(
            ValueError, match="Input validation failed.*'age' is a required property"
        ):
            validate_template_variables(
                template_variables=template_variables,
                input_schema=input_schema,
            )

    def test_raises_on_wrong_type(self) -> None:
        template_variables = {"count": "not a number"}
        input_schema = {
            "type": "object",
            "properties": {
                "count": {"type": "number"},
            },
            "required": ["count"],
        }
        with pytest.raises(ValueError, match="Input validation failed.*is not of type 'number'"):
            validate_template_variables(
                template_variables=template_variables,
                input_schema=input_schema,
            )

    def test_passes_with_extra_fields(self) -> None:
        template_variables = {"name": "Alice", "extra": "ignored"}
        input_schema = {
            "type": "object",
            "properties": {
                "name": {"type": "string"},
            },
            "required": ["name"],
        }
        validate_template_variables(
            template_variables=template_variables,
            input_schema=input_schema,
        )

    def test_with_nested_object_schema(self) -> None:
        template_variables = {
            "user": {"name": "Alice", "email": "alice@example.com"},
        }
        input_schema = {
            "type": "object",
            "properties": {
                "user": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "email": {"type": "string"},
                    },
                    "required": ["name", "email"],
                },
            },
            "required": ["user"],
        }
        # Should not raise
        validate_template_variables(
            template_variables=template_variables,
            input_schema=input_schema,
        )


class TestLevenshteinDistance:
    def test_identical_strings_returns_zero(self) -> None:
        assert levenshtein_distance("hello", "hello") == 0

    def test_empty_strings_returns_zero(self) -> None:
        assert levenshtein_distance("", "") == 0

    def test_one_empty_string_returns_length_of_other(self) -> None:
        assert levenshtein_distance("hello", "") == 5
        assert levenshtein_distance("", "world") == 5

    def test_single_character_difference(self) -> None:
        assert levenshtein_distance("cat", "bat") == 1

    def test_insertion(self) -> None:
        assert levenshtein_distance("cat", "cats") == 1

    def test_deletion(self) -> None:
        assert levenshtein_distance("cats", "cat") == 1

    def test_multiple_operations(self) -> None:
        assert levenshtein_distance("kitten", "sitting") == 3

    def test_completely_different_strings(self) -> None:
        assert levenshtein_distance("abc", "xyz") == 3

    def test_case_sensitive(self) -> None:
        assert levenshtein_distance("Hello", "hello") == 1

    def test_unicode_characters(self) -> None:
        assert levenshtein_distance("café", "cafe") == 1


class TestJsonDiffCount:
    def test_identical_objects_returns_zero(self) -> None:
        obj = {"a": 1, "b": 2}
        assert json_diff_count(obj, obj.copy()) == 0

    def test_identical_arrays_returns_zero(self) -> None:
        arr = [1, 2, 3]
        assert json_diff_count(arr, arr.copy()) == 0

    def test_identical_primitives_returns_zero(self) -> None:
        assert json_diff_count(42, 42) == 0
        assert json_diff_count("hello", "hello") == 0
        assert json_diff_count(True, True) == 0
        assert json_diff_count(None, None) == 0

    def test_different_primitives_returns_one(self) -> None:
        assert json_diff_count(42, 43) == 1
        assert json_diff_count("hello", "world") == 1
        assert json_diff_count(True, False) == 1

    def test_different_types_returns_one(self) -> None:
        assert json_diff_count(42, "42") == 1
        assert json_diff_count([1, 2], {"a": 1}) == 1
        assert json_diff_count(None, 0) == 1

    def test_object_with_missing_key(self) -> None:
        assert json_diff_count({"a": 1}, {"a": 1, "b": 2}) == 1
        assert json_diff_count({"a": 1, "b": 2}, {"a": 1}) == 1

    def test_object_with_different_value(self) -> None:
        assert json_diff_count({"a": 1}, {"a": 2}) == 1

    def test_nested_objects(self) -> None:
        expected = {"a": {"b": 1}}
        actual = {"a": {"b": 2}}
        assert json_diff_count(expected, actual) == 1

    def test_deeply_nested_difference(self) -> None:
        expected = {"a": {"b": {"c": 1}}}
        actual = {"a": {"b": {"c": 2}}}
        assert json_diff_count(expected, actual) == 1

    def test_multiple_nested_differences(self) -> None:
        expected = {"a": 1, "b": {"c": 2, "d": 3}}
        actual = {"a": 2, "b": {"c": 2, "d": 4}}
        assert json_diff_count(expected, actual) == 2

    def test_array_with_different_length(self) -> None:
        assert json_diff_count([1, 2, 3], [1, 2]) == 1
        assert json_diff_count([1, 2], [1, 2, 3]) == 1

    def test_array_with_different_elements(self) -> None:
        assert json_diff_count([1, 2, 3], [1, 2, 4]) == 1

    def test_array_length_and_element_differences(self) -> None:
        assert json_diff_count([1, 2, 3], [1, 4]) == 2

    def test_complex_nested_structure(self) -> None:
        expected = {"users": [{"name": "Alice", "age": 30}, {"name": "Bob", "age": 25}]}
        actual = {"users": [{"name": "Alice", "age": 31}, {"name": "Charlie", "age": 25}]}
        assert json_diff_count(expected, actual) == 2

    def test_empty_objects_returns_zero(self) -> None:
        assert json_diff_count({}, {}) == 0

    def test_empty_arrays_returns_zero(self) -> None:
        assert json_diff_count([], []) == 0


class TestBuiltInEvaluatorsWithLLMContextStructures:
    """Tests for builtin evaluators using context structures similar to subscriptions.py.

    These tests verify that evaluators work correctly when context values are
    objects/dicts (not stringified JSON), allowing deep path access via JSONPath.
    """

    async def test_contains_evaluator_with_deep_path_into_output_messages(self) -> None:
        """Test ContainsEvaluator with path mapping into nested message structures."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        # Context similar to subscriptions.py chat_completion context_dict
        context = {
            "input": [{"message": {"role": "user", "content": "Tell me about Paris"}}],
            "output": [
                {"message": {"role": "assistant", "content": "Paris is the capital of France."}}
            ],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"text": "$.output[0].message.content"}),
            literal_mapping=JSON({"words": "France,capital"}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 1.0
        assert result["explanation"] is not None
        assert "found" in result["explanation"]

    async def test_contains_evaluator_with_dataset_context_structure(self) -> None:
        """Test ContainsEvaluator with dataset experiment context structure."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        # Context similar to subscriptions.py chat_completion_over_dataset context_dict
        context = {
            "input": {"question": "What is the capital of France?", "topic": "geography"},
            "expected": {"answer": "Paris", "country": "France"},
            "output": {
                "messages": [{"role": "assistant", "content": "The capital of France is Paris."}]
            },
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"text": "$.output.messages[0].content"}),
            literal_mapping=JSON({"words": "Paris"}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 1.0

    async def test_exact_match_evaluator_with_nested_paths(self) -> None:
        """Test ExactMatchEvaluator extracting values from nested structures."""
        from phoenix.server.api.evaluators import ExactMatchEvaluator

        evaluator = ExactMatchEvaluator()
        output_config = evaluator.output_configs[0]
        context = {
            "expected": {"answer": "Paris"},
            "output": {"response": {"text": "Paris"}},
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected.answer",
                    "actual": "$.output.response.text",
                }
            ),
            literal_mapping=JSON({}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="exact_match",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 1.0
        assert result["explanation"] is not None
        assert "matches" in result["explanation"]

    async def test_exact_match_evaluator_mismatch_with_nested_paths(self) -> None:
        """Test ExactMatchEvaluator with mismatched values from nested paths."""
        from phoenix.server.api.evaluators import ExactMatchEvaluator

        evaluator = ExactMatchEvaluator()
        output_config = evaluator.output_configs[0]
        context = {
            "expected": {"answer": "Paris"},
            "output": {"response": {"text": "London"}},
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected.answer",
                    "actual": "$.output.response.text",
                }
            ),
            literal_mapping=JSON({}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="exact_match",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 0.0
        assert result["explanation"] is not None
        assert "does not match" in result["explanation"]

    async def test_json_distance_evaluator_with_json_string_values(self) -> None:
        """Test JSONDistanceEvaluator with JSON string values in context.

        Note: JSONDistanceEvaluator expects valid JSON strings, not Python objects.
        When extracting objects via JSONPath and casting to string, they become
        Python repr strings (single quotes), not valid JSON. For this evaluator,
        context values should be pre-serialized JSON strings or provided via literals.
        """
        import json

        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        # Provide JSON strings directly in context
        context = {
            "expected_json": json.dumps({"items": [1, 2, 3]}),
            "actual_json": json.dumps({"items": [1, 2, 4]}),
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected_json",
                    "actual": "$.actual_json",
                }
            ),
            literal_mapping=JSON({}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # Distance is 1 because one element differs
        assert result["score"] == 1.0

    async def test_contains_evaluator_with_context_fallback(self) -> None:
        """Test evaluator using context fallback when no path mapping provided."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        # Direct context values (fallback behavior)
        context = {
            "words": "hello,world",
            "text": "Hello, World! How are you?",
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON({}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 1.0

    async def test_levenshtein_evaluator_with_message_content_paths(self) -> None:
        """Test LevenshteinDistanceEvaluator with LLM message structures."""
        from phoenix.server.api.evaluators import LevenshteinDistanceEvaluator

        evaluator = LevenshteinDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context = {
            "expected": [{"message": {"content": "Hello"}}],
            "output": [{"message": {"content": "Hallo"}}],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected[0].message.content",
                    "actual": "$.output[0].message.content",
                }
            ),
            literal_mapping=JSON({}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="levenshtein_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # Distance between "Hello" and "Hallo" is 1
        assert result["score"] == 1.0

    async def test_regex_evaluator_with_deep_path_extraction(self) -> None:
        """Test RegexEvaluator extracting text from nested structures."""
        from phoenix.server.api.evaluators import RegexEvaluator

        evaluator = RegexEvaluator()
        output_config = evaluator.output_configs[0]
        context = {
            "output": {
                "response": {
                    "content": "The answer is 42.",
                }
            }
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"text": "$.output.response.content"}),
            literal_mapping=JSON({"pattern": r"\d+"}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="regex",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 1.0
        assert result["explanation"] is not None
        assert "matched" in result["explanation"]

    async def test_evaluator_handles_list_values_from_context(self) -> None:
        """Test that evaluators can handle list values extracted from context."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        context = {
            "output": [
                {"message": {"content": "First response"}},
                {"message": {"content": "Second response with keyword"}},
            ],
        }
        # When extracting all message contents, the result is stringified by cast_template_variable_types
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"text": "$.output[*].message.content"}),
            literal_mapping=JSON({"words": "keyword"}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # The list gets stringified and searched
        assert result["score"] == 1.0

    async def test_evaluator_handles_dict_values_cast_to_string(self) -> None:
        """Test that dict values are properly cast to strings for evaluators expecting string input."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        context = {
            "output": {"nested": {"value": "contains target word"}},
        }
        # Extracting the whole nested object, which will be stringified
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"text": "$.output.nested"}),
            literal_mapping=JSON({"words": "target"}),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 1.0

    async def test_contains_evaluator_trailing_comma_no_false_positive(self) -> None:
        """Trailing comma should not produce an empty token that matches everything."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        result = (
            await evaluator.evaluate(
                context={"words": "cumin,", "text": "no spices here"},
                input_mapping=EvaluatorInputMappingInput(
                    path_mapping=JSON({}), literal_mapping=JSON({})
                ).to_orm(),
                name="contains",
                output_configs=[evaluator.output_configs[0]],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 0.0

    async def test_contains_evaluator_explanation_format(self) -> None:
        """Explanation should use clean word list (not repr) and include the text value."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        result = (
            await evaluator.evaluate(
                context={"words": "cumin,paprika", "text": "The recipe uses cumin and paprika"},
                input_mapping=EvaluatorInputMappingInput(
                    path_mapping=JSON({}), literal_mapping=JSON({})
                ).to_orm(),
                name="contains",
                output_configs=[evaluator.output_configs[0]],
            )
        )[0]
        assert result["error"] is None
        explanation = result["explanation"]
        assert explanation is not None
        assert "[cumin, paprika]" in explanation
        assert "The recipe uses cumin" in explanation

    async def test_contains_evaluator_empty_words_require_all_no_false_positive(self) -> None:
        """Empty words list with require_all=True should not produce a false positive."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        result = (
            await evaluator.evaluate(
                context={"words": ",", "text": "any text", "require_all": True},
                input_mapping=EvaluatorInputMappingInput(
                    path_mapping=JSON({}), literal_mapping=JSON({})
                ).to_orm(),
                name="contains",
                output_configs=[evaluator.output_configs[0]],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 0.0
        assert result["explanation"] == "No valid words were provided to check."


class TestJSONDistanceParseStringsToggle:
    """Tests for JSONDistanceEvaluator parse_strings toggle behavior.

    When parse_strings=true (default), string inputs are json.loads()'d before comparison.
    When parse_strings=false, inputs pass directly to json_diff_count() without parsing.
    """

    async def test_parse_strings_true_with_json_strings(self) -> None:
        """When parse_strings=true, JSON strings are parsed before comparison."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context: dict[str, Any] = {}
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON(
                {
                    "expected": json.dumps({"a": 1, "b": [2, 3]}),
                    "actual": json.dumps({"a": 1, "b": [2, 4]}),
                    "parse_strings": True,
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # One difference: b[1] is 3 vs 4
        assert result["score"] == 1.0

    async def test_parse_strings_false_with_dict_inputs(self) -> None:
        """When parse_strings=false, dict inputs pass directly to comparison."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        # Provide dicts directly in context (simulating JSONPath extraction)
        context: dict[str, Any] = {
            "expected_obj": {"items": [1, 2, 3], "name": "test"},
            "actual_obj": {"items": [1, 2, 3], "name": "changed"},
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected_obj",
                    "actual": "$.actual_obj",
                }
            ),
            literal_mapping=JSON(
                {
                    "parse_strings": False,
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # One difference: name "test" vs "changed"
        assert result["score"] == 1.0

    async def test_parse_strings_false_with_list_inputs(self) -> None:
        """When parse_strings=false, list inputs pass directly to comparison."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context: dict[str, Any] = {
            "expected_list": [1, 2, 3],
            "actual_list": [1, 2, 3],
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected_list",
                    "actual": "$.actual_list",
                }
            ),
            literal_mapping=JSON(
                {
                    "parse_strings": False,
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # Identical lists, distance 0
        assert result["score"] == 0.0

    async def test_parse_strings_false_with_bare_strings(self) -> None:
        """When parse_strings=false, bare strings are compared as-is without JSON parsing."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context: dict[str, Any] = {}
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON(
                {
                    "expected": "hello world",
                    "actual": "hello world",
                    "parse_strings": False,
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # Same strings compared directly, distance 0
        assert result["score"] == 0.0

    async def test_parse_strings_default_behaves_as_true(self) -> None:
        """When parse_strings is not provided, it defaults to true (JSON parsing)."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context: dict[str, Any] = {}
        # No parse_strings in literal_mapping — should default to true
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON(
                {
                    "expected": json.dumps({"key": "value"}),
                    "actual": json.dumps({"key": "value"}),
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # Identical JSON, distance 0
        assert result["score"] == 0.0

    async def test_parse_strings_true_with_identical_json_strings(self) -> None:
        """Identical JSON strings with parse_strings=true produce distance 0."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context: dict[str, Any] = {}
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON(
                {
                    "expected": json.dumps({"a": 1}),
                    "actual": json.dumps({"a": 1}),
                    "parse_strings": True,
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 0.0

    async def test_parse_strings_false_with_nested_dict_differences(self) -> None:
        """When parse_strings=false, nested dict differences are counted correctly."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context: dict[str, Any] = {
            "expected_obj": {"a": {"b": 1, "c": 2}, "d": 3},
            "actual_obj": {"a": {"b": 99, "c": 2}, "d": 100},
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected_obj",
                    "actual": "$.actual_obj",
                }
            ),
            literal_mapping=JSON(
                {
                    "parse_strings": False,
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        # Two differences: a.b (1 vs 99) and d (3 vs 100)
        assert result["score"] == 2.0

    async def test_parse_strings_false_tracing_metadata_serializes_non_strings(self) -> None:
        """When parse_strings=false with dict inputs, metadata contains serialized forms."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        expected_dict = {"key": "value"}
        actual_dict = {"key": "different"}
        context: dict[str, Any] = {
            "expected_obj": expected_dict,
            "actual_obj": actual_dict,
        }
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON(
                {
                    "expected": "$.expected_obj",
                    "actual": "$.actual_obj",
                }
            ),
            literal_mapping=JSON(
                {
                    "parse_strings": False,
                }
            ),
        )
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="json_distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["error"] is None
        assert result["score"] == 1.0
        # Metadata should contain readable serialized forms of the inputs
        metadata = result["metadata"]
        assert "expected" in metadata
        assert "actual" in metadata
        # Non-string inputs should be serialized to JSON strings in metadata
        assert json.loads(metadata["expected"]) == expected_dict
        assert json.loads(metadata["actual"]) == actual_dict

    async def test_input_schema_includes_parse_strings(self) -> None:
        """Verify the input_schema exposes parse_strings as a boolean property."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        schema = evaluator.input_schema
        assert "parse_strings" in schema["properties"]
        assert schema["properties"]["parse_strings"]["type"] == "boolean"
        # parse_strings should NOT be required
        assert "parse_strings" not in schema.get("required", [])

    async def test_input_schema_expected_actual_allow_non_string_types(self) -> None:
        """Verify expected/actual in input_schema don't restrict to type: string."""
        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        schema = evaluator.input_schema
        # expected and actual should NOT have "type": "string"
        assert "type" not in schema["properties"]["expected"]
        assert "type" not in schema["properties"]["actual"]


class TestBuiltInEvaluatorOutputConfigUsage:
    """Tests for builtin evaluator output_config usage at execution time."""

    async def test_contains_evaluator_uses_name_from_output_config(self) -> None:
        """Test that ContainsEvaluator uses the name in the result."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        context = {"words": "hello", "text": "hello world"}
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="My Custom Contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["name"] == "My Custom Contains"
        assert result["error"] is None

    async def test_contains_evaluator_maps_true_to_label_from_output_config(self) -> None:
        """Test that ContainsEvaluator maps a true result to the correct label."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        context = {"words": "hello", "text": "hello world"}
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["label"] == "true"
        assert result["score"] == 1.0

    async def test_contains_evaluator_maps_false_to_label_from_output_config(self) -> None:
        """Test that ContainsEvaluator maps a false result to the correct label."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        output_config = evaluator.output_configs[0]
        context = {"words": "hello", "text": "goodbye world"}
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="contains",
                output_configs=[output_config],
            )
        )[0]
        assert result["label"] == "false"
        assert result["score"] == 0.0

    async def test_exact_match_evaluator_uses_custom_output_config(self) -> None:
        """Test that ExactMatchEvaluator uses a custom output config."""
        from phoenix.server.api.evaluators import ExactMatchEvaluator

        evaluator = ExactMatchEvaluator()
        # Create a custom output config with different labels
        custom_config = CategoricalOutputConfig(
            type="CATEGORICAL",
            name="exact",
            optimization_direction=OptimizationDirection.MAXIMIZE,
            values=[
                CategoricalAnnotationValue(label="match", score=1.0),
                CategoricalAnnotationValue(label="no_match", score=0.0),
            ],
        )
        context = {"expected": "Paris", "actual": "Paris"}
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="Exact Match Custom",
                output_configs=[custom_config],
            )
        )[0]
        assert result["name"] == "Exact Match Custom"
        assert result["label"] == "match"
        assert result["score"] == 1.0

    async def test_regex_evaluator_uses_custom_output_config(self) -> None:
        """Test that RegexEvaluator uses a custom output config."""
        from phoenix.server.api.evaluators import RegexEvaluator

        evaluator = RegexEvaluator()
        # Create a custom output config with different labels
        custom_config = CategoricalOutputConfig(
            type="CATEGORICAL",
            name="regex",
            optimization_direction=OptimizationDirection.MAXIMIZE,
            values=[
                CategoricalAnnotationValue(label="pattern_found", score=1.0),
                CategoricalAnnotationValue(label="pattern_not_found", score=0.0),
            ],
        )
        context = {"pattern": r"\d+", "text": "The answer is 42"}
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="Regex Custom",
                output_configs=[custom_config],
            )
        )[0]
        assert result["name"] == "Regex Custom"
        assert result["label"] == "pattern_found"
        assert result["score"] == 1.0

    async def test_contains_evaluator_uses_custom_scores(self) -> None:
        """Test that ContainsEvaluator correctly uses custom scores from output config."""
        from phoenix.server.api.evaluators import ContainsEvaluator

        evaluator = ContainsEvaluator()
        custom_config = CategoricalOutputConfig(
            type="CATEGORICAL",
            name="contains_custom_scores",
            optimization_direction=OptimizationDirection.MAXIMIZE,
            values=[
                CategoricalAnnotationValue(label="Pass", score=100.0),
                CategoricalAnnotationValue(label="Fail", score=-50.0),
            ],
        )
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))

        result_match = (
            await evaluator.evaluate(
                context={"words": "hello", "text": "hello world"},
                input_mapping=input_mapping.to_orm(),
                name="Contains Custom",
                output_configs=[custom_config],
            )
        )[0]
        assert result_match["label"] == "Pass"
        assert result_match["score"] == 100.0

        result_no_match = (
            await evaluator.evaluate(
                context={"words": "goodbye", "text": "hello world"},
                input_mapping=input_mapping.to_orm(),
                name="Contains Custom",
                output_configs=[custom_config],
            )
        )[0]
        assert result_no_match["label"] == "Fail"
        assert result_no_match["score"] == -50.0

    async def test_levenshtein_evaluator_uses_name(self) -> None:
        """Test that LevenshteinDistanceEvaluator uses the name in the result."""
        from phoenix.server.api.evaluators import LevenshteinDistanceEvaluator

        evaluator = LevenshteinDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context = {"expected": "hello", "actual": "hallo"}
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="Edit Distance",
                output_configs=[output_config],
            )
        )[0]
        assert result["name"] == "Edit Distance"
        assert result["score"] == 1.0
        assert result["label"] is None

    async def test_json_distance_evaluator_uses_name(self) -> None:
        """Test that JSONDistanceEvaluator uses the name in the result."""
        import json

        from phoenix.server.api.evaluators import JSONDistanceEvaluator

        evaluator = JSONDistanceEvaluator()
        output_config = evaluator.output_configs[0]
        context = {
            "expected": json.dumps({"a": 1}),
            "actual": json.dumps({"a": 1}),
        }
        input_mapping = EvaluatorInputMappingInput(path_mapping=JSON({}), literal_mapping=JSON({}))
        result = (
            await evaluator.evaluate(
                context=context,
                input_mapping=input_mapping.to_orm(),
                name="JSON Diff",
                output_configs=[output_config],
            )
        )[0]
        assert result["name"] == "JSON Diff"
        assert result["score"] == 0.0
        assert result["label"] is None


class TestLLMEvaluator:
    @pytest.fixture
    async def project(self, db: DbSessionFactory) -> models.Project:
        project = models.Project(name="test-project")
        async with db() as session:
            session.add(project)
        return project

    @pytest.fixture
    async def gpt_4o_mini_generative_model(self, db: DbSessionFactory) -> models.GenerativeModel:
        model = models.GenerativeModel(
            name="gpt-4o-mini",
            provider="openai",
            start_time=datetime(2024, 1, 1, tzinfo=timezone.utc),
            name_pattern=re.compile("gpt-4o-mini"),
            is_built_in=True,
            token_prices=[
                models.TokenPrice(
                    token_type="input",
                    is_prompt=True,
                    base_rate=0.15 / 1_000_000,  # $0.15 per million tokens
                    customization=None,
                ),
                models.TokenPrice(
                    token_type="output",
                    is_prompt=False,
                    base_rate=0.60 / 1_000_000,  # $0.60 per million tokens
                    customization=None,
                ),
            ],
        )
        async with db() as session:
            session.add(model)

        return model

    @pytest.fixture
    async def generative_model_store(
        self,
        db: DbSessionFactory,
        gpt_4o_mini_generative_model: models.GenerativeModel,
    ) -> GenerativeModelStore:
        store = GenerativeModelStore(db=db)
        await store._fetch_models()
        return store

    @pytest.fixture
    def span_cost_calculator(
        self,
        db: DbSessionFactory,
        generative_model_store: GenerativeModelStore,
    ) -> SpanCostCalculator:
        return SpanCostCalculator(db=db, model_store=generative_model_store)

    @pytest.fixture
    def tracer(self, span_cost_calculator: SpanCostCalculator) -> Tracer:
        return Tracer(span_cost_calculator=span_cost_calculator)

    @pytest.fixture
    def openai_streaming_client(
        self,
        openai_api_key: str,
    ) -> "OpenAIStreamingClient":
        @asynccontextmanager
        async def create_openai_client() -> Any:
            yield AsyncOpenAI(api_key=openai_api_key, max_retries=0)

        client_factory = LLMClientFactory(
            create_openai_client, openai_rate_limit_key(openai_api_key, None)
        )
        return OpenAIStreamingClient(
            client_factory=client_factory,
            model_name="gpt-4o-mini",
            provider="openai",
        )

    @pytest.fixture
    def llm_evaluator_prompt_version(self) -> models.PromptVersion:
        return models.PromptVersion(
            prompt_id=1,
            template_type=PromptTemplateType.CHAT,
            template_format=PromptTemplateFormat.MUSTACHE,
            template=PromptChatTemplate(
                type="chat",
                messages=[
                    PromptMessage(
                        role="system",
                        content="You are an evaluator. Assess whether the output correctly answers the input question.",
                    ),
                    PromptMessage(
                        role="user",
                        content="Input: {{input}}\n\nOutput: {{output}}\n\nIs this output correct?",
                    ),
                ],
            ),
            invocation_parameters=PromptOpenAIInvocationParameters(
                type="openai",
                openai=PromptOpenAIInvocationParametersContent(temperature=0.0),
            ),
            tools=PromptTools(
                type="tools",
                tools=[
                    PromptToolFunction(
                        type="function",
                        function=PromptToolFunctionDefinition(
                            name="correctness",
                            description="Evaluates the correctness of the output",
                            parameters={
                                "type": "object",
                                "properties": {
                                    "label": {
                                        "type": "string",
                                        "enum": ["correct", "incorrect"],
                                        "description": "correctness",
                                    },
                                    "explanation": {
                                        "type": "string",
                                        "description": "Brief explanation for the label",
                                    },
                                },
                                "required": ["label", "explanation"],
                            },
                        ),
                    )
                ],
                tool_choice=PromptToolChoiceOneOrMore(type="one_or_more"),
            ),
            response_format=None,
            model_provider=ModelProvider.OPENAI,
            model_name="gpt-4o-mini",
            metadata_={},
        )

    @pytest.fixture
    def llm_evaluator(
        self,
        llm_evaluator_prompt_version: models.PromptVersion,
        output_config: CategoricalOutputConfig,
        openai_streaming_client: "OpenAIStreamingClient",
    ) -> LLMEvaluator:
        template = llm_evaluator_prompt_version.template
        assert isinstance(template, PromptChatTemplate)
        tools = llm_evaluator_prompt_version.tools
        assert tools is not None

        return LLMEvaluator(
            name="correctness",
            description="Evaluates correctness",
            template=template,
            template_format=llm_evaluator_prompt_version.template_format,
            tools=tools,
            invocation_parameters=llm_evaluator_prompt_version.invocation_parameters,
            model_provider=llm_evaluator_prompt_version.model_provider,
            llm_client=openai_streaming_client,
            output_configs=[output_config],
            prompt_name="test-prompt",
        )

    @pytest.fixture
    def input_mapping(self) -> EvaluatorInputMappingInput:
        return EvaluatorInputMappingInput(
            path_mapping=JSON({}),
            literal_mapping=JSON({}),
        )

    @pytest.fixture
    def multipart_llm_evaluator_prompt_version(self) -> models.PromptVersion:
        return models.PromptVersion(
            prompt_id=1,
            template_type=PromptTemplateType.CHAT,
            template_format=PromptTemplateFormat.MUSTACHE,
            template=PromptChatTemplate(
                type="chat",
                messages=[
                    PromptMessage(
                        role="system",
                        content=[
                            TextContentPart(
                                type="text",
                                text="You are an evaluator. ",
                            ),
                            TextContentPart(
                                type="text",
                                text="Assess whether the output correctly answers the input question.",
                            ),
                        ],
                    ),
                    PromptMessage(
                        role="user",
                        content=[
                            TextContentPart(
                                type="text",
                                text="Input: {{input}}\n\nOutput: {{output}}\n\nIs this output correct?",
                            ),
                        ],
                    ),
                ],
            ),
            invocation_parameters=PromptOpenAIInvocationParameters(
                type="openai",
                openai=PromptOpenAIInvocationParametersContent(temperature=0.0),
            ),
            tools=PromptTools(
                type="tools",
                tools=[
                    PromptToolFunction(
                        type="function",
                        function=PromptToolFunctionDefinition(
                            name="correctness",
                            description="Evaluates the correctness of the output",
                            parameters={
                                "type": "object",
                                "properties": {
                                    "label": {
                                        "type": "string",
                                        "enum": ["correct", "incorrect"],
                                        "description": "correctness",
                                    },
                                    "explanation": {
                                        "type": "string",
                                        "description": "Brief explanation for the label",
                                    },
                                },
                                "required": ["label", "explanation"],
                            },
                        ),
                    )
                ],
                tool_choice=PromptToolChoiceOneOrMore(type="one_or_more"),
            ),
            response_format=None,
            model_provider=ModelProvider.OPENAI,
            model_name="gpt-4o-mini",
            metadata_={},
        )

    @pytest.fixture
    def multipart_llm_evaluator(
        self,
        multipart_llm_evaluator_prompt_version: models.PromptVersion,
        output_config: CategoricalOutputConfig,
        openai_streaming_client: "OpenAIStreamingClient",
    ) -> LLMEvaluator:
        template = multipart_llm_evaluator_prompt_version.template
        assert isinstance(template, PromptChatTemplate)
        tools = multipart_llm_evaluator_prompt_version.tools
        assert tools is not None

        return LLMEvaluator(
            name="correctness",
            description="Evaluates correctness",
            template=template,
            template_format=multipart_llm_evaluator_prompt_version.template_format,
            tools=tools,
            invocation_parameters=multipart_llm_evaluator_prompt_version.invocation_parameters,
            model_provider=multipart_llm_evaluator_prompt_version.model_provider,
            llm_client=openai_streaming_client,
            output_configs=[output_config],
            prompt_name="test-prompt",
        )

    async def test_evaluate_returns_correct_result(
        self,
        db: DbSessionFactory,
        project: models.Project,
        tracer: Tracer,
        llm_evaluator: LLMEvaluator,
        output_config: CategoricalOutputConfig,
        input_mapping: EvaluatorInputMappingInput,
        custom_vcr: CustomVCR,
        gpt_4o_mini_generative_model: models.GenerativeModel,
    ) -> None:
        with custom_vcr.use_cassette():
            evaluation_result = (
                await llm_evaluator.evaluate(
                    context={"input": "What is 2 + 2?", "output": "4"},
                    input_mapping=input_mapping.to_orm(),
                    name="correctness",
                    output_configs=[output_config],
                    tracer=tracer,
                )
            )[0]

        result = dict(evaluation_result)
        assert result.pop("error") is None
        assert result.pop("label") == "correct"
        assert result.pop("score") == 1.0
        assert result.pop("explanation") is not None
        assert result.pop("annotator_kind") == "LLM"
        assert result.pop("name") == "correctness"
        trace_id = result.pop("trace_id")
        assert isinstance(trace_id, str)
        assert isinstance(result.pop("start_time"), datetime)
        assert isinstance(result.pop("end_time"), datetime)
        assert result.pop("metadata") == {}
        result.pop("error_exc", None)
        assert not result

        async with db() as session:
            db_traces = tracer.get_db_traces(project_id=project.id)
            session.add_all(db_traces)
            await session.flush()

        assert len(db_traces) == 1
        db_trace = db_traces[0]
        assert db_trace.trace_id == trace_id
        db_spans = db_trace.spans
        span_costs = db_trace.span_costs
        assert len(db_spans) == 5

        evaluator_span = None
        input_mapping_span = None
        prompt_span = None
        llm_span = None
        parse_eval_result_span = None
        for span in db_spans:
            if span.span_kind == "EVALUATOR":
                evaluator_span = span
            elif span.span_kind == "LLM":
                llm_span = span
            elif span.span_kind == "PROMPT":
                prompt_span = span
            elif span.span_kind == "CHAIN":
                if span.name == "Input Mapping":
                    input_mapping_span = span
                elif span.name == "Parse Eval Result":
                    parse_eval_result_span = span

        assert evaluator_span is not None
        assert input_mapping_span is not None
        assert prompt_span is not None
        assert llm_span is not None
        assert parse_eval_result_span is not None
        assert evaluator_span.parent_id is None
        assert input_mapping_span.parent_id == evaluator_span.span_id
        assert prompt_span.parent_id == evaluator_span.span_id
        assert llm_span.parent_id == evaluator_span.span_id
        assert parse_eval_result_span.parent_id == evaluator_span.span_id

        # evaluator span
        assert evaluator_span.name == "Evaluator: correctness"
        assert evaluator_span.status_code == "OK"
        assert not evaluator_span.events
        attributes = dict(flatten(evaluator_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "EVALUATOR"
        raw_input_value = attributes.pop(INPUT_VALUE)
        assert raw_input_value is not None
        input_value = json.loads(raw_input_value)
        assert set(input_value.keys()) == {"input", "output"}
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        raw_output_value = attributes.pop(OUTPUT_VALUE)
        assert raw_output_value is not None
        output_value = json.loads(raw_output_value)
        assert set(output_value.keys()) == {"results"}
        assert len(output_value["results"]) == 1
        assert output_value["results"][0]["name"] == "correctness"
        assert output_value["results"][0]["label"] == "correct"
        assert output_value["results"][0]["score"] == 1.0
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # Input Mapping span
        assert input_mapping_span.name == "Input Mapping"
        assert input_mapping_span.status_code == "OK"
        assert not input_mapping_span.events
        attributes = dict(flatten(input_mapping_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "CHAIN"
        raw_input_value = attributes.pop(INPUT_VALUE)
        input_value = json.loads(raw_input_value)
        assert input_value == {
            "input_mapping": {"path_mapping": {}, "literal_mapping": {}},
            "template_variables": {"input": "What is 2 + 2?", "output": "4"},
        }
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        raw_output_value = attributes.pop(OUTPUT_VALUE)
        output_value = json.loads(raw_output_value)
        assert output_value == {"input": "What is 2 + 2?", "output": "4"}
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # Prompt span
        assert prompt_span.name == "Prompt: test-prompt"
        assert prompt_span.status_code == "OK"
        assert not prompt_span.events
        attributes = dict(flatten(prompt_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "PROMPT"
        raw_input_value = attributes.pop(INPUT_VALUE)
        input_value = json.loads(raw_input_value)
        assert input_value == {"input": "What is 2 + 2?", "output": "4"}
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        raw_output_value = attributes.pop(OUTPUT_VALUE)
        output_value = json.loads(raw_output_value)
        assert output_value == {
            "messages": [
                {
                    "role": "system",
                    "content": "You are an evaluator. Assess whether the output correctly answers the input question.",
                },
                {
                    "role": "user",
                    "content": "Input: What is 2 + 2?\n\nOutput: 4\n\nIs this output correct?",
                },
            ]
        }
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # llm span
        assert llm_span.name == "ChatCompletion"
        assert llm_span.status_code == "OK"
        assert not llm_span.events
        assert llm_span.llm_token_count_prompt is not None
        assert llm_span.llm_token_count_prompt > 0
        assert llm_span.llm_token_count_completion is not None
        assert llm_span.llm_token_count_completion > 0
        assert llm_span.cumulative_llm_token_count_prompt > 0
        assert llm_span.cumulative_llm_token_count_completion > 0
        attributes = dict(flatten(llm_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "LLM"
        assert attributes.pop(LLM_MODEL_NAME) == "gpt-4o-mini"
        assert attributes.pop(LLM_PROVIDER) == "openai"
        assert attributes.pop(LLM_SYSTEM) == "openai"
        assert attributes.pop(f"{LLM_INPUT_MESSAGES}.0.{MESSAGE_ROLE}") == "system"
        assert "evaluator" in attributes.pop(f"{LLM_INPUT_MESSAGES}.0.{MESSAGE_CONTENT}").lower()
        assert attributes.pop(f"{LLM_INPUT_MESSAGES}.1.{MESSAGE_ROLE}") == "user"
        user_message = attributes.pop(f"{LLM_INPUT_MESSAGES}.1.{MESSAGE_CONTENT}")
        assert "What is 2 + 2?" in user_message
        assert "4" in user_message
        # Check token count attributes exist and are integers
        token_count_attribute_keys = [
            key for key in attributes if key.startswith("llm.token_count.")
        ]
        for key in token_count_attribute_keys:
            assert isinstance(attributes.pop(key), int)
        assert attributes.pop(URL_FULL) == "https://api.openai.com/v1/chat/completions"
        assert attributes.pop(URL_PATH) == "chat/completions"
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        raw_output_value = attributes.pop(OUTPUT_VALUE)
        output_value = json.loads(raw_output_value)
        assert output_value == {
            "id": "chatcmpl-DQeuwqU8iYEn0jmW2dXui8MlYo5sn",
            "object": "chat.completion",
            "created": 1775246186,
            "model": "gpt-4o-mini-2024-07-18",
            "choices": [
                {
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "tool_calls": [
                            {
                                "id": "call_yUjsmpi4tVtBJrk9lDCaeHXm",
                                "type": "function",
                                "function": {
                                    "name": "correctness",
                                    "arguments": (
                                        '{"label":"correct","explanation":"The output correctly '
                                        'states that 2 + 2 equals 4."}'
                                    ),
                                },
                            }
                        ],
                        "annotations": [],
                    },
                    "finish_reason": "tool_calls",
                }
            ],
            "usage": {
                "prompt_tokens": 98,
                "completion_tokens": 32,
                "total_tokens": 130,
                "prompt_tokens_details": {
                    "cached_tokens": 0,
                    "audio_tokens": 0,
                },
                "completion_tokens_details": {
                    "reasoning_tokens": 0,
                    "audio_tokens": 0,
                    "accepted_prediction_tokens": 0,
                    "rejected_prediction_tokens": 0,
                },
            },
            "service_tier": "default",
            "system_fingerprint": "fp_ebf4e532f9",
        }
        assert attributes.pop(f"{LLM_OUTPUT_MESSAGES}.0.{MESSAGE_ROLE}") == "assistant"
        assert isinstance(
            attributes.pop(
                f"{LLM_OUTPUT_MESSAGES}.0.{MessageAttributes.MESSAGE_TOOL_CALLS}.0.{TOOL_CALL_ID}"
            ),
            str,
        )
        assert (
            attributes.pop(
                f"{LLM_OUTPUT_MESSAGES}.0.{MessageAttributes.MESSAGE_TOOL_CALLS}.0.{TOOL_CALL_FUNCTION_NAME}"
            )
            == "correctness"
        )
        raw_arguments = attributes.pop(
            f"{LLM_OUTPUT_MESSAGES}.0.{MessageAttributes.MESSAGE_TOOL_CALLS}.0.{TOOL_CALL_FUNCTION_ARGUMENTS}"
        )
        assert isinstance(raw_arguments, str)
        arguments = json.loads(raw_arguments)
        assert arguments.pop("label") == "correct"
        assert isinstance(arguments.pop("explanation"), str)
        assert not arguments
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        input_value = json.loads(attributes.pop(INPUT_VALUE))
        assert input_value.pop("messages") == [
            {
                "role": "system",
                "content": "You are an evaluator. Assess whether the output correctly answers the input question.",
            },
            {
                "role": "user",
                "content": "Input: What is 2 + 2?\n\nOutput: 4\n\nIs this output correct?",
            },
        ]
        assert isinstance(invocation_parameters := attributes.pop("llm.invocation_parameters"), str)
        assert json.loads(invocation_parameters) == {
            "temperature": 0.0,
            "tool_choice": "required",
        }
        expected_tool = {
            "type": "function",
            "function": {
                "name": "correctness",
                "description": "Evaluates the correctness of the output",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "label": {
                            "type": "string",
                            "enum": ["correct", "incorrect"],
                            "description": "correctness",
                        },
                        "explanation": {
                            "type": "string",
                            "description": "Brief explanation for the label",
                        },
                    },
                    "required": ["label", "explanation"],
                },
                "strict": None,
            },
        }
        assert isinstance(tool_schtemas := attributes.pop("llm.tools.0.tool.json_schema"), str)
        assert json.loads(tool_schtemas) == expected_tool
        assert not attributes

        # Parse Eval Result span
        assert parse_eval_result_span.name == "Parse Eval Result"
        assert parse_eval_result_span.status_code == "OK"
        assert not parse_eval_result_span.events
        attributes = dict(flatten(parse_eval_result_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "CHAIN"
        raw_input_value = attributes.pop(INPUT_VALUE)
        assert raw_input_value is not None
        input_value = json.loads(raw_input_value)
        tool_calls = input_value.pop("tool_calls")
        assert len(tool_calls) == 1
        tool_call = next(
            iter(tool_calls.values())
        )  # the key is a random tool call ID from the LLM response
        assert tool_call == {
            "name": "correctness",
            "arguments": '{"label":"correct","explanation":"The output correctly states that 2 + 2 equals 4."}',
        }
        assert input_value == {
            "output_configs": {
                "correctness": {
                    "values": [
                        {"label": "correct", "score": 1.0},
                        {"label": "incorrect", "score": 0.0},
                    ],
                }
            },
        }
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert json.loads(attributes.pop(OUTPUT_VALUE)) == {
            "results": [
                {
                    "name": "correctness",
                    "label": "correct",
                    "score": 1.0,
                    "explanation": "The output correctly states that 2 + 2 equals 4.",
                }
            ],
        }
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # Verify span costs for LLM span
        assert len(span_costs) == 1
        span_cost = span_costs[0]

        assert span_cost.span_rowid == llm_span.id
        assert span_cost.trace_rowid == llm_span.trace_rowid
        assert span_cost.model_id == gpt_4o_mini_generative_model.id
        assert span_cost.span_start_time == llm_span.start_time
        prompt_token_prices = next(
            p for p in gpt_4o_mini_generative_model.token_prices if p.is_prompt
        )
        completion_token_prices = next(
            p for p in gpt_4o_mini_generative_model.token_prices if not p.is_prompt
        )
        prompt_base_rate = prompt_token_prices.base_rate
        completion_base_rate = completion_token_prices.base_rate
        expected_prompt_cost = llm_span.llm_token_count_prompt * prompt_base_rate
        expected_completion_cost = llm_span.llm_token_count_completion * completion_base_rate
        expected_total_cost = expected_prompt_cost + expected_completion_cost
        assert span_cost.total_cost == pytest.approx(expected_total_cost)
        assert span_cost.total_tokens == llm_span.llm_token_count_total
        assert span_cost.prompt_tokens == llm_span.llm_token_count_prompt
        assert span_cost.prompt_cost == pytest.approx(expected_prompt_cost)
        assert span_cost.completion_tokens == llm_span.llm_token_count_completion
        assert span_cost.completion_cost == pytest.approx(expected_completion_cost)

        # Verify span cost details
        assert len(span_cost.span_cost_details) >= 2
        input_detail = next(
            d for d in span_cost.span_cost_details if d.is_prompt and d.token_type == "input"
        )
        output_detail = next(
            d for d in span_cost.span_cost_details if not d.is_prompt and d.token_type == "output"
        )

        assert input_detail.span_cost_id == span_cost.id
        assert input_detail.token_type == "input"
        assert input_detail.is_prompt is True
        assert input_detail.tokens == llm_span.llm_token_count_prompt
        assert input_detail.cost == pytest.approx(expected_prompt_cost)
        assert input_detail.cost_per_token == prompt_base_rate

        assert output_detail.span_cost_id == span_cost.id
        assert output_detail.token_type == "output"
        assert output_detail.is_prompt is False
        assert output_detail.tokens == llm_span.llm_token_count_completion
        assert output_detail.cost == pytest.approx(expected_completion_cost)
        assert output_detail.cost_per_token == completion_base_rate

    async def test_evaluate_with_invalid_jsonpath_rejected_at_construction(
        self,
        db: DbSessionFactory,
        project: models.Project,
        tracer: Tracer,
        llm_evaluator: LLMEvaluator,
        output_config: CategoricalOutputConfig,
    ) -> None:
        # Invalid JSONPath expressions are now validated at construction time
        with pytest.raises(BadRequest) as exc_info:
            EvaluatorInputMappingInput(
                path_mapping=JSON({"output": "[[[invalid jsonpath"}),  # invalid JSONPath syntax
                literal_mapping=JSON({}),
            )
        assert "Invalid JSONPath expression for key 'output'" in str(exc_info.value)

    async def test_evaluate_with_nonexistent_path_records_error_on_template_span(
        self,
        db: DbSessionFactory,
        project: models.Project,
        tracer: Tracer,
        llm_evaluator: LLMEvaluator,
        output_config: CategoricalOutputConfig,
    ) -> None:
        # Valid JSONPath syntax but path doesn't exist in context
        input_mapping = EvaluatorInputMappingInput(
            path_mapping=JSON({"output": "nonexistent.path"}),
            literal_mapping=JSON({}),
        )
        evaluation_result = (
            await llm_evaluator.evaluate(
                context={"input": "What is 2 + 2?", "output": "4"},
                input_mapping=input_mapping.to_orm(),
                name="correctness",
                output_configs=[output_config],
                tracer=tracer,
            )
        )[0]

        result = dict(evaluation_result)
        error = result.pop("error")
        assert isinstance(error, str)
        assert result.pop("label") is None
        assert result.pop("score") is None
        assert result.pop("explanation") is None
        assert result.pop("annotator_kind") == "LLM"
        assert result.pop("name") == "correctness"
        trace_id = result.pop("trace_id")
        assert isinstance(trace_id, str)
        assert isinstance(result.pop("start_time"), datetime)
        assert isinstance(result.pop("end_time"), datetime)
        assert result.pop("metadata") == {}
        result.pop("error_exc", None)
        assert not result

        # Check spans
        async with db() as session:
            db_traces = tracer.get_db_traces(project_id=project.id)
            session.add_all(db_traces)
            await session.flush()
        assert len(db_traces) == 1
        db_trace = db_traces[0]
        db_spans = db_trace.spans
        db_span_costs = db_trace.span_costs
        assert len(db_spans) == 2  # Only EVALUATOR and Input Mapping (CHAIN) (no Prompt or LLM)
        assert len(db_span_costs) == 0  # no span costs since there's no LLM span

        evaluator_span = None
        input_mapping_span = None
        for span in db_spans:
            if span.span_kind == "EVALUATOR":
                evaluator_span = span
            elif span.span_kind == "CHAIN":
                input_mapping_span = span

        assert evaluator_span is not None
        assert input_mapping_span is not None
        assert evaluator_span.parent_id is None
        assert input_mapping_span.parent_id == evaluator_span.span_id

        # Verify evaluator span has error
        assert evaluator_span.status_code == "ERROR"
        assert len(evaluator_span.events) == 1
        exception_event = dict(evaluator_span.events[0])
        assert exception_event.pop("name") == "exception"
        assert isinstance(exception_event.pop("timestamp"), str)
        event_attributes = dict(exception_event.pop("attributes"))
        assert event_attributes.pop("exception.type") == "ValueError"
        assert event_attributes.pop("exception.escaped") == "False"
        assert "Traceback" in event_attributes.pop("exception.stacktrace")
        event_attributes.pop("exception.message")  # Don't assert on specific message
        assert not event_attributes
        assert not exception_event

        attributes = dict(flatten(evaluator_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "EVALUATOR"
        assert json.loads(attributes.pop(INPUT_VALUE)) == {"input": "What is 2 + 2?", "output": "4"}
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # Verify Input Mapping span has error
        assert input_mapping_span.status_code == "ERROR"
        assert len(input_mapping_span.events) == 1
        exception_event = dict(input_mapping_span.events[0])
        assert exception_event.pop("name") == "exception"
        assert isinstance(exception_event.pop("timestamp"), str)
        event_attributes = dict(exception_event.pop("attributes"))
        assert event_attributes.pop("exception.type") == "ValueError"
        assert "did not match any values" in event_attributes.pop("exception.message")
        assert event_attributes.pop("exception.escaped") == "False"
        assert "Traceback" in event_attributes.pop("exception.stacktrace")
        assert not event_attributes
        assert not exception_event

        attributes = dict(flatten(input_mapping_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "CHAIN"
        assert json.loads(attributes.pop(INPUT_VALUE)) == {
            "input_mapping": {
                "path_mapping": {"output": "nonexistent.path"},
                "literal_mapping": {},
            },
            "template_variables": {"input": "What is 2 + 2?", "output": "4"},
        }
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert not attributes

    async def test_evaluate_with_invalid_api_key_returns_error(
        self,
        db: DbSessionFactory,
        project: models.Project,
        tracer: Tracer,
        llm_evaluator: LLMEvaluator,
        output_config: CategoricalOutputConfig,
        input_mapping: EvaluatorInputMappingInput,
        custom_vcr: CustomVCR,
    ) -> None:
        with custom_vcr.use_cassette():
            evaluation_result = (
                await llm_evaluator.evaluate(
                    context={"input": "What is 2 + 2?", "output": "4"},
                    input_mapping=input_mapping.to_orm(),
                    name="correctness",
                    output_configs=[output_config],
                    tracer=tracer,
                )
            )[0]

        result = dict(evaluation_result)
        error = result.pop("error")
        assert isinstance(error, str)
        assert "401" in error
        assert "invalid_api_key" in error
        assert result.pop("label") is None
        assert result.pop("score") is None
        assert result.pop("explanation") is None
        assert result.pop("annotator_kind") == "LLM"
        assert result.pop("name") == "correctness"
        trace_id = result.pop("trace_id")
        assert isinstance(trace_id, str)
        assert isinstance(result.pop("start_time"), datetime)
        assert isinstance(result.pop("end_time"), datetime)
        assert result.pop("metadata") == {}
        result.pop("error_exc", None)
        assert not result

        async with db() as session:
            db_traces = tracer.get_db_traces(project_id=project.id)
            session.add_all(db_traces)
            await session.flush()

        assert len(db_traces) == 1
        db_trace = db_traces[0]
        db_spans = db_trace.spans
        db_span_costs = db_trace.span_costs
        assert trace_id == db_trace.trace_id

        assert (
            len(db_spans) == 4
        )  # EVALUATOR, Input Mapping (CHAIN), Prompt (PROMPT), LLM (no Parse Eval Result due to error)

        evaluator_span = None
        input_mapping_span = None
        prompt_span = None
        llm_span = None
        for span in db_spans:
            if span.span_kind == "EVALUATOR":
                evaluator_span = span
            elif span.span_kind == "LLM":
                llm_span = span
            elif span.span_kind == "PROMPT":
                prompt_span = span
            elif span.span_kind == "CHAIN":
                if span.name == "Input Mapping":
                    input_mapping_span = span

        assert evaluator_span is not None
        assert input_mapping_span is not None
        assert prompt_span is not None
        assert llm_span is not None
        assert evaluator_span.parent_id is None
        assert input_mapping_span.parent_id == evaluator_span.span_id
        assert prompt_span.parent_id == evaluator_span.span_id
        assert llm_span.parent_id == evaluator_span.span_id

        assert len(db_span_costs) == 0  # LLM span has no token counts due to API error

        # Input Mapping span (CHAIN)
        assert input_mapping_span.name == "Input Mapping"
        assert input_mapping_span.status_code == "OK"
        assert not input_mapping_span.events
        attributes = dict(flatten(input_mapping_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "CHAIN"
        assert json.loads(attributes.pop(INPUT_VALUE)) == {
            "input_mapping": {"path_mapping": {}, "literal_mapping": {}},
            "template_variables": {"input": "What is 2 + 2?", "output": "4"},
        }
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert json.loads(attributes.pop(OUTPUT_VALUE)) == {
            "input": "What is 2 + 2?",
            "output": "4",
        }
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # Prompt span
        assert prompt_span.name == "Prompt: test-prompt"
        assert prompt_span.status_code == "OK"
        assert not prompt_span.events
        attributes = dict(flatten(prompt_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "PROMPT"
        assert json.loads(attributes.pop(INPUT_VALUE)) == {
            "input": "What is 2 + 2?",
            "output": "4",
        }
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert json.loads(attributes.pop(OUTPUT_VALUE)) == {
            "messages": [
                {
                    "role": "system",
                    "content": "You are an evaluator. Assess whether the output correctly answers the input question.",
                },
                {
                    "role": "user",
                    "content": "Input: What is 2 + 2?\n\nOutput: 4\n\nIs this output correct?",
                },
            ]
        }
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # evaluator span
        assert evaluator_span.name == "Evaluator: correctness"
        assert evaluator_span.status_code == "ERROR"
        assert "401" in evaluator_span.status_message
        assert "invalid_api_key" in evaluator_span.status_message
        assert len(evaluator_span.events) == 1
        exception_event = dict(evaluator_span.events[0])
        assert exception_event.pop("name") == "exception"
        assert isinstance(exception_event.pop("timestamp"), str)
        event_attributes = dict(exception_event.pop("attributes"))
        assert event_attributes.pop("exception.type") == "openai.AuthenticationError"
        exception_message = event_attributes.pop("exception.message")
        assert "401" in exception_message
        assert "invalid_api_key" in exception_message
        assert event_attributes.pop("exception.escaped") == "False"
        assert "Traceback" in event_attributes.pop("exception.stacktrace")
        assert not event_attributes
        assert not exception_event
        attributes = dict(flatten(evaluator_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "EVALUATOR"
        assert json.loads(attributes.pop(INPUT_VALUE)) == {"input": "What is 2 + 2?", "output": "4"}
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # llm span
        assert llm_span.name == "ChatCompletion"
        assert llm_span.status_code == "ERROR"
        assert len(llm_span.events) == 1
        exception_event = dict(llm_span.events[0])
        assert exception_event.pop("name") == "exception"
        assert isinstance(exception_event.pop("timestamp"), str)
        event_attributes = dict(exception_event.pop("attributes"))
        assert event_attributes.pop("exception.type") == "openai.AuthenticationError"
        exception_message = event_attributes.pop("exception.message")
        assert "401" in exception_message
        assert "invalid_api_key" in exception_message
        assert event_attributes.pop("exception.escaped") == "False"
        assert "Traceback" in event_attributes.pop("exception.stacktrace")
        assert not event_attributes
        assert not exception_event
        attributes = dict(flatten(llm_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "LLM"
        assert attributes.pop(LLM_MODEL_NAME) == "gpt-4o-mini"
        assert attributes.pop(LLM_PROVIDER) == "openai"
        assert attributes.pop(LLM_SYSTEM) == "openai"
        assert attributes.pop(f"{LLM_INPUT_MESSAGES}.0.{MESSAGE_ROLE}") == "system"
        assert "evaluator" in attributes.pop(f"{LLM_INPUT_MESSAGES}.0.{MESSAGE_CONTENT}").lower()
        assert attributes.pop(f"{LLM_INPUT_MESSAGES}.1.{MESSAGE_ROLE}") == "user"
        user_message = attributes.pop(f"{LLM_INPUT_MESSAGES}.1.{MESSAGE_CONTENT}")
        assert "What is 2 + 2?" in user_message
        assert "4" in user_message
        assert attributes.pop(URL_FULL) == "https://api.openai.com/v1/chat/completions"
        assert attributes.pop(URL_PATH) == "chat/completions"
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        input_value = json.loads(attributes.pop(INPUT_VALUE))
        assert input_value.pop("messages") == [
            {
                "role": "system",
                "content": "You are an evaluator. Assess whether the output correctly answers the input question.",
            },
            {
                "role": "user",
                "content": "Input: What is 2 + 2?\n\nOutput: 4\n\nIs this output correct?",
            },
        ]
        assert isinstance(invocation_parameters := attributes.pop("llm.invocation_parameters"), str)
        assert json.loads(invocation_parameters) == {
            "temperature": 0.0,
            "tool_choice": "required",
        }
        expected_tool = {
            "type": "function",
            "function": {
                "name": "correctness",
                "description": "Evaluates the correctness of the output",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "label": {
                            "type": "string",
                            "enum": ["correct", "incorrect"],
                            "description": "correctness",
                        },
                        "explanation": {
                            "type": "string",
                            "description": "Brief explanation for the label",
                        },
                    },
                    "required": ["label", "explanation"],
                },
                "strict": None,
            },
        }
        assert isinstance(tool_schtemas := attributes.pop("llm.tools.0.tool.json_schema"), str)
        assert json.loads(tool_schtemas) == expected_tool
        assert not attributes

    async def test_evaluate_with_no_tool_calls_records_error_on_chain_span(
        self,
        db: DbSessionFactory,
        project: models.Project,
        tracer: Tracer,
        llm_evaluator: LLMEvaluator,
        output_config: CategoricalOutputConfig,
        input_mapping: EvaluatorInputMappingInput,
    ) -> None:
        text_response_body = json.dumps(
            {
                "id": "chatcmpl-mock",
                "object": "chat.completion",
                "created": 1700000000,
                "model": "gpt-4o-mini",
                "choices": [
                    {
                        "index": 0,
                        "message": {
                            "role": "assistant",
                            "content": "I cannot evaluate this.",
                        },
                        "finish_reason": "stop",
                    }
                ],
                "usage": {
                    "prompt_tokens": 10,
                    "completion_tokens": 5,
                    "total_tokens": 15,
                },
            }
        )
        with respx.mock:
            respx.post("https://api.openai.com/v1/chat/completions").mock(
                return_value=httpx.Response(
                    200,
                    content=text_response_body,
                    headers={"content-type": "application/json"},
                )
            )
            evaluation_results = await llm_evaluator.evaluate(
                context={"input": "What is 2 + 2?", "output": "4"},
                input_mapping=input_mapping.to_orm(),
                name="correctness",
                output_configs=[output_config],
                tracer=tracer,
            )
            assert len(evaluation_results) == 1
            evaluation_result = evaluation_results[0]

        # Check evaluation result
        result = dict(evaluation_result)
        error = result.pop("error")
        assert isinstance(error, str)
        assert "No tool calls received from LLM" in error
        assert result.pop("label") is None
        assert result.pop("score") is None
        assert result.pop("explanation") is None
        assert result.pop("annotator_kind") == "LLM"
        assert result.pop("name") == "correctness"
        trace_id = result.pop("trace_id")
        assert isinstance(trace_id, str)
        assert isinstance(result.pop("start_time"), datetime)
        assert isinstance(result.pop("end_time"), datetime)
        assert result.pop("metadata") == {}
        result.pop("error_exc", None)
        assert not result

        # Check spans
        async with db() as session:
            db_traces = tracer.get_db_traces(project_id=project.id)
            session.add_all(db_traces)
            await session.flush()

        assert len(db_traces) == 1
        db_trace = db_traces[0]
        db_spans = db_trace.spans
        assert (
            len(db_spans) == 5
        )  # EVALUATOR, Input Mapping (CHAIN), Prompt (PROMPT), LLM, Parse Eval Result (CHAIN) with error

        evaluator_span = None
        input_mapping_span = None
        prompt_span = None
        llm_span = None
        parse_eval_result_span = None
        for span in db_spans:
            if span.span_kind == "EVALUATOR":
                evaluator_span = span
            elif span.span_kind == "LLM":
                llm_span = span
            elif span.span_kind == "PROMPT":
                prompt_span = span
            elif span.span_kind == "CHAIN":
                if span.name == "Input Mapping":
                    input_mapping_span = span
                elif span.name == "Parse Eval Result":
                    parse_eval_result_span = span

        assert evaluator_span is not None
        assert input_mapping_span is not None
        assert prompt_span is not None
        assert llm_span is not None
        assert parse_eval_result_span is not None

        # Verify parent-child relationships
        assert evaluator_span.parent_id is None
        assert input_mapping_span.parent_id == evaluator_span.span_id
        assert prompt_span.parent_id == evaluator_span.span_id
        assert llm_span.parent_id == evaluator_span.span_id
        assert parse_eval_result_span.parent_id == evaluator_span.span_id

        # Input Mapping span
        assert input_mapping_span.name == "Input Mapping"
        assert input_mapping_span.status_code == "OK"
        assert not input_mapping_span.events

        # Prompt span
        assert prompt_span.name == "Prompt: test-prompt"
        assert prompt_span.status_code == "OK"
        assert not prompt_span.events

        # LLM span
        assert llm_span.name == "ChatCompletion"
        assert llm_span.status_code == "OK"
        assert not llm_span.events

        # Parse Eval Result span (has error)
        assert parse_eval_result_span.name == "Parse Eval Result"
        assert parse_eval_result_span.status_code == "ERROR"
        assert "No tool calls received from LLM" in parse_eval_result_span.status_message
        assert len(parse_eval_result_span.events) == 1
        exception_event = dict(parse_eval_result_span.events[0])
        assert exception_event.pop("name") == "exception"
        assert isinstance(exception_event.pop("timestamp"), str)
        event_attributes = dict(exception_event.pop("attributes"))
        assert event_attributes.pop("exception.type") == "ValueError"
        assert "No tool calls received from LLM" in event_attributes.pop("exception.message")
        assert event_attributes.pop("exception.escaped") == "False"
        assert "Traceback" in event_attributes.pop("exception.stacktrace")
        assert not event_attributes
        assert not exception_event
        attributes = dict(flatten(parse_eval_result_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "CHAIN"
        assert json.loads(attributes.pop(INPUT_VALUE)) == {
            "tool_calls": {},
            "output_configs": {
                "correctness": {
                    "values": [
                        {"label": "correct", "score": 1.0},
                        {"label": "incorrect", "score": 0.0},
                    ]
                }
            },
        }
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert not attributes

        # evaluator span
        assert evaluator_span.name == "Evaluator: correctness"
        assert evaluator_span.status_code == "ERROR"
        assert "No tool calls received from LLM" in evaluator_span.status_message
        assert len(evaluator_span.events) == 1
        exception_event = dict(evaluator_span.events[0])
        assert exception_event.pop("name") == "exception"
        assert isinstance(exception_event.pop("timestamp"), str)
        event_attributes = dict(exception_event.pop("attributes"))
        assert event_attributes.pop("exception.type") == "ValueError"
        assert "No tool calls received from LLM" in event_attributes.pop("exception.message")
        assert event_attributes.pop("exception.escaped") == "False"
        assert "Traceback" in event_attributes.pop("exception.stacktrace")
        assert not event_attributes
        assert not exception_event
        attributes = dict(flatten(evaluator_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "EVALUATOR"
        assert json.loads(attributes.pop(INPUT_VALUE)) == {"input": "What is 2 + 2?", "output": "4"}
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        assert not attributes

    async def test_evaluate_with_missing_label_in_tool_call_records_error(
        self,
        db: DbSessionFactory,
        project: models.Project,
        tracer: Tracer,
        llm_evaluator: LLMEvaluator,
        output_config: CategoricalOutputConfig,
        input_mapping: EvaluatorInputMappingInput,
    ) -> None:
        tool_call_response_body = json.dumps(
            {
                "id": "chatcmpl-mock",
                "object": "chat.completion",
                "created": 1700000000,
                "model": "gpt-4o-mini",
                "choices": [
                    {
                        "index": 0,
                        "message": {
                            "role": "assistant",
                            "content": None,
                            "tool_calls": [
                                {
                                    "id": "call_abc123",
                                    "type": "function",
                                    "function": {
                                        "name": "correctness",
                                        "arguments": '{"explanation": "The output is correct."}',
                                    },
                                }
                            ],
                        },
                        "finish_reason": "tool_calls",
                    }
                ],
                "usage": {
                    "prompt_tokens": 10,
                    "completion_tokens": 20,
                    "total_tokens": 30,
                },
            }
        )
        with respx.mock:
            respx.post("https://api.openai.com/v1/chat/completions").mock(
                return_value=httpx.Response(
                    200,
                    content=tool_call_response_body,
                    headers={"content-type": "application/json"},
                )
            )
            evaluation_results = await llm_evaluator.evaluate(
                context={"input": "What is 2 + 2?", "output": "4"},
                input_mapping=input_mapping.to_orm(),
                name="correctness",
                output_configs=[output_config],
                tracer=tracer,
            )
            assert len(evaluation_results) == 1
            evaluation_result = evaluation_results[0]

        result = dict(evaluation_result)
        error = result.pop("error")
        assert isinstance(error, str)
        assert "LLM response missing required 'label' field in tool call arguments" in error
        assert result.pop("label") is None
        assert result.pop("score") is None
        assert result.pop("explanation") is None
        assert result.pop("annotator_kind") == "LLM"
        assert result.pop("name") == "correctness"
        trace_id = result.pop("trace_id")
        assert isinstance(trace_id, str)
        assert isinstance(result.pop("start_time"), datetime)
        assert isinstance(result.pop("end_time"), datetime)
        assert result.pop("metadata") == {}
        result.pop("error_exc", None)
        assert not result

    async def test_evaluate_with_multipart_template(
        self,
        db: DbSessionFactory,
        project: models.Project,
        tracer: Tracer,
        output_config: CategoricalOutputConfig,
        input_mapping: EvaluatorInputMappingInput,
        multipart_llm_evaluator: LLMEvaluator,
        custom_vcr: CustomVCR,
        gpt_4o_mini_generative_model: models.GenerativeModel,
    ) -> None:
        with custom_vcr.use_cassette():
            evaluation_result = (
                await multipart_llm_evaluator.evaluate(
                    context={"input": "What is 2 + 2?", "output": "4"},
                    input_mapping=input_mapping.to_orm(),
                    name="correctness",
                    output_configs=[output_config],
                    tracer=tracer,
                )
            )[0]

        result = dict(evaluation_result)
        assert result.pop("error") is None
        assert result.pop("label") == "correct"
        assert result.pop("score") == 1.0
        assert result.pop("explanation") is not None
        assert result.pop("annotator_kind") == "LLM"
        assert result.pop("name") == "correctness"
        trace_id = result.pop("trace_id")
        assert isinstance(trace_id, str)
        assert isinstance(result.pop("start_time"), datetime)
        assert isinstance(result.pop("end_time"), datetime)
        assert result.pop("metadata") == {}
        result.pop("error_exc", None)
        assert not result

        async with db() as session:
            db_traces = tracer.get_db_traces(project_id=project.id)
            session.add_all(db_traces)
            await session.flush()

        assert len(db_traces) == 1
        db_trace = db_traces[0]
        assert db_trace.trace_id == trace_id
        db_spans = db_trace.spans

        prompt_span = None
        for span in db_spans:
            if span.span_kind == "PROMPT" and span.name.startswith("Prompt:"):
                prompt_span = span
                break

        assert prompt_span is not None
        assert prompt_span.name == "Prompt: test-prompt"
        attributes = dict(flatten(prompt_span.attributes, recurse_on_sequence=True))
        assert attributes.pop(OPENINFERENCE_SPAN_KIND) == "PROMPT"
        input_value = json.loads(attributes.pop(INPUT_VALUE))
        assert input_value == {"input": "What is 2 + 2?", "output": "4"}
        assert attributes.pop(INPUT_MIME_TYPE) == "application/json"
        output_value = json.loads(attributes.pop(OUTPUT_VALUE))
        assert output_value == {
            "messages": [
                {
                    "role": "system",
                    "content": "You are an evaluator. Assess whether the output correctly answers the input question.",
                },
                {
                    "role": "user",
                    "content": "Input: What is 2 + 2?\n\nOutput: 4\n\nIs this output correct?",
                },
            ]
        }
        assert attributes.pop(OUTPUT_MIME_TYPE) == "application/json"
        assert not attributes


class TestGetEvaluators:
    async def test_returns_evaluators_in_input_order_with_interspersed_types(
        self,
        db: Any,
        correctness_llm_evaluator: models.LLMEvaluator,
        openai_api_key: str,
        synced_builtin_evaluators: None,
    ) -> None:
        from sqlalchemy import select

        # Look up builtin evaluator IDs from the database by key
        async with db() as session:
            contains_id = await session.scalar(
                select(models.BuiltinEvaluator.id).where(models.BuiltinEvaluator.key == "contains")
            )
            exact_match_id = await session.scalar(
                select(models.BuiltinEvaluator.id).where(
                    models.BuiltinEvaluator.key == "exact_match"
                )
            )
            regex_id = await session.scalar(
                select(models.BuiltinEvaluator.id).where(models.BuiltinEvaluator.key == "regex")
            )
        assert contains_id is not None
        assert exact_match_id is not None
        assert regex_id is not None

        # Create a dataset and DatasetEvaluators
        async with db() as session:
            dataset = models.Dataset(name="test-dataset", metadata_={})
            session.add(dataset)
            await session.flush()

            # Create DatasetEvaluators for each evaluator type
            de_contains = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=contains_id,
                name=Identifier("contains-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="contains-project", description=""),
            )
            de_llm = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=correctness_llm_evaluator.id,
                name=Identifier("llm-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="llm-project", description=""),
            )
            de_exact_match = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=exact_match_id,
                name=Identifier("exact-match-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="exact-match-project", description=""),
            )
            de_regex = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=regex_id,
                name=Identifier("regex-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="regex-project", description=""),
            )
            session.add_all([de_contains, de_llm, de_exact_match, de_regex])
            await session.flush()

            input_ids = [de_contains.id, de_llm.id, de_exact_match.id, de_regex.id]

            evaluators = await get_evaluators(
                dataset_evaluator_ids=input_ids,
                session=session,
                decrypt=lambda x: x,
                experiment_id=0,
                credentials=None,
                sandbox_session_manager=SandboxSessionManager(),
            )

        assert len(evaluators) == 4
        assert isinstance(evaluators[0], BuiltInEvaluator)
        assert evaluators[0].name == "contains"
        assert isinstance(evaluators[1], LLMEvaluatorClass)
        assert evaluators[1].name == correctness_llm_evaluator.name.root
        assert isinstance(evaluators[2], BuiltInEvaluator)
        assert evaluators[2].name == "exact_match"
        assert isinstance(evaluators[3], BuiltInEvaluator)
        assert evaluators[3].name == "regex"

    async def test_raises_value_error_for_missing_dataset_evaluator(
        self,
        db: Any,
    ) -> None:
        non_existent_id = 999999
        input_ids = [non_existent_id]

        async with db() as session:
            with pytest.raises(NotFound, match="DatasetEvaluator.*not found"):
                await get_evaluators(
                    dataset_evaluator_ids=input_ids,
                    session=session,
                    decrypt=lambda x: x,
                    experiment_id=0,
                    credentials=None,
                    sandbox_session_manager=SandboxSessionManager(),
                )

    async def test_returns_empty_list_for_empty_input(
        self,
        db: Any,
    ) -> None:
        async with db() as session:
            evaluators = await get_evaluators(
                dataset_evaluator_ids=[],
                session=session,
                decrypt=lambda x: x,
                experiment_id=0,
                credentials=None,
                sandbox_session_manager=SandboxSessionManager(),
            )

        assert evaluators == []

    async def test_code_evaluator_not_found_message_contains_name_and_settings_hint(
        self,
        db: Any,
        seed_languages: None,
    ) -> None:
        async with db() as session:
            dataset = models.Dataset(name="test-dataset-code-no-sandbox", metadata_={})
            session.add(dataset)
            await session.flush()

            code_eval = models.CodeEvaluator(
                name=Identifier("my-code-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                output_configs=[],
                language="PYTHON",
                sandbox_config_id=None,
            )
            session.add(code_eval)
            await session.flush()
            python_lang = await session.scalar(
                select(models.Language).where(models.Language.name == "PYTHON")
            )
            if python_lang is None:
                python_lang = models.Language(name="PYTHON")
                session.add(python_lang)
                await session.flush()
            version = models.CodeEvaluatorVersion(
                code_evaluator_id=code_eval.id,
                source_code="def evaluate(output): return 1.0",
            )
            session.add(version)
            await session.flush()

            de_code = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=code_eval.id,
                name=Identifier("my-code-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="code-eval-no-sandbox-project", description=""),
            )
            session.add(de_code)
            await session.flush()

            with pytest.raises(NotFound) as exc_info:
                await get_evaluators(
                    dataset_evaluator_ids=[de_code.id],
                    session=session,
                    decrypt=lambda x: x,
                    experiment_id=0,
                    credentials=None,
                    sandbox_session_manager=SandboxSessionManager(),
                )

        assert "my-code-eval" in str(exc_info.value)
        assert "/settings/sandboxes" in str(exc_info.value)

    async def test_code_evaluator_disabled_sandbox_config_raises_bad_request(
        self,
        db: Any,
        seed_sandbox_providers: None,
    ) -> None:
        async with db() as session:
            provider = await session.scalar(
                select(models.SandboxProvider).where(models.SandboxProvider.backend_type == "WASM")
            )
            assert provider is not None
            sandbox_config = models.SandboxConfig(
                backend_type=provider.backend_type,
                language="PYTHON",
                name=Identifier("disabled-runtime-config"),
                config={},
                timeout=30,
                enabled=False,
            )
            dataset = models.Dataset(name="test-disabled-runtime-config", metadata_={})
            session.add_all([sandbox_config, dataset])
            await session.flush()
            code_eval = models.CodeEvaluator(
                name=Identifier("disabled-runtime-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                output_configs=[],
                language="PYTHON",
                sandbox_config_id=sandbox_config.id,
            )
            session.add(code_eval)
            await session.flush()
            session.add(
                models.CodeEvaluatorVersion(
                    code_evaluator_id=code_eval.id,
                    source_code="def evaluate(input): return {'score': 1.0}",
                )
            )
            dataset_evaluator = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=code_eval.id,
                name=Identifier("disabled-runtime-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="disabled-runtime-project", description=""),
            )
            session.add(dataset_evaluator)
            await session.flush()

            with pytest.raises(
                BadRequest,
                match="Sandbox configuration 'disabled-runtime-config' is disabled",
            ):
                await get_evaluators(
                    dataset_evaluator_ids=[dataset_evaluator.id],
                    session=session,
                    decrypt=lambda x: x,
                    experiment_id=0,
                    credentials=None,
                    sandbox_session_manager=SandboxSessionManager(),
                )

    async def test_code_evaluator_reuses_backend_for_shared_sandbox_config(
        self,
        db: Any,
        seed_sandbox_providers: None,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        calls = 0

        async def fake_build_sandbox_backend(*args: Any, **kwargs: Any) -> object:
            nonlocal calls
            calls += 1
            return object()

        monkeypatch.setattr(
            "phoenix.server.api.evaluators.build_sandbox_backend",
            fake_build_sandbox_backend,
        )

        async with db() as session:
            provider = await session.scalar(
                select(models.SandboxProvider).where(models.SandboxProvider.backend_type == "WASM")
            )
            assert provider is not None
            sandbox_config = models.SandboxConfig(
                backend_type=provider.backend_type,
                language="PYTHON",
                name=Identifier("shared-runtime-config"),
                config={},
                timeout=30,
            )
            dataset = models.Dataset(name="test-shared-runtime-config", metadata_={})
            session.add_all([sandbox_config, dataset])
            await session.flush()

            dataset_evaluator_ids = []
            for index in range(2):
                code_eval = models.CodeEvaluator(
                    name=Identifier(f"shared-runtime-eval-{index}"),
                    input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                    output_configs=[],
                    language="PYTHON",
                    sandbox_config_id=sandbox_config.id,
                )
                session.add(code_eval)
                await session.flush()
                session.add(
                    models.CodeEvaluatorVersion(
                        code_evaluator_id=code_eval.id,
                        source_code="def evaluate(input): return {'score': 1.0}",
                    )
                )
                dataset_evaluator = models.DatasetEvaluators(
                    dataset_id=dataset.id,
                    evaluator_id=code_eval.id,
                    name=Identifier(f"shared-runtime-eval-{index}"),
                    input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                    project=models.Project(
                        name=f"shared-runtime-project-{index}",
                        description="",
                    ),
                )
                session.add(dataset_evaluator)
                await session.flush()
                dataset_evaluator_ids.append(dataset_evaluator.id)

            evaluators = await get_evaluators(
                dataset_evaluator_ids=dataset_evaluator_ids,
                session=session,
                decrypt=lambda x: x,
                experiment_id=0,
                credentials=None,
                sandbox_session_manager=SandboxSessionManager(),
            )

        assert len(evaluators) == 2
        assert calls == 1

    async def test_preserves_order_with_only_builtin_evaluators(
        self,
        db: Any,
        synced_builtin_evaluators: None,
    ) -> None:
        from sqlalchemy import select

        # Look up builtin evaluator IDs from the database by key
        async with db() as session:
            levenshtein_id = await session.scalar(
                select(models.BuiltinEvaluator.id).where(
                    models.BuiltinEvaluator.key == "levenshtein_distance"
                )
            )
            json_distance_id = await session.scalar(
                select(models.BuiltinEvaluator.id).where(
                    models.BuiltinEvaluator.key == "json_distance"
                )
            )
            contains_id = await session.scalar(
                select(models.BuiltinEvaluator.id).where(models.BuiltinEvaluator.key == "contains")
            )
        assert levenshtein_id is not None
        assert json_distance_id is not None
        assert contains_id is not None

        async with db() as session:
            dataset = models.Dataset(name="test-dataset-builtins", metadata_={})
            session.add(dataset)
            await session.flush()

            de_levenshtein = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=levenshtein_id,
                name=Identifier("levenshtein-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="levenshtein-project", description=""),
            )
            de_json_distance = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=json_distance_id,
                name=Identifier("json-distance-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="json-distance-project", description=""),
            )
            de_contains = models.DatasetEvaluators(
                dataset_id=dataset.id,
                evaluator_id=contains_id,
                name=Identifier("contains-eval"),
                input_mapping=InputMapping(literal_mapping={}, path_mapping={}),
                project=models.Project(name="contains-project", description=""),
            )
            session.add_all([de_levenshtein, de_json_distance, de_contains])
            await session.flush()

            input_ids = [de_levenshtein.id, de_json_distance.id, de_contains.id]

            evaluators = await get_evaluators(
                dataset_evaluator_ids=input_ids,
                session=session,
                decrypt=lambda x: x,
                experiment_id=0,
                credentials=None,
                sandbox_session_manager=SandboxSessionManager(),
            )

        assert len(evaluators) == 3
        assert all(isinstance(e, BuiltInEvaluator) for e in evaluators)
        assert evaluators[0].name == "levenshtein_distance"
        assert evaluators[1].name == "json_distance"
        assert evaluators[2].name == "contains"


@pytest.fixture
def output_config() -> CategoricalOutputConfig:
    return CategoricalOutputConfig(
        type="CATEGORICAL",
        name="correctness",
        optimization_direction=OptimizationDirection.MAXIMIZE,
        description="correctness evaluation",
        values=[
            CategoricalAnnotationValue(label="correct", score=1.0),
            CategoricalAnnotationValue(label="incorrect", score=0.0),
        ],
    )


@pytest.fixture
def prompt_version() -> models.PromptVersion:
    return models.PromptVersion(
        prompt_id=1,
        template_type=PromptTemplateType.CHAT,
        template_format=PromptTemplateFormat.MUSTACHE,
        template=PromptChatTemplate(
            type="chat",
            messages=[
                PromptMessage(
                    role="user",
                    content=[
                        TextContentPart(
                            type="text",
                            text="Evaluate the correctness: {{input}}",
                        )
                    ],
                )
            ],
        ),
        invocation_parameters=PromptOpenAIInvocationParameters(
            type="openai",
            openai=PromptOpenAIInvocationParametersContent(),
        ),
        tools=PromptTools(
            type="tools",
            tools=[
                PromptToolFunction(
                    type="function",
                    function=PromptToolFunctionDefinition(
                        name="correctness",
                        description=UNDEFINED,
                        parameters={
                            "type": "object",
                            "properties": {
                                "label": {
                                    "type": "string",
                                    "enum": ["correct", "incorrect"],
                                    "description": "correctness",
                                }
                            },
                            "required": ["label"],
                        },
                    ),
                )
            ],
            tool_choice=PromptToolChoiceOneOrMore(
                type="one_or_more",
            ),
        ),
        response_format=None,
        model_provider=ModelProvider.OPENAI,
        model_name="gpt-4",
        metadata_={},
    )


# message attributes
MESSAGE_CONTENT = MessageAttributes.MESSAGE_CONTENT
MESSAGE_CONTENTS = MessageAttributes.MESSAGE_CONTENTS
MESSAGE_ROLE = MessageAttributes.MESSAGE_ROLE

# span attributes
OPENINFERENCE_SPAN_KIND = SpanAttributes.OPENINFERENCE_SPAN_KIND
LLM_MODEL_NAME = SpanAttributes.LLM_MODEL_NAME
LLM_SYSTEM = SpanAttributes.LLM_SYSTEM
LLM_PROVIDER = SpanAttributes.LLM_PROVIDER
LLM_INPUT_MESSAGES = SpanAttributes.LLM_INPUT_MESSAGES
LLM_OUTPUT_MESSAGES = SpanAttributes.LLM_OUTPUT_MESSAGES
LLM_INVOCATION_PARAMETERS = SpanAttributes.LLM_INVOCATION_PARAMETERS
LLM_TOOLS = SpanAttributes.LLM_TOOLS
INPUT_VALUE = SpanAttributes.INPUT_VALUE
INPUT_MIME_TYPE = SpanAttributes.INPUT_MIME_TYPE
OUTPUT_VALUE = SpanAttributes.OUTPUT_VALUE
OUTPUT_MIME_TYPE = SpanAttributes.OUTPUT_MIME_TYPE

# tool attributes
TOOL_JSON_SCHEMA = ToolAttributes.TOOL_JSON_SCHEMA

# tool call attributes
TOOL_CALL_ID = ToolCallAttributes.TOOL_CALL_ID
TOOL_CALL_FUNCTION_NAME = ToolCallAttributes.TOOL_CALL_FUNCTION_NAME
TOOL_CALL_FUNCTION_ARGUMENTS = ToolCallAttributes.TOOL_CALL_FUNCTION_ARGUMENTS_JSON
