/deepeval/evaluate/execute/loop.py

import logging

from rich.progress import (
    Progress,
    TextColumn,
    BarColumn,
    TimeElapsedColumn,
    TaskProgressColumn,
)
from typing import (
    Callable,
    List,
    Optional,
    Awaitable,
    Iterator,
)
import asyncio
import time

from deepeval.evaluate.configs import (
    ErrorConfig,
    DisplayConfig,
    CacheConfig,
    AsyncConfig,
)
from deepeval.tracing.tracing import (
    Observer,
    trace_manager,
    Trace,
    BaseSpan,
    AgentSpan,
    LlmSpan,
    RetrieverSpan,
    ToolSpan,
)
from deepeval.tracing.context import current_trace_context
from deepeval.tracing.api import (
    BaseApiSpan,
)
from deepeval.dataset import Golden
from deepeval.contextvars import set_current_golden, reset_current_golden
from deepeval.constants import PYTEST_TRACE_TEST_WRAPPER_SPAN_NAME
from deepeval.errors import DeepEvalError
from deepeval.metrics.utils import copy_metrics
from deepeval.utils import (
    shorten,
    len_medium,
    format_error_text,
    get_per_task_timeout_seconds,
    get_gather_timeout,
)
from deepeval.telemetry import capture_evaluation_run
from deepeval.metrics import BaseMetric

from deepeval.test_case import (
    LLMTestCase,
)
from deepeval.test_case.api import create_api_test_case
from deepeval.test_run import (
    global_test_run_manager,
    TestRunManager,
)
from deepeval.evaluate.types import TestResult
from deepeval.evaluate.utils import (
    create_api_trace,
    create_metric_data,
    create_test_result,
    count_metrics_in_trace,
    count_total_metrics_for_trace,
    count_metrics_in_span_subtree,
    extract_trace_test_results,
)
from deepeval.utils import add_pbar, update_pbar, custom_console
from deepeval.tracing.types import (
    EvalMode,
    EvalSession,
    TestCaseMetricPair,
    TraceSpanStatus,
)
from deepeval.tracing.api import TraceSpanApiStatus
from deepeval.config.settings import get_settings

logger = logging.getLogger(__name__)


from deepeval.evaluate.execute._common import (
    _await_with_outer_deadline,
    _execute_metric,
    _log_gather_timeout,
    _pick_root_for_marking,
    _resolve_trace_and_root_for_task,
    _skip_metrics_for_error,
    _snapshot_tasks,
    _trace_error,
    filter_duplicate_results,
    log_prompt,
)
from deepeval.evaluate.execute.agentic import (
    _a_execute_agentic_test_case,
)
from deepeval.evaluate.execute.e2e import _evaluate_test_case_pairs


def _span_subtree_has_metrics(span: BaseSpan) -> bool:
    """True if ``span`` or any of its descendants declares a metric source."""
    if span.metrics:
        return True
    return any(_span_subtree_has_metrics(c) for c in span.children)


def _has_any_evaluable_metrics(
    trace_metrics: Optional[List[BaseMetric]],
    traces: List[Trace],
    test_case_metrics: List[TestCaseMetricPair],
) -> bool:
    """Return True if at least one metric source exists for this eval run.

    Metrics can come from: ``trace_metrics`` (iterator arg), ``trace.metrics``
    (``update_current_trace``/root ``@observe``), ``span.metrics`` anywhere in
    a trace subtree, or ``test_case_metrics`` (external SDK path). This check
    is intentionally lazy (post-iteration) since span metrics only exist after
    user code has run.
    """
    if trace_metrics:
        return True
    if test_case_metrics:
        return True
    for trace in traces:
        if not isinstance(trace, Trace):
            continue
        if trace.metrics:
            return True
        if any(_span_subtree_has_metrics(s) for s in trace.root_spans):
            return True
    return False


def _raise_no_metrics_error() -> None:
    """Raise a uniform NoMetricsError with actionable guidance."""
    from deepeval.errors import NoMetricsError

    raise NoMetricsError(
        "evals_iterator was started but no metrics were declared anywhere. "
        "An evaluation run with zero metric sources cannot produce results.\n"
    )


def execute_agentic_test_cases_from_loop(
    goldens: List[Golden],
    trace_metrics: Optional[List[BaseMetric]],
    test_results: List[TestResult],
    display_config: Optional[DisplayConfig] = DisplayConfig(),
    cache_config: Optional[CacheConfig] = CacheConfig(),
    error_config: Optional[ErrorConfig] = ErrorConfig(),
    identifier: Optional[str] = None,
    _use_bar_indicator: bool = True,
    _is_assert_test: bool = False,
) -> Iterator[TestResult]:

    test_run_manager = global_test_run_manager
    test_run_manager.save_to_disk = cache_config.write_cache
    test_run_manager.get_test_run(identifier=identifier)

    local_trace_manager = trace_manager
    local_trace_manager.eval_session = EvalSession(mode=EvalMode.ITERATOR_SYNC)

    def evaluate_test_cases(
        progress: Optional[Progress] = None,
        pbar_id: Optional[int] = None,
    ) -> Iterator[Golden]:
        count = 0
        show_metric_indicator = (
            display_config.show_indicator and not _use_bar_indicator
        )
        # Per-run buffer of traces produced by user code. Accumulated locally
        # so the post-iteration "any metrics?" guard only inspects THIS run.
        processed_traces: List[Trace] = []

        for golden in goldens:
            token = set_current_golden(golden)
            with capture_evaluation_run("golden"):
                # yield golden
                count += 1
                pbar_tags_id = add_pbar(
                    progress, f"\t⚡ Invoking observed callback (#{count})"
                )
                with Observer(
                    "custom",
                    func_name=PYTEST_TRACE_TEST_WRAPPER_SPAN_NAME,
                    _progress=progress,
                    _pbar_callback_id=pbar_tags_id,
                ):
                    try:
                        # yield golden to user code
                        yield golden
                        # control has returned from user code without error, capture trace now
                        current_trace: Trace = current_trace_context.get()
                        processed_traces.append(current_trace)
                    finally:
                        # after user code returns control, always reset the context
                        reset_current_golden(token)

                update_pbar(progress, pbar_tags_id)
                update_pbar(progress, pbar_id)

                # Create empty trace api for llm api test case
                trace_api = create_api_trace(trace=current_trace, golden=golden)

                # Format golden as test case to create llm api test case
                test_case = LLMTestCase(
                    input=golden.input,
                    actual_output=(
                        str(current_trace.output)
                        if current_trace.output is not None
                        else None
                    ),
                    expected_output=current_trace.expected_output,
                    context=current_trace.context,
                    retrieval_context=current_trace.retrieval_context,
                    metadata=golden.additional_metadata,
                    tools_called=current_trace.tools_called,
                    expected_tools=current_trace.expected_tools,
                    comments=golden.comments,
                    name=golden.name,
                    _dataset_alias=golden._dataset_alias,
                    _dataset_id=golden._dataset_id,
                )
                api_test_case = create_api_test_case(
                    test_case=test_case,
                    trace=trace_api,
                    index=count if not _is_assert_test else None,
                )

                # Run DFS to calculate metrics synchronously
                def dfs(
                    span: BaseSpan,
                    progress: Optional[Progress] = None,
                    pbar_eval_id: Optional[int] = None,
                ):
                    # Create API Span
                    metrics: List[BaseMetric] = list(span.metrics or [])

                    api_span: BaseApiSpan = (
                        trace_manager._convert_span_to_api_span(span)
                    )

                    if isinstance(span, AgentSpan):
                        trace_api.agent_spans.append(api_span)
                    elif isinstance(span, LlmSpan):
                        trace_api.llm_spans.append(api_span)
                        log_prompt(span, test_run_manager)
                    elif isinstance(span, RetrieverSpan):
                        trace_api.retriever_spans.append(api_span)
                    elif isinstance(span, ToolSpan):
                        trace_api.tool_spans.append(api_span)
                    else:
                        trace_api.base_spans.append(api_span)

                    # Skip errored trace/span
                    if _skip_metrics_for_error(span=span, trace=current_trace):
                        api_span.status = TraceSpanApiStatus.ERRORED
                        api_span.error = span.error or _trace_error(
                            current_trace
                        )
                        if progress and pbar_eval_id is not None:
                            update_pbar(
                                progress,
                                pbar_eval_id,
                                advance=count_metrics_in_span_subtree(span),
                            )
                        return

                    for child in span.children:
                        dfs(child, progress, pbar_eval_id)

                    if not span.metrics:
                        return

                    requires_trace = any(
                        metric.requires_trace for metric in metrics
                    )

                    llm_test_case = None
                    if span.input is not None:
                        llm_test_case = LLMTestCase(
                            input=str(span.input),
                            actual_output=(
                                str(span.output)
                                if span.output is not None
                                else None
                            ),
                            expected_output=span.expected_output,
                            context=span.context,
                            retrieval_context=span.retrieval_context,
                            tools_called=span.tools_called,
                            expected_tools=span.expected_tools,
                        )

                    if requires_trace:
                        if llm_test_case is None:
                            llm_test_case = LLMTestCase(input="None")
                        llm_test_case._trace_dict = (
                            trace_manager.create_nested_spans_dict(span)
                        )
                    else:
                        if llm_test_case is None:
                            api_span.status = TraceSpanApiStatus.ERRORED
                            api_span.error = format_error_text(
                                DeepEvalError(
                                    "Span has metrics but no LLMTestCase. "
                                    "Are you sure you called `update_current_span()`?"
                                )
                            )
                            if progress and pbar_eval_id is not None:
                                update_pbar(
                                    progress,
                                    pbar_eval_id,
                                    advance=count_metrics_in_span_subtree(span),
                                )
                            return

                    # Preparing metric calculation
                    api_span.metrics_data = []
                    for metric in metrics:
                        metric.skipped = False
                        metric.error = None
                        if display_config.verbose_mode is not None:
                            metric.verbose_mode = display_config.verbose_mode

                    # Metric calculation
                    for metric in metrics:
                        metric_data = None
                        res = _execute_metric(
                            metric=metric,
                            test_case=llm_test_case,
                            show_metric_indicator=show_metric_indicator,
                            in_component=True,
                            error_config=error_config,
                        )
                        if res == "skip":
                            continue

                        metric_data = create_metric_data(metric)
                        api_span.metrics_data.append(metric_data)
                        api_test_case.update_status(metric_data.success)
                        update_pbar(progress, pbar_eval_id)

                if trace_metrics:
                    current_trace.metrics = trace_metrics

                trace_level_metrics_count = (
                    len(current_trace.metrics) if current_trace.metrics else 0
                )
                pbar_eval_id = add_pbar(
                    progress,
                    f"     🎯 Evaluating component(s) (#{count})",
                    total=count_metrics_in_trace(trace=current_trace)
                    + trace_level_metrics_count,
                )

                start_time = time.perf_counter()

                # On errored traces, skip trace-level metrics (no test case
                # to judge) but DO run the span-level DFS walker below —
                # it's what hydrates ``trace_api.*_spans`` for the dashboard,
                # and per-span metric skip is handled inside ``dfs``.
                skip_metrics_for_this_golden = False
                if _skip_metrics_for_error(trace=current_trace):
                    trace_api.status = TraceSpanApiStatus.ERRORED
                    if progress and pbar_eval_id is not None:
                        update_pbar(
                            progress,
                            pbar_eval_id,
                            advance=count_total_metrics_for_trace(
                                current_trace
                            ),
                        )
                elif current_trace.metrics:
                    requires_trace = any(
                        metric.requires_trace
                        for metric in current_trace.metrics
                    )

                    # Build the trace-level LLMTestCase from the golden
                    # directly, the same way the async iterator does
                    # (see ``_a_evaluate_trace``). This makes top-level
                    # ``metrics=[...]`` work out of the box even when the
                    # user never calls ``update_current_trace(input=...)``.
                    llm_test_case = LLMTestCase(
                        input=golden.input,
                        actual_output=(
                            str(current_trace.output)
                            if current_trace.output is not None
                            else golden.actual_output
                        ),
                        expected_output=current_trace.expected_output,
                        context=current_trace.context,
                        retrieval_context=current_trace.retrieval_context,
                        tools_called=current_trace.tools_called,
                        expected_tools=current_trace.expected_tools,
                    )

                    if requires_trace:
                        llm_test_case._trace_dict = (
                            trace_manager.create_nested_spans_dict(
                                current_trace.root_spans[0]
                            )
                        )

                    if not skip_metrics_for_this_golden:
                        for metric in current_trace.metrics:
                            metric.skipped = False
                            metric.error = None
                            if display_config.verbose_mode is not None:
                                metric.verbose_mode = (
                                    display_config.verbose_mode
                                )

                        trace_api.metrics_data = []
                        for metric in current_trace.metrics:
                            res = _execute_metric(
                                metric=metric,
                                test_case=llm_test_case,
                                show_metric_indicator=show_metric_indicator,
                                in_component=True,
                                error_config=error_config,
                            )
                            if res == "skip":
                                continue

                            if not metric.skipped:
                                metric_data = create_metric_data(metric)
                                trace_api.metrics_data.append(metric_data)
                                api_test_case.update_metric_data(metric_data)
                                api_test_case.update_status(metric_data.success)
                                update_pbar(progress, pbar_eval_id)

                # Always walk spans, even on errored traces — the walker
                # hydrates ``trace_api.*_spans`` and the user needs that
                # data on the dashboard to diagnose. Walk EVERY root, not
                # just ``root_spans[0]``: OTel integrations can land
                # multiple logical roots when a child ends before its
                # parent. Mirrors the async path in ``agentic.py``.
                for root in current_trace.root_spans:
                    dfs(root, progress, pbar_eval_id)

            end_time = time.perf_counter()
            run_duration = end_time - start_time
            # Update test run
            api_test_case.update_run_duration(run_duration)
            test_run_manager.update_test_run(api_test_case, test_case)
            main_result = create_test_result(api_test_case)
            trace_results = extract_trace_test_results(trace_api)
            unique_trace_results = filter_duplicate_results(
                main_result, trace_results
            )
            test_results.append(main_result)
            test_results.extend(unique_trace_results)

            update_pbar(progress, pbar_id)

        # Post-iteration guard: refuse a run that ran with no metric source
        # at any level. Must happen AFTER the for-loop since span-level
        # @observe metrics only become visible after user code has run.
        if not _has_any_evaluable_metrics(
            trace_metrics=trace_metrics,
            traces=processed_traces,
            test_case_metrics=trace_manager.eval_session.test_case_metrics,
        ):
            _raise_no_metrics_error()

    try:
        if display_config.show_indicator and _use_bar_indicator:
            progress = Progress(
                TextColumn("{task.description}"),
                BarColumn(bar_width=60),
                TaskProgressColumn(),
                TimeElapsedColumn(),
                console=custom_console,
            )
            with progress:
                pbar_id = add_pbar(
                    progress,
                    "Running Component-Level Evals (sync)",
                    total=len(goldens) * 2,
                )
                yield from evaluate_test_cases(
                    progress=progress, pbar_id=pbar_id
                )
        else:
            yield from evaluate_test_cases()
    except Exception:
        raise
    finally:
        # Atomic exit cleanup: replacing the session resets mode + every
        # per-run collection in a single assignment, so state can't leak
        # into the next run.
        local_trace_manager.eval_session = EvalSession()


def a_execute_agentic_test_cases_from_loop(
    goldens: List[Golden],
    trace_metrics: Optional[List[BaseMetric]],
    test_results: List[TestResult],
    loop: asyncio.AbstractEventLoop,
    display_config: Optional[DisplayConfig] = DisplayConfig(),
    cache_config: Optional[CacheConfig] = CacheConfig(),
    error_config: Optional[ErrorConfig] = ErrorConfig(),
    async_config: Optional[AsyncConfig] = AsyncConfig(),
    identifier: Optional[str] = None,
    _use_bar_indicator: bool = True,
    _is_assert_test: bool = False,
) -> Iterator[TestResult]:

    semaphore = asyncio.Semaphore(async_config.max_concurrent)
    original_create_task = asyncio.create_task

    test_run_manager = global_test_run_manager
    test_run_manager.save_to_disk = cache_config.write_cache
    test_run = test_run_manager.get_test_run(identifier=identifier)

    local_trace_manager = trace_manager
    local_trace_manager.eval_session = EvalSession(mode=EvalMode.ITERATOR_ASYNC)

    async def execute_callback_with_semaphore(coroutine: Awaitable):
        async with semaphore:
            return await _await_with_outer_deadline(
                coroutine, timeout=get_per_task_timeout_seconds()
            )

    def evaluate_test_cases(
        progress: Optional[Progress] = None,
        pbar_id: Optional[int] = None,
        pbar_callback_id: Optional[int] = None,
    ):
        # Tasks we scheduled during this iterator run on this event loop.
        # by gathering these tasks we can avoid re-awaiting coroutines which
        # can cause cross loop mixups that trigger "future belongs to a different loop" errors
        created_tasks: list[asyncio.Task] = []
        task_meta: dict[asyncio.Task, dict] = {}
        current_golden_ctx = {"index": -1, "name": None, "input": None}

        def create_callback_task(coro, **kwargs):
            # build a descriptive task name for tracking
            coro_desc = repr(coro)
            task_name = f"callback[{current_golden_ctx['index']}]:{coro_desc.split()[1] if ' ' in coro_desc else coro_desc}"

            # Wrap the user coroutine in our semaphore runner and bind it to THIS loop.
            # Keep the resulting Task so we can gather tasks (not raw coroutines) later,
            # without touching tasks from other loops or already awaited coroutines.
            task = loop.create_task(
                execute_callback_with_semaphore(coro), name=task_name
            )

            # record metadata for debugging
            started = time.perf_counter()
            short_input = current_golden_ctx.get("input")
            if isinstance(short_input, str):
                short_input = shorten(short_input, len_medium())

            task_meta[task] = {
                "golden_index": current_golden_ctx["index"],
                "golden_name": current_golden_ctx["name"],
                "input": short_input,
                "coro": coro_desc,
                "started": started,
            }

            def on_task_done(t: asyncio.Task):
                cancelled = False
                exc = None
                trace = None
                root = None
                resolved_trace_from_task = False
                resolved_root_from_task = False

                # Task.exception() raises CancelledError if task was cancelled
                try:
                    exc = t.exception()
                except asyncio.CancelledError:
                    cancelled = True
                    exc = None

                meta = task_meta.get(t, {})
                golden_index = meta.get("golden_index")

                if golden_index is not None and 0 <= golden_index < len(
                    goldens
                ):
                    golden = goldens[golden_index]

                    def _mark_trace_error(trace, root, msg: str):
                        now = time.perf_counter()
                        trace.status = TraceSpanStatus.ERRORED
                        # Close the trace so the API layer has a proper endTime
                        if trace.end_time is None:
                            trace.end_time = now
                        if root:
                            root.status = TraceSpanStatus.ERRORED
                            root.error = msg
                            if root.end_time is None:
                                root.end_time = now

                    if exc is not None:
                        msg = format_error_text(exc)
                        trace, root = _resolve_trace_and_root_for_task(t)
                        resolved_trace_from_task = bool(trace)
                        resolved_root_from_task = bool(root)
                        if trace:
                            _mark_trace_error(trace, root, msg)
                        else:
                            for (
                                trace
                            ) in trace_manager.eval_session.traces_to_evaluate:
                                if (
                                    trace_manager.eval_session.trace_uuid_to_golden.get(
                                        trace.uuid
                                    )
                                    is golden
                                ):
                                    root = _pick_root_for_marking(trace)
                                    _mark_trace_error(trace, root, msg)
                                    break

                    elif cancelled or t.cancelled():
                        cancel_exc = DeepEvalError(
                            "Task was cancelled (likely due to timeout)."
                        )
                        msg = format_error_text(cancel_exc)
                        trace, root = _resolve_trace_and_root_for_task(t)
                        resolved_trace_from_task = bool(trace)
                        resolved_root_from_task = bool(root)
                        if trace:
                            _mark_trace_error(trace, root, msg)
                        else:
                            for (
                                trace
                            ) in trace_manager.eval_session.traces_to_evaluate:
                                if (
                                    trace_manager.eval_session.trace_uuid_to_golden.get(
                                        trace.uuid
                                    )
                                    is golden
                                ):
                                    root = _pick_root_for_marking(trace)
                                    _mark_trace_error(trace, root, msg)
                                    break

                if get_settings().DEEPEVAL_DEBUG_ASYNC:
                    # Using info level here to make it easy to spot these logs.
                    golden_name = meta.get("golden_name")
                    duration = time.perf_counter() - meta.get(
                        "started", started
                    )

                    if cancelled or exc is not None:
                        if not resolved_trace_from_task:
                            logger.warning(
                                "[deepeval] on_task_done: no binding for task; falling back to golden->trace. task=%s golden=%r",
                                t.get_name(),
                                golden_name,
                            )
                        elif not resolved_root_from_task:
                            logger.warning(
                                "[deepeval] on_task_done: bound trace found but no bound root; using heuristic. task=%s trace=%s",
                                t.get_name(),
                                trace.uuid,
                            )

                    if cancelled:
                        logger.info(
                            "[deepeval] task CANCELLED %s after %.2fs meta=%r",
                            t.get_name(),
                            duration,
                            meta,
                        )
                    elif exc is not None:

                        show_trace = bool(
                            get_settings().DEEPEVAL_LOG_STACK_TRACES
                        )
                        exc_info = (
                            (
                                type(exc),
                                exc,
                                getattr(exc, "__traceback__", None),
                            )
                            if show_trace
                            else None
                        )
                        logger.error(
                            "[deepeval] task ERROR %s after %.2fs meta=%r",
                            t.get_name(),
                            duration,
                            meta,
                            exc_info=exc_info,
                        )
                    else:
                        logger.info(
                            "[deepeval] task OK %s after %.2fs meta={'golden_index': %r}",
                            t.get_name(),
                            duration,
                            meta.get("golden_index"),
                        )

                try:
                    trace_manager.task_bindings.pop(t, None)
                except Exception:
                    pass
                update_pbar(progress, pbar_callback_id)
                update_pbar(progress, pbar_id)

            task.add_done_callback(on_task_done)
            created_tasks.append(task)
            return task

        asyncio.create_task = create_callback_task
        # DEBUG
        # Snapshot tasks that already exist on this loop so we can detect strays
        baseline_tasks = loop.run_until_complete(_snapshot_tasks())

        try:
            for index, golden in enumerate(goldens):
                token = set_current_golden(golden)
                current_golden_ctx.update(
                    {
                        "index": index,
                        "name": getattr(golden, "name", None),
                        "input": getattr(golden, "input", None),
                    }
                )
                prev_task_length = len(created_tasks)
                try:
                    yield golden
                finally:
                    reset_current_golden(token)
                # if this golden created no tasks, bump bars now
                if len(created_tasks) == prev_task_length:
                    update_pbar(progress, pbar_callback_id)
                    update_pbar(progress, pbar_id)
        finally:
            asyncio.create_task = original_create_task

        if created_tasks:
            # Only await tasks we created on this loop in this run.
            # This will prevent re-awaiting and avoids cross loop "future belongs to a different loop" errors
            try:
                loop.run_until_complete(
                    asyncio.wait_for(
                        asyncio.gather(*created_tasks, return_exceptions=True),
                        timeout=get_gather_timeout(),
                    )
                )

            except (asyncio.TimeoutError, TimeoutError) as e:
                import traceback

                settings = get_settings()
                pending = [t for t in created_tasks if not t.done()]

                _log_gather_timeout(logger, exc=e, pending=len(pending))

                # Log the elapsed time for each task that was pending
                for t in pending:
                    meta = task_meta.get(t, {})
                    start_time = meta.get("started", time.perf_counter())
                    elapsed_time = time.perf_counter() - start_time

                    # Determine if it was a per task or gather timeout based on task's elapsed time
                    if not settings.DEEPEVAL_DISABLE_TIMEOUTS:
                        timeout_type = (
                            "per-task"
                            if elapsed_time >= get_per_task_timeout_seconds()
                            else "gather"
                        )
                        logger.info(
                            "  - PENDING %s elapsed_time=%.2fs timeout_type=%s meta=%s",
                            t.get_name(),
                            elapsed_time,
                            timeout_type,
                            meta,
                        )
                    else:
                        logger.info(
                            "  - PENDING %s elapsed_time=%.2fs meta=%s",
                            t.get_name(),
                            elapsed_time,
                            meta,
                        )

                    if loop.get_debug() and get_settings().DEEPEVAL_DEBUG_ASYNC:
                        frames = t.get_stack(limit=6)
                        if frames:
                            logger.info("    stack:")
                            for fr in frames:
                                for line in traceback.format_stack(fr):
                                    logger.info("      " + line.rstrip())

                # Cancel and drain the tasks
                for t in pending:
                    t.cancel()
                loop.run_until_complete(
                    asyncio.gather(*created_tasks, return_exceptions=True)
                )
            finally:

                # if it is already closed, we are done
                if loop.is_closed():
                    return

                try:
                    current_tasks = set()
                    # Find tasks that were created during this run but we didn’t track
                    current_tasks = loop.run_until_complete(_snapshot_tasks())
                except RuntimeError:
                    # this might happen if the loop is already closing
                    pass

                leftovers = [
                    t
                    for t in current_tasks
                    if t not in baseline_tasks
                    and t not in created_tasks
                    and not t.done()
                ]

                if get_settings().DEEPEVAL_DEBUG_ASYNC:
                    if len(leftovers) > 0:
                        logger.warning(
                            "[deepeval] %d stray task(s) not tracked; cancelling...",
                            len(leftovers),
                        )
                    for t in leftovers:
                        meta = task_meta.get(t, {})
                        name = t.get_name()
                        logger.warning("  - STRAY %s meta=%s", name, meta)

                if leftovers:
                    for t in leftovers:
                        t.cancel()

                    # Drain strays so they don’t leak into the next iteration
                    try:
                        loop.run_until_complete(
                            asyncio.gather(*leftovers, return_exceptions=True)
                        )
                    except RuntimeError:
                        # If the loop is closing here, just continue
                        if get_settings().DEEPEVAL_DEBUG_ASYNC:
                            logger.warning(
                                "[deepeval] failed to drain stray tasks because loop is closing"
                            )

        # Pre-evaluation guard: refuse a run that has no metric source.
        # Lazy check is the only correct option because span-level metrics
        # on @observe-decorated functions only become visible after user
        # code has actually run.
        session = trace_manager.eval_session
        if not _has_any_evaluable_metrics(
            trace_metrics=trace_metrics,
            traces=session.traces_to_evaluate,
            test_case_metrics=session.test_case_metrics,
        ):
            _raise_no_metrics_error()

        # Evaluate traces
        if trace_manager.eval_session.traces_to_evaluate:
            loop.run_until_complete(
                _a_evaluate_traces(
                    traces_to_evaluate=trace_manager.eval_session.traces_to_evaluate,
                    goldens=goldens,
                    test_run_manager=test_run_manager,
                    test_results=test_results,
                    trace_metrics=trace_metrics,
                    verbose_mode=display_config.verbose_mode,
                    ignore_errors=error_config.ignore_errors,
                    skip_on_missing_params=error_config.skip_on_missing_params,
                    show_indicator=display_config.show_indicator,
                    throttle_value=async_config.throttle_value,
                    max_concurrent=async_config.max_concurrent,
                    _use_bar_indicator=_use_bar_indicator,
                    _is_assert_test=_is_assert_test,
                    progress=progress,
                    pbar_id=pbar_id,
                )
            )
        elif trace_manager.eval_session.test_case_metrics:
            loop.run_until_complete(
                _evaluate_test_case_pairs(
                    test_case_pairs=trace_manager.eval_session.test_case_metrics,
                    test_run=test_run,
                    test_run_manager=test_run_manager,
                    test_results=test_results,
                    ignore_errors=error_config.ignore_errors,
                    skip_on_missing_params=error_config.skip_on_missing_params,
                    show_indicator=display_config.show_indicator,
                    verbose_mode=display_config.verbose_mode,
                    throttle_value=async_config.throttle_value,
                    max_concurrent=async_config.max_concurrent,
                    _use_bar_indicator=_use_bar_indicator,
                    _is_assert_test=_is_assert_test,
                    progress=progress,
                    pbar_id=pbar_id,
                )
            )

    try:
        if display_config.show_indicator and _use_bar_indicator:
            progress = Progress(
                TextColumn("{task.description}"),
                BarColumn(bar_width=60),
                TaskProgressColumn(),
                TimeElapsedColumn(),
                console=custom_console,
            )
            with progress:
                pbar_id = add_pbar(
                    progress,
                    "Running Component-Level Evals (async)",
                    total=len(goldens) * 2,
                )
                pbar_callback_id = add_pbar(
                    progress,
                    f"\t⚡ Calling LLM app (with {len(goldens)} goldens)",
                    total=len(goldens),
                )
                yield from evaluate_test_cases(
                    progress=progress,
                    pbar_id=pbar_id,
                    pbar_callback_id=pbar_callback_id,
                )
        else:
            yield from evaluate_test_cases()
    except Exception:
        raise
    finally:
        # Atomic exit cleanup: replacing the session resets mode + every
        # per-run collection in a single assignment.
        local_trace_manager.eval_session = EvalSession()


async def _a_evaluate_traces(
    traces_to_evaluate: List[Trace],
    goldens: List[Golden],
    test_run_manager: TestRunManager,
    test_results: List[TestResult],
    verbose_mode: Optional[bool],
    ignore_errors: bool,
    skip_on_missing_params: bool,
    show_indicator: bool,
    _use_bar_indicator: bool,
    _is_assert_test: bool,
    progress: Optional[Progress],
    pbar_id: Optional[int],
    throttle_value: int,
    max_concurrent: int,
    trace_metrics: Optional[List[BaseMetric]],
):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def execute_evals_with_semaphore(func: Callable, *args, **kwargs):
        async with semaphore:
            return await _await_with_outer_deadline(
                func, *args, timeout=get_per_task_timeout_seconds(), **kwargs
            )

    eval_tasks = []
    # Here, we will work off a fixed-set copy to avoid surprises from potential
    # mid-iteration mutation
    traces_snapshot = list(traces_to_evaluate or [])

    for count, trace in enumerate(traces_snapshot):
        # Prefer the explicit mapping from trace -> golden captured at trace creation.
        golden = trace_manager.eval_session.trace_uuid_to_golden.get(trace.uuid)
        if not golden:
            # trace started during the iterator run but the CURRENT_GOLDEN was
            # not set for some reason. We can’t map it to a golden, so the best
            # we can do is skip evaluation for this trace.
            if (
                logger.isEnabledFor(logging.DEBUG)
                and get_settings().DEEPEVAL_VERBOSE_MODE
            ):
                logger.debug(
                    "Skipping trace %s: no golden association found in eval_session",
                    trace.uuid,
                )
            continue
        copied_trace_metrics: Optional[List[BaseMetric]] = None
        if trace_metrics:
            copied_trace_metrics = copy_metrics(trace_metrics)
        with capture_evaluation_run("golden"):
            task = execute_evals_with_semaphore(
                func=_a_execute_agentic_test_case,
                golden=golden,
                trace=trace,
                test_run_manager=test_run_manager,
                test_results=test_results,
                count=count,
                verbose_mode=verbose_mode,
                ignore_errors=ignore_errors,
                skip_on_missing_params=skip_on_missing_params,
                show_indicator=show_indicator,
                _use_bar_indicator=_use_bar_indicator,
                _is_assert_test=_is_assert_test,
                progress=progress,
                pbar_id=pbar_id,
                trace_metrics=copied_trace_metrics,
            )
            eval_tasks.append(asyncio.create_task(task))
            await asyncio.sleep(throttle_value)

    try:
        await asyncio.wait_for(
            asyncio.gather(*eval_tasks),
            timeout=get_gather_timeout(),
        )
    except (asyncio.TimeoutError, TimeoutError):
        for t in eval_tasks:
            if not t.done():
                t.cancel()
        await asyncio.gather(*eval_tasks, return_exceptions=True)
        raise