/deepeval/test_run/test_run.py

from enum import Enum
import os
import json
from pathlib import Path
from pydantic import BaseModel, Field
from typing import Any, Optional, List, Dict, Union, Tuple
import sys
from rich.table import Table
from rich.console import Console
from rich import print


from deepeval.metrics import BaseMetric
from deepeval.confident.api import Api, Endpoints, HttpMethods, is_confident
from deepeval.test_run.api import (
    LLMApiTestCase,
    ConversationalApiTestCase,
    TestRunHttpResponse,
    MetricData,
)
from deepeval.tracing.utils import make_json_serializable
from deepeval.tracing.api import SpanApiType, span_api_type_literals
from deepeval.test_case import LLMTestCase, ConversationalTestCase
from deepeval.utils import (
    delete_file_if_exists,
    get_is_running_deepeval,
    get_test_run_official,
    is_read_only_env,
    open_browser,
    shorten,
    format_turn,
    len_short,
)
from deepeval.test_run.cache import global_test_run_cache_manager
from deepeval.constants import CONFIDENT_TEST_CASE_BATCH_SIZE, HIDDEN_DIR
from deepeval.prompt import (
    PromptMessage,
    ModelSettings,
    PromptInterpolationType,
    OutputType,
)
from rich.panel import Panel
from rich.columns import Columns

portalocker = None
if not is_read_only_env():
    try:
        import portalocker
    except Exception as e:
        print(
            f"Warning: failed to import portalocker: {e}",
            file=sys.stderr,
        )
else:
    print(
        "Warning: DeepEval is configured for read only environment. Test runs will not be written to disk."
    )


TEMP_FILE_PATH = f"{HIDDEN_DIR}/.temp_test_run_data.json"
LATEST_TEST_RUN_FILE_PATH = f"{HIDDEN_DIR}/.latest_test_run.json"
# Full TestRun payload (same as timestamped exports); overwritten each run.
LATEST_FULL_TEST_RUN_FILE_PATH = f"{HIDDEN_DIR}/.latest_run_full.json"
LATEST_TEST_RUN_DATA_KEY = "testRunData"
LATEST_TEST_RUN_LINK_KEY = "testRunLink"
console = Console()


class TestRunResultDisplay(Enum):
    ALL = "all"
    FAILING = "failing"
    PASSING = "passing"


class MetricScoreType(BaseModel):
    metric: str
    score: float

    @classmethod
    def from_metric(cls, metric: BaseMetric):
        return cls(metric=metric.__name__, score=metric.score)


class MetricScores(BaseModel):
    metric: str
    scores: List[float]
    passes: int
    fails: int
    errors: int


class TraceMetricScores(BaseModel):
    agent: Dict[str, Dict[str, MetricScores]] = Field(default_factory=dict)
    tool: Dict[str, Dict[str, MetricScores]] = Field(default_factory=dict)
    retriever: Dict[str, Dict[str, MetricScores]] = Field(default_factory=dict)
    llm: Dict[str, Dict[str, MetricScores]] = Field(default_factory=dict)
    base: Dict[str, Dict[str, MetricScores]] = Field(default_factory=dict)


class PromptData(BaseModel):
    alias: Optional[str] = None
    hash: Optional[str] = None
    version: Optional[str] = None
    text_template: Optional[str] = None
    messages_template: Optional[List[PromptMessage]] = None
    model_settings: Optional[ModelSettings] = None
    output_type: Optional[OutputType] = None
    interpolation_type: Optional[PromptInterpolationType] = None


class MetricsAverageDict:
    def __init__(self):
        self.metric_dict = {}
        self.metric_count = {}

    def add_metric(self, metric_name, score):
        if metric_name not in self.metric_dict:
            self.metric_dict[metric_name] = score
            self.metric_count[metric_name] = 1
        else:
            self.metric_dict[metric_name] += score
            self.metric_count[metric_name] += 1

    def get_average_metric_score(self):
        return [
            MetricScoreType(
                metric=metric,
                score=self.metric_dict[metric] / self.metric_count[metric],
            )
            for metric in self.metric_dict
        ]


class RemainingTestRun(BaseModel):
    testRunId: str
    test_cases: List[LLMApiTestCase] = Field(
        alias="testCases", default_factory=lambda: []
    )
    conversational_test_cases: List[ConversationalApiTestCase] = Field(
        alias="conversationalTestCases", default_factory=lambda: []
    )


class TestRun(BaseModel):
    test_file: Optional[str] = Field(
        None,
        alias="testFile",
    )
    test_cases: List[LLMApiTestCase] = Field(
        alias="testCases", default_factory=lambda: []
    )
    conversational_test_cases: List[ConversationalApiTestCase] = Field(
        alias="conversationalTestCases", default_factory=lambda: []
    )
    metrics_scores: List[MetricScores] = Field(
        default_factory=lambda: [], alias="metricsScores"
    )
    trace_metrics_scores: Optional[TraceMetricScores] = Field(
        None, alias="traceMetricsScores"
    )
    identifier: Optional[str] = None
    hyperparameters: Optional[Dict[str, Any]] = Field(None)
    prompts: Optional[List[PromptData]] = Field(None)
    test_passed: Optional[int] = Field(None, alias="testPassed")
    test_failed: Optional[int] = Field(None, alias="testFailed")
    run_duration: float = Field(0.0, alias="runDuration")
    evaluation_cost: Union[float, None] = Field(None, alias="evaluationCost")
    dataset_alias: Optional[str] = Field(None, alias="datasetAlias")
    dataset_id: Optional[str] = Field(None, alias="datasetId")
    official: bool = False

    def add_test_case(
        self, api_test_case: Union[LLMApiTestCase, ConversationalApiTestCase]
    ):
        if isinstance(api_test_case, ConversationalApiTestCase):
            self.conversational_test_cases.append(api_test_case)
        else:
            self.test_cases.append(api_test_case)

        if api_test_case.evaluation_cost is not None:
            if self.evaluation_cost is None:
                self.evaluation_cost = api_test_case.evaluation_cost
            else:
                self.evaluation_cost += api_test_case.evaluation_cost

    def set_dataset_properties(
        self,
        test_case: Union[LLMTestCase, ConversationalTestCase],
    ):
        if self.dataset_alias is None:
            self.dataset_alias = test_case._dataset_alias

        if self.dataset_id is None:
            self.dataset_id = test_case._dataset_id

    @staticmethod
    def _assign_unique_orders(test_cases):
        """Assign unique sequential orders to a sorted list of test cases.

        Preserves the original gap-filling behaviour (only touch test cases
        whose order is ``None``) **unless** duplicates are detected.  When
        multiple ``evaluate()`` calls accumulate into the same test run each
        call starts its order counter from 0, producing duplicates such as
        ``[0, 0, 1, 1, ...]``.  Confident AI treats ``order`` as a unique
        position identifier, so duplicates cause earlier test cases to be
        displayed as *Skipped*.  In that case we fall back to a full
        sequential re-number to guarantee uniqueness.
        """
        # --- original logic: fill Nones, keep existing values ---
        highest_order = 0
        for test_case in test_cases:
            if test_case.order is None:
                test_case.order = highest_order
            highest_order = test_case.order + 1

        # --- check for duplicates introduced by accumulation ---
        seen = set()
        has_duplicates = False
        for test_case in test_cases:
            if test_case.order in seen:
                has_duplicates = True
                break
            seen.add(test_case.order)

        if has_duplicates:
            for i, test_case in enumerate(test_cases):
                test_case.order = i

    def sort_test_cases(self):
        self.test_cases.sort(
            key=lambda x: (x.order if x.order is not None else float("inf"))
        )
        self._assign_unique_orders(self.test_cases)

        self.conversational_test_cases.sort(
            key=lambda x: (x.order if x.order is not None else float("inf"))
        )
        self._assign_unique_orders(self.conversational_test_cases)

    def construct_metrics_scores(self) -> int:
        # Use a dict to aggregate scores, passes, and fails for each metric.
        metrics_dict: Dict[str, Dict[str, Any]] = {}
        # Add dict for trace metrics
        trace_metrics_dict: Dict[
            span_api_type_literals, Dict[str, Dict[str, Dict[str, Any]]]
        ] = {
            SpanApiType.AGENT.value: {},
            SpanApiType.TOOL.value: {},
            SpanApiType.RETRIEVER.value: {},
            SpanApiType.LLM.value: {},
            SpanApiType.BASE.value: {},
        }
        valid_scores = 0

        def process_metric_data(metric_data: MetricData):
            """
            Process and aggregate metric data for overall test metrics.

            Args:
                metric_data: The metric data to process
            """
            nonlocal valid_scores
            metric_name = metric_data.name
            score = metric_data.score
            success = metric_data.success

            if metric_name not in metrics_dict:
                metrics_dict[metric_name] = {
                    "scores": [],
                    "passes": 0,
                    "fails": 0,
                    "errors": 0,
                }

            metric_dict = metrics_dict[metric_name]

            if score is None or success is None:
                metric_dict["errors"] += 1
            else:
                valid_scores += 1
                metric_dict["scores"].append(score)
                if success:
                    metric_dict["passes"] += 1
                else:
                    metric_dict["fails"] += 1

        def process_span_metric_data(
            metric_data: MetricData,
            span_type: span_api_type_literals,
            span_name: str,
        ):
            """
            Process and aggregate metric data for a specific span.

            Args:
                metric_data: The metric data to process
                span_type: The type of span (agent, tool, retriever, llm, base)
                span_name: The name of the span
            """
            metric_name = metric_data.name
            score = metric_data.score
            success = metric_data.success

            if span_name not in trace_metrics_dict[span_type]:
                trace_metrics_dict[span_type][span_name] = {}

            if metric_name not in trace_metrics_dict[span_type][span_name]:
                trace_metrics_dict[span_type][span_name][metric_name] = {
                    "scores": [],
                    "passes": 0,
                    "fails": 0,
                    "errors": 0,
                }

            metric_dict = trace_metrics_dict[span_type][span_name][metric_name]

            if score is None or success is None:
                metric_dict["errors"] += 1
            else:
                metric_dict["scores"].append(score)
                if success:
                    metric_dict["passes"] += 1
                else:
                    metric_dict["fails"] += 1

        def process_spans(spans, span_type: span_api_type_literals):
            """
            Process all metrics for a list of spans of a specific type.

            Args:
                spans: List of spans to process
                span_type: The type of spans being processed
            """
            for span in spans:
                if span.metrics_data is not None:
                    for metric_data in span.metrics_data:
                        process_metric_data(metric_data)
                        process_span_metric_data(
                            metric_data, span_type, span.name
                        )

        # Process non-conversational test cases.
        for test_case in self.test_cases:
            if test_case.metrics_data is None:
                continue
            for metric_data in test_case.metrics_data:
                process_metric_data(metric_data)

            if test_case.trace is None:
                continue

            # Process all span types using the helper function
            process_spans(test_case.trace.agent_spans, SpanApiType.AGENT.value)
            process_spans(test_case.trace.tool_spans, SpanApiType.TOOL.value)
            process_spans(
                test_case.trace.retriever_spans, SpanApiType.RETRIEVER.value
            )
            process_spans(test_case.trace.llm_spans, SpanApiType.LLM.value)
            process_spans(test_case.trace.base_spans, SpanApiType.BASE.value)

        # Process conversational test cases.
        for convo_test_case in self.conversational_test_cases:
            if convo_test_case.metrics_data is not None:
                for metric_data in convo_test_case.metrics_data:
                    process_metric_data(metric_data)

        # Create MetricScores objects with the aggregated data.
        self.metrics_scores = [
            MetricScores(
                metric=metric,
                scores=data["scores"],
                passes=data["passes"],
                fails=data["fails"],
                errors=data["errors"],
            )
            for metric, data in metrics_dict.items()
        ]

        # Create a single TraceMetricScores object instead of a list
        trace_metrics_score = TraceMetricScores()
        has_span_metrics = False

        for span_type, spans in trace_metrics_dict.items():
            if not spans:  # Skip empty span types
                continue

            span_dict = {}
            for span_name, metrics in spans.items():
                span_dict[span_name] = {
                    metric_name: MetricScores(
                        metric=metric_name,
                        scores=metric_data["scores"],
                        passes=metric_data["passes"],
                        fails=metric_data["fails"],
                        errors=metric_data["errors"],
                    )
                    for metric_name, metric_data in metrics.items()
                }

            if span_dict:  # Only set if there are spans
                has_span_metrics = True
                setattr(trace_metrics_score, span_type, span_dict)

        # Set to None if no span metrics were found
        self.trace_metrics_scores = (
            trace_metrics_score if has_span_metrics else None
        )
        return valid_scores

    def calculate_test_passes_and_fails(self):
        test_passed = 0
        test_failed = 0
        for test_case in self.test_cases:
            if test_case.success is not None:
                if test_case.success:
                    test_passed += 1
                else:
                    test_failed += 1

        for test_case in self.conversational_test_cases:
            # we don't count for conversational messages success
            if test_case.success is not None:
                if test_case.success:
                    test_passed += 1
                else:
                    test_failed += 1

        self.test_passed = test_passed
        self.test_failed = test_failed

    def save(self, f):
        try:
            body = self.model_dump(by_alias=True, exclude_none=True)
        except AttributeError:
            body = self.dict(by_alias=True, exclude_none=True)
        json.dump(body, f, cls=TestRunEncoder)
        f.flush()
        os.fsync(f.fileno())
        return self

    @classmethod
    def load(cls, f):
        data: dict = json.load(f)
        return cls(**data)

    def guard_mllm_test_cases(self):
        for test_case in self.test_cases:
            if test_case.is_multimodal():
                raise ValueError(
                    "Unable to send multimodal test cases to Confident AI."
                )


class TestRunEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Enum):
            return obj.value
        return make_json_serializable(obj)


class TestRunManager:
    def __init__(self):
        self.test_run = None
        self.temp_file_path = TEMP_FILE_PATH
        self.save_to_disk = False
        self.disable_request = False
        self.results_folder: Optional[str] = None
        self.results_subfolder: Optional[str] = None
        # Timestamped export if one was written, else rolling snapshot.
        # Consumed by the post-run inspect prompt.
        self.last_saved_path: Optional[Path] = None

    def reset(self):
        self.test_run = None
        self.temp_file_path = TEMP_FILE_PATH
        self.save_to_disk = False
        self.disable_request = False
        self.results_folder = None
        self.results_subfolder = None
        self.last_saved_path = None

    def configure_local_store(
        self,
        results_folder: Optional[str] = None,
        results_subfolder: Optional[str] = None,
    ):
        """Configure where `save_test_run_locally` writes the full TestRun JSON.

        Values set here take precedence over the `DEEPEVAL_RESULTS_FOLDER`
        env var. Intended to be called from `evaluate()` / `evals_iterator()`
        right before `wrap_up_test_run()`.
        """
        self.results_folder = results_folder
        self.results_subfolder = results_subfolder
        # The manager is a long-lived singleton, so a previous run's path
        # could linger and mislead the inspect prompt into offering a stale
        # file. Clear it whenever a new run configures its local store.
        self.last_saved_path = None

    def set_test_run(self, test_run: TestRun):
        self.test_run = test_run

    def create_test_run(
        self,
        identifier: Optional[str] = None,
        file_name: Optional[str] = None,
        disable_request: Optional[bool] = False,
    ):
        self.disable_request = disable_request
        test_run = TestRun(
            identifier=identifier,
            testFile=file_name,
            testCases=[],
            metricsScores=[],
            hyperparameters=None,
            testPassed=None,
            testFailed=None,
        )
        self.set_test_run(test_run)

        if self.save_to_disk:
            self.save_test_run(self.temp_file_path)

    def get_test_run(self, identifier: Optional[str] = None):
        if self.test_run is None:
            self.create_test_run(identifier=identifier)

        if portalocker and self.save_to_disk:
            try:
                with portalocker.Lock(
                    self.temp_file_path,
                    mode="r",
                    flags=portalocker.LOCK_SH | portalocker.LOCK_NB,
                ) as file:
                    loaded = self.test_run.load(file)
                    # only overwrite if loading actually worked
                    self.test_run = loaded
            except (
                FileNotFoundError,
                json.JSONDecodeError,
                portalocker.exceptions.LockException,
            ) as e:
                print(
                    f"Warning: Could not load test run from disk: {e}",
                    file=sys.stderr,
                )

        return self.test_run

    def save_test_run(self, path: str, save_under_key: Optional[str] = None):
        if portalocker and self.save_to_disk:
            try:
                # ensure parent directory exists
                parent = os.path.dirname(path)
                if parent:
                    os.makedirs(parent, exist_ok=True)

                with portalocker.Lock(path, mode="w") as file:
                    if save_under_key:
                        try:
                            test_run_data = self.test_run.model_dump(
                                by_alias=True, exclude_none=True
                            )
                        except AttributeError:
                            # Pydantic version below 2.0
                            test_run_data = self.test_run.dict(
                                by_alias=True, exclude_none=True
                            )
                        wrapper_data = {save_under_key: test_run_data}
                        json.dump(wrapper_data, file, cls=TestRunEncoder)
                        file.flush()
                        os.fsync(file.fileno())
                    else:
                        self.test_run.save(file)
            except portalocker.exceptions.LockException:
                pass

    def save_final_test_run_link(self, link: str):
        if portalocker:
            try:
                with portalocker.Lock(
                    LATEST_TEST_RUN_FILE_PATH, mode="w"
                ) as file:
                    json.dump({LATEST_TEST_RUN_LINK_KEY: link}, file)
                    file.flush()
                    os.fsync(file.fileno())
            except portalocker.exceptions.LockException:
                pass

    def update_test_run(
        self,
        api_test_case: Union[LLMApiTestCase, ConversationalApiTestCase],
        test_case: Union[LLMTestCase, ConversationalTestCase],
    ):
        if (
            api_test_case.metrics_data is not None
            and len(api_test_case.metrics_data) == 0
            and api_test_case.trace is None
        ):
            return

        if portalocker and self.save_to_disk:
            try:
                with portalocker.Lock(
                    self.temp_file_path,
                    mode="r+",
                    flags=portalocker.LOCK_EX,
                ) as file:
                    file.seek(0)
                    self.test_run = self.test_run.load(file)

                    # Update the test run object
                    self.test_run.add_test_case(api_test_case)
                    self.test_run.set_dataset_properties(test_case)

                    # Save the updated test run back to the file
                    file.seek(0)
                    file.truncate()
                    self.test_run.save(file)
            except (
                FileNotFoundError,
                json.JSONDecodeError,
                portalocker.exceptions.LockException,
            ) as e:
                print(
                    f"Warning: Could not update test run on disk: {e}",
                    file=sys.stderr,
                )
                if self.test_run is None:
                    # guarantee a valid in-memory run so the update can proceed.
                    # never destroy in-memory state on I/O failure.
                    self.create_test_run()
                self.test_run.add_test_case(api_test_case)
                self.test_run.set_dataset_properties(test_case)
        else:
            if self.test_run is None:
                self.create_test_run()

            self.test_run.add_test_case(api_test_case)
            self.test_run.set_dataset_properties(test_case)

    def clear_test_run(self):
        self.test_run = None

    @staticmethod
    def _calculate_success_rate(pass_count: int, fail_count: int) -> str:
        """Calculate success rate percentage or return error message."""
        total = pass_count + fail_count
        if total > 0:
            return str(round((100 * pass_count) / total, 2))
        return "Cannot display metrics for component-level evals, please run 'deepeval view' to see results on Confident AI."

    @staticmethod
    def _get_metric_status(metric_data: MetricData) -> str:
        """Get formatted status string for a metric."""
        if metric_data.error:
            return "[red]ERRORED[/red]"
        elif metric_data.success:
            return "[green]PASSED[/green]"
        return "[red]FAILED[/red]"

    @staticmethod
    def _format_metric_score(metric_data: MetricData) -> str:
        """Format metric score with evaluation details."""
        evaluation_model = metric_data.evaluation_model or "n/a"
        metric_score = (
            round(metric_data.score, 2)
            if metric_data.score is not None
            else None
        )

        return (
            f"{metric_score} "
            f"(threshold={metric_data.threshold}, "
            f"evaluation model={evaluation_model}, "
            f"reason={metric_data.reason}, "
            f"error={metric_data.error})"
        )

    @staticmethod
    def _should_skip_test_case(
        test_case, display: TestRunResultDisplay
    ) -> bool:
        """Determine if test case should be skipped based on display filter."""
        if display == TestRunResultDisplay.PASSING and not test_case.success:
            return True
        elif display == TestRunResultDisplay.FAILING and test_case.success:
            return True
        return False

    @staticmethod
    def _count_metric_results(
        metrics_data: List[MetricData],
    ) -> tuple[int, int]:
        """Count passing and failing metrics."""
        pass_count = 0
        fail_count = 0
        for metric_data in metrics_data:
            if metric_data.success:
                pass_count += 1
            else:
                fail_count += 1
        return pass_count, fail_count

    def _add_test_case_header_row(
        self,
        table: Table,
        test_case_name: str,
        pass_count: int,
        fail_count: int,
    ):
        """Add test case header row with name and success rate."""
        success_rate = self._calculate_success_rate(pass_count, fail_count)
        table.add_row(
            test_case_name,
            *[""] * 3,
            f"{success_rate}%",
        )

    def _add_metric_rows(self, table: Table, metrics_data: List[MetricData]):
        """Add metric detail rows to the table."""
        for metric_data in metrics_data:
            status = self._get_metric_status(metric_data)
            formatted_score = self._format_metric_score(metric_data)

            table.add_row(
                "",
                str(metric_data.name),
                formatted_score,
                status,
                "",
            )

    def _add_separator_row(self, table: Table):
        """Add empty separator row between test cases."""
        table.add_row(*[""] * len(table.columns))

    def display_results_table(
        self, test_run: TestRun, display: TestRunResultDisplay
    ):
        """Display test results in a formatted table."""

        table = Table(title="Test Results")
        column_config = dict(justify="left")
        column_names = [
            "Test case",
            "Metric",
            "Score",
            "Status",
            "Overall Success Rate",
        ]

        for name in column_names:
            table.add_column(name, **column_config)

        # Process regular test cases
        for index, test_case in enumerate(test_run.test_cases):
            if test_case.metrics_data is None or self._should_skip_test_case(
                test_case, display
            ):
                continue
            pass_count, fail_count = self._count_metric_results(
                test_case.metrics_data
            )
            self._add_test_case_header_row(
                table, test_case.name, pass_count, fail_count
            )
            self._add_metric_rows(table, test_case.metrics_data)

            if index < len(test_run.test_cases) - 1:
                self._add_separator_row(table)

        # Process conversational test cases
        for index, conversational_test_case in enumerate(
            test_run.conversational_test_cases
        ):
            if self._should_skip_test_case(conversational_test_case, display):
                continue

            conversational_test_case_name = conversational_test_case.name

            if conversational_test_case.turns:
                turns_table = Table(
                    title=f"Conversation - {conversational_test_case_name}",
                    show_header=True,
                    header_style="bold",
                )
                turns_table.add_column("#", justify="right", width=3)
                turns_table.add_column("Role", justify="left", width=10)

                # subtract fixed widths + borders and padding.
                # ~20 as a safe buffer
                details_max_width = max(
                    48, min(120, console.width - 3 - 10 - 20)
                )
                turns_table.add_column(
                    "Details",
                    justify="left",
                    overflow="fold",
                    max_width=details_max_width,
                )

                # truncate when too long
                tools_max_width = min(60, max(24, console.width // 3))
                turns_table.add_column(
                    "Tools",
                    justify="left",
                    no_wrap=True,
                    overflow="ellipsis",
                    max_width=tools_max_width,
                )

                sorted_turns = sorted(
                    conversational_test_case.turns, key=lambda t: t.order
                )

                for t in sorted_turns:
                    tools = t.tools_called or []
                    tool_names = ", ".join(tc.name for tc in tools)

                    # omit order, role and tools since we show them in a separate columns.
                    details = format_turn(
                        t,
                        include_tools_in_header=False,
                        include_order_role_in_header=False,
                    )

                    turns_table.add_row(
                        str(t.order),
                        t.role,
                        details,
                        shorten(tool_names, len_short()),
                    )

                console.print(turns_table)
            else:
                console.print(
                    f"[dim]No turns recorded for {conversational_test_case_name}.[/dim]"
                )
            if conversational_test_case.metrics_data is not None:
                pass_count, fail_count = self._count_metric_results(
                    conversational_test_case.metrics_data
                )
                self._add_test_case_header_row(
                    table, conversational_test_case.name, pass_count, fail_count
                )
                self._add_metric_rows(
                    table, conversational_test_case.metrics_data
                )

            if index < len(test_run.conversational_test_cases) - 1:
                self._add_separator_row(table)

            if index < len(test_run.test_cases) - 1:
                self._add_separator_row(table)

        table.add_row(
            "[bold red]Note: Use Confident AI with DeepEval to analyze failed test cases for more details[/bold red]",
            *[""] * (len(table.columns) - 1),
        )
        print(table)

    def post_test_run(self, test_run: TestRun) -> Optional[Tuple[str, str]]:
        if (
            len(test_run.test_cases) == 0
            and len(test_run.conversational_test_cases) == 0
        ):
            print("No test cases found, unable to upload to Confident AI.")
            return

        api = Api()

        is_conversational_run = len(test_run.conversational_test_cases) > 0
        all_test_cases_to_process = (
            test_run.conversational_test_cases
            if is_conversational_run
            else test_run.test_cases
        )

        custom_batch_size = os.getenv(CONFIDENT_TEST_CASE_BATCH_SIZE)
        if custom_batch_size and custom_batch_size.isdigit():
            BATCH_SIZE = int(custom_batch_size)
        else:
            BATCH_SIZE = 20 if is_conversational_run else 40

        initial_batch = all_test_cases_to_process[:BATCH_SIZE]
        remaining_test_cases_to_process = all_test_cases_to_process[BATCH_SIZE:]

        if len(remaining_test_cases_to_process) > 0:
            console.print(
                "Sending a large test run to Confident, this might take a bit longer than usual..."
            )

        ####################
        ### POST REQUEST ###
        ####################
        if is_conversational_run:
            test_run.conversational_test_cases = initial_batch
        else:
            test_run.test_cases = initial_batch

        try:
            test_run.prompts = None
            body = test_run.model_dump(by_alias=True, exclude_none=True)
        except AttributeError:
            # Pydantic version below 2.0
            body = test_run.dict(by_alias=True, exclude_none=True)

        json_str = json.dumps(body, cls=TestRunEncoder)
        body = json.loads(json_str)

        data, link = api.send_request(
            method=HttpMethods.POST,
            endpoint=Endpoints.TEST_RUN_ENDPOINT,
            body=body,
        )

        if not isinstance(data, dict) or "id" not in data:
            # try to show helpful details
            detail = None
            if isinstance(data, dict):
                detail = (
                    data.get("detail")
                    or data.get("message")
                    or data.get("error")
                )
            # fall back to repr for visibility
            raise RuntimeError(
                f"Confident API response missing 'id'. "
                f"detail={detail!r} raw={type(data).__name__}:{repr(data)[:500]}"
            )

        res = TestRunHttpResponse(
            id=data["id"],
        )

        ################################################
        ### Send the remaining test cases in batches ###
        ################################################
        total_remaining = len(remaining_test_cases_to_process)
        num_remaining_batches = (
            (total_remaining + BATCH_SIZE - 1) // BATCH_SIZE
            if total_remaining > 0
            else 0
        )

        for i in range(num_remaining_batches):
            start_index = i * BATCH_SIZE
            batch = remaining_test_cases_to_process[
                start_index : start_index + BATCH_SIZE
            ]

            if len(batch) == 0:
                break  # Should not happen with correct num_remaining_batches, but as a safeguard

            # Create RemainingTestRun with the correct list populated
            if is_conversational_run:
                remaining_test_run = RemainingTestRun(
                    testRunId=res.id,
                    testCases=[],  # This will be empty
                    conversationalTestCases=batch,
                )
            else:
                remaining_test_run = RemainingTestRun(
                    testRunId=res.id,
                    testCases=batch,
                    conversationalTestCases=[],  # This will be empty
                )

            body = None
            try:
                body = remaining_test_run.model_dump(
                    by_alias=True, exclude_none=True
                )
            except AttributeError:
                # Pydantic version below 2.0
                body = remaining_test_run.dict(by_alias=True, exclude_none=True)

            try:
                _, _ = api.send_request(
                    method=HttpMethods.PUT,
                    endpoint=Endpoints.TEST_RUN_ENDPOINT,
                    body=body,
                )
            except Exception as e:
                message = f"Unexpected error when sending some test cases. Incomplete test run available at {link}"
                raise Exception(message) from e

        console.print(
            "[rgb(5,245,141)]✓[/rgb(5,245,141)] Done 🎉! View results on "
            f"[link={link}]{link}[/link]"
        )
        self.save_final_test_run_link(link)
        open_browser(link)
        return link, res.id

    def save_test_run_locally(self):
        """Persist the current TestRun to disk.

        Always writes a rolling snapshot to `.deepeval/.latest_run_full.json`.
        Additionally writes a timestamped `test_run_<YYYYMMDD_HHMMSS>.json` to
        `results_folder` (or `DEEPEVAL_RESULTS_FOLDER`) when set.
        """
        if self.test_run is None:
            return

        from deepeval.evaluate.local_store import (
            resolve_target_dir,
            write_rolling_test_run,
            write_test_run,
        )

        rolling_path = write_rolling_test_run(self.test_run)
        if rolling_path is not None:
            self.last_saved_path = rolling_path

        target_dir = resolve_target_dir(
            results_folder=self.results_folder,
            results_subfolder=self.results_subfolder,
        )
        if target_dir is None:
            return

        if target_dir.exists() and target_dir.is_file():
            print(
                f"❌ Error: results_folder={target_dir} already exists and is a file.\n"
                "Detailed results won't be saved. Please specify a folder or an available path."
            )
            return

        try:
            path = write_test_run(target_dir, self.test_run)
            self.last_saved_path = path
            print(f"Test run saved at {path}")
        except Exception as e:
            print(
                f"Warning: failed to save test run to {target_dir}: {e}",
                file=sys.stderr,
            )

    def wrap_up_test_run(
        self,
        runDuration: float,
        display_table: bool = True,
        display: Optional[TestRunResultDisplay] = TestRunResultDisplay.ALL,
    ) -> Optional[Tuple[str, str]]:
        test_run = self.get_test_run()
        if test_run is None:
            print("Test Run is empty, please try again.")
            delete_file_if_exists(self.temp_file_path)
            return
        elif (
            len(test_run.test_cases) == 0
            and len(test_run.conversational_test_cases) == 0
        ):
            print("No test cases found, please try again.")
            delete_file_if_exists(self.temp_file_path)
            return

        # Mark the run as the official if requested via the `--official` CLI flag or `evaluate(official=True)`.
        # Set here, in the main process right before upload, so it rides along in the test run
        # creation payload (and survives any xdist worker disk round-trips).
        if get_test_run_official():
            test_run.official = True

        # Don't block the post when all metrics errored — the spans still
        # carry the underlying error info (populated by ``Observer.__exit__``)
        # which the dashboard can render. Just warn so it's not mistaken
        # for a successful run.
        valid_scores = test_run.construct_metrics_scores()
        if valid_scores == 0:
            console.print(
                "\n[bold yellow]⚠ WARNING:[/bold yellow] All metrics errored "
                "across every test case — no metric scores were recorded. "
                "Posting the run anyway so you can inspect the trace + span "
                "errors on the Confident AI dashboard.\n"
            )
        test_run.run_duration = runDuration
        test_run.calculate_test_passes_and_fails()
        test_run.sort_test_cases()

        if global_test_run_cache_manager.disable_write_cache is None:
            global_test_run_cache_manager.disable_write_cache = not bool(
                get_is_running_deepeval()
            )
        global_test_run_cache_manager.wrap_up_cached_test_run()

        if display_table:
            self.display_results_table(test_run, display)

        if test_run.hyperparameters is None:
            console.print(
                "\n[bold yellow]⚠ WARNING:[/bold yellow] No hyperparameters logged.\n"
                "» [bold blue][link=https://deepeval.com/docs/evaluation-prompts]Log hyperparameters[/link][/bold blue] to attribute prompts and models to your test runs.\n\n"
                + "=" * 80
            )
        else:
            if not test_run.prompts:
                console.print(
                    "\n[bold yellow]⚠ WARNING:[/bold yellow] No prompts logged.\n"
                    "» [bold blue][link=https://deepeval.com/docs/evaluation-prompts]Log prompts[/link][/bold blue] to evaluate and optimize your prompt templates and models.\n\n"
                    + "=" * 80
                )
            else:
                console.print("\n[bold green]✓ Prompts Logged[/bold green]\n")
                self._render_prompts_panels(prompts=test_run.prompts)

        self.save_test_run_locally()
        delete_file_if_exists(self.temp_file_path)
        confident_enabled = is_confident()
        if confident_enabled and self.disable_request is False:
            return self.post_test_run(test_run)
        else:
            self.save_test_run(
                LATEST_TEST_RUN_FILE_PATH,
                save_under_key=LATEST_TEST_RUN_DATA_KEY,
            )
            token_cost = (
                f"{test_run.evaluation_cost} USD"
                if test_run.evaluation_cost
                else "None"
            )
            console.print(
                f"\n\n[rgb(5,245,141)]✓[/rgb(5,245,141)] Evaluation completed 🎉! (time taken: {round(runDuration, 2)}s | token cost: {token_cost})\n"
                f"» Test Results ({test_run.test_passed + test_run.test_failed} total tests):\n",
                f"  » Pass Rate: {round((test_run.test_passed / (test_run.test_passed + test_run.test_failed)) * 100, 2)}% | Passed: [bold green]{test_run.test_passed}[/bold green] | Failed: [bold red]{test_run.test_failed}[/bold red]\n\n",
                "=" * 80,
                "\n\n» Want to share evals with your team, or a place for your test cases to live? ❤️ 🏡\n"
                "  » Run [bold]'deepeval view'[/bold] to analyze and save testing results on [rgb(106,0,255)]Confident AI[/rgb(106,0,255)].\n\n",
            )

    def get_latest_test_run_data(self) -> Optional[TestRun]:
        try:
            if os.path.exists(LATEST_TEST_RUN_FILE_PATH):
                with open(LATEST_TEST_RUN_FILE_PATH, "r") as file:
                    data = json.load(file)
                    return TestRun.model_validate(
                        data[LATEST_TEST_RUN_DATA_KEY]
                    )
        except (FileNotFoundError, json.JSONDecodeError, Exception):
            pass
        return None

    def get_latest_test_run_link(self) -> Optional[str]:
        try:
            if os.path.exists(LATEST_TEST_RUN_FILE_PATH):
                with open(LATEST_TEST_RUN_FILE_PATH, "r") as file:
                    data = json.load(file)
                    return data[LATEST_TEST_RUN_LINK_KEY]
        except (FileNotFoundError, json.JSONDecodeError, Exception):
            pass
        return None

    def _render_prompts_panels(self, prompts: List[PromptData]) -> None:

        def format_string(
            v, default="[dim]None[/dim]", color: Optional[str] = None
        ):
            formatted_string = str(v) if v not in (None, "", []) else default
            return (
                f"{formatted_string}"
                if color is None or v in (None, "", [])
                else f"[{color}]{formatted_string}[/]"
            )

        panels = []
        for prompt in prompts:
            lines = []
            p_type = (
                "messages"
                if prompt.messages_template
                else ("text" if prompt.text_template else "—")
            )
            if p_type:
                lines.append(f"type: {format_string(p_type, color='blue')}")
            if prompt.output_type:
                lines.append(
                    f"output_type: {format_string(prompt.output_type, color='blue')}"
                )
            if prompt.interpolation_type:
                lines.append(
                    f"interpolation_type: {format_string(prompt.interpolation_type, color='blue')}"
                )
            if prompt.model_settings:
                ms = prompt.model_settings
                settings_lines = [
                    "Model Settings:",
                    f"  – provider: {format_string(ms.provider, color='green')}",
                    f"  – name: {format_string(ms.name, color='green')}",
                    f"  – temperature: {format_string(ms.temperature, color='green')}",
                    f"  – max_tokens: {format_string(ms.max_tokens, color='green')}",
                    f"  – top_p: {format_string(ms.top_p, color='green')}",
                    f"  – frequency_penalty: {format_string(ms.frequency_penalty, color='green')}",
                    f"  – presence_penalty: {format_string(ms.presence_penalty, color='green')}",
                    f"  – stop_sequence: {format_string(ms.stop_sequence, color='green')}",
                    f"  – reasoning_effort: {format_string(ms.reasoning_effort, color='green')}",
                    f"  – verbosity: {format_string(ms.verbosity, color='green')}",
                ]
                lines.append("")
                lines.extend(settings_lines)
            title = f"{format_string(prompt.alias)}"
            if prompt.hash:
                title += f" ({prompt.hash})"
            body = "\n".join(lines)
            panel = Panel(
                body,
                title=title,
                title_align="left",
                expand=False,
                padding=(1, 6, 1, 2),
            )
            panels.append(panel)

        if panels:
            console.print(Columns(panels, equal=False, expand=False))


global_test_run_manager = TestRunManager()