From 2790915b4970d08d1842e67c3fb0b75ba6875b7a Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Thu, 3 Jul 2025 10:51:10 -0400
Subject: [PATCH 01/26] chore: Remove promptflow dependencies

---
 sdk/evaluation/azure-ai-evaluation/dev_requirements.txt | 3 ++-
 sdk/evaluation/azure-ai-evaluation/setup.py             | 2 --
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt b/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt
index 08c80c0f929b..1ccece0d2952 100644
--- a/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt
+++ b/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt
@@ -9,4 +9,5 @@ pytest-xdist
 azure-ai-inference>=1.0.0b4
 azure-ai-projects<=1.0.0b10
 aiohttp
--e ../azure-ai-evaluation
\ No newline at end of file
+filelock
+-e ../azure-ai-evaluation
diff --git a/sdk/evaluation/azure-ai-evaluation/setup.py b/sdk/evaluation/azure-ai-evaluation/setup.py
index 81244f026ace..31e23067eb41 100644
--- a/sdk/evaluation/azure-ai-evaluation/setup.py
+++ b/sdk/evaluation/azure-ai-evaluation/setup.py
@@ -66,8 +66,6 @@
     ),
     python_requires=">=3.9",
     install_requires=[
-        "promptflow-devkit>=1.17.1",
-        "promptflow-core>=1.17.1",
         "pyjwt>=2.8.0",
         # pickle support for credentials was added to this release
         "azure-identity>=1.16.0",

From a3dfa6e951e5fbf795c14c48720baa5ce717deb2 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Wed, 16 Jul 2025 08:18:44 -0400
Subject: [PATCH 02/26] fix: Filter inputs before calling function in Batch
 Engine

    Added to match behavior in promptflow

    see also:

        https://github.com/microsoft/promptflow/blob/5e6c183474c0a2575bb416d18201e4f9fd562b2e/src/promptflow-core/promptflow/executor/_script_executor.py#L162
---
 .../ai/evaluation/_legacy/_batch_engine/_engine.py  | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index 9205504953b9..d0bed1913908 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -283,6 +283,13 @@ async def create_under_semaphore(index: int, inputs: Mapping[str, Any]):
                 # TODO ralphe: set logger to use here
             )
 
+    def __preprocess_inputs(self, inputs: Mapping[str, Any]) -> Mapping[str, Any]:
+
+        func_params = inspect.signature(self._func).parameters
+
+        filtered_params = {key: value for key, value in inputs.items() if key in func_params}
+        return filtered_params
+
     async def _exec_line_async(
         self,
         run_id: str,
@@ -313,13 +320,15 @@ async def _exec_line_async(
                     #       For now we will just run the function in the current process, but in the future we may
                     #       want to consider running the function in a separate process for isolation reasons.
                     output: Any
+
+                    processed_inputs = self.__preprocess_inputs(inputs)
                     if is_async_callable(self._func):
-                        output = await self._func(**inputs)
+                        output = await self._func(**processed_inputs)
                     else:
                         # to maximize the parallelism, we run the synchronous function in a separate thread
                         # and await its result
                         output = await asyncio.get_event_loop().run_in_executor(
-                            self._executor, partial(self._func, **inputs)
+                            self._executor, partial(self._func, **processed_inputs)
                         )
 
                     # This should in theory never happen but as an extra precaution, let's check if the output

From 85eecc5a3559071918598fef7546fac8afdb5d84 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Wed, 16 Jul 2025 18:50:18 -0400
Subject: [PATCH 03/26] refactor: Use enumerate instead of manually keeping
 track of line number

---
 .../azure/ai/evaluation/_legacy/_batch_engine/_engine.py    | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index d0bed1913908..181390608859 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -123,11 +123,9 @@ def _apply_column_mapping(
         data = data[:max_lines] if max_lines else data
 
         inputs: Sequence[Mapping[str, Any]] = []
-        line: int = 0
         defaults = cast(Mapping[str, Any], column_mapping.get(DEFAULTS_KEY, {}))
 
-        for input in data:
-            line += 1
+        for line_number, input in enumerate(data, start=1):
             mapped: Dict[str, Any] = {}
             missing_inputs: Set[str] = set()
 
@@ -159,7 +157,7 @@ def _apply_column_mapping(
 
             if missing_inputs:
                 missing = ", ".join(missing_inputs)
-                raise BatchEngineValidationError(f"Missing inputs for line {line}: '{missing}'")
+                raise BatchEngineValidationError(f"Missing inputs for line {line_number}: '{missing}'")
 
             inputs.append(mapped)
 

From cafe8bac8299dafa5fc8350e2ef1033cb50ec204 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Thu, 17 Jul 2025 10:13:32 -0400
Subject: [PATCH 04/26] fix,refactor: Unconditionally inject default column
 mapping from data -> params

         In promptflow's logic for applying a column mapping to data,
         it will unconditionally inject a mapping from function
         parameter do data of the same name:

                  https://github.com/microsoft/promptflow/blob/3e297112a2c142caf7c185bcba644d0f66422539/src/promptflow-devkit/promptflow/batch/_batch_inputs_processor.py#L110-L141

         This behavior deviated from the existing logic in this SDK,
         where generating that mapping was conditional on the user
         not providing a column mapping:

                  https://github.com/Azure/azure-sdk-for-python/blob/f3740540eb5b3d22dc1bccba0eb00b652b124d5f/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py#L51-L52

         This deviation caused one of the parameterized cases of
         `test_evaluate_another_questions` to fail, because there
         was a user provided column mapping that mapped to a parameter
         not present in the evaluator, so the lack of a default mapping
         caused the evaluation to fail because the required parameter
         was missing

                  https://github.com/Azure/azure-sdk-for-python/blob/f3740540eb5b3d22dc1bccba0eb00b652b124d5f/sdk/evaluation/azure-ai-evaluation/tests/e2etests/test_evaluate.py#L216

         This commit aligns the application of the column mapping in the SDK
         more closely to the promptflow implementation
---
 .../_legacy/_batch_engine/_engine.py          | 64 +++++++++++++++++--
 .../evaluation/_legacy/_batch_engine/_run.py  |  4 +-
 .../_legacy/_batch_engine/_run_submitter.py   | 28 ++------
 3 files changed, 64 insertions(+), 32 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index 181390608859..63934c1d6a68 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -20,7 +20,21 @@
 from functools import partial
 from contextlib import contextmanager
 from datetime import datetime, timezone
-from typing import Any, Callable, Dict, Final, Generator, Mapping, MutableMapping, Optional, Sequence, Set, Tuple, cast
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Final,
+    Generator,
+    Mapping,
+    MutableMapping,
+    Optional,
+    Sequence,
+    Set,
+    Tuple,
+    cast,
+    Literal,
+)
 from uuid import uuid4
 
 from ._utils import DEFAULTS_KEY, get_int_env_var, get_value_from_path, is_async_callable
@@ -85,15 +99,13 @@ def __init__(
     async def run(
         self,
         data: Sequence[Mapping[str, Any]],
-        column_mapping: Mapping[str, str],
+        column_mapping: Optional[Mapping[str, str]],
         *,
         id: Optional[str] = None,
         max_lines: Optional[int] = None,
     ) -> BatchResult:
         if not data:
             raise BatchEngineValidationError("Please provide a non-empty data mapping.")
-        if not column_mapping:
-            raise BatchEngineValidationError("The column mapping is required.")
 
         start_time = datetime.now(timezone.utc)
 
@@ -114,12 +126,52 @@ def cancel(self):
         # TODO ralphe: Make sure this works
         self._is_canceled = True
 
-    @staticmethod
     def _apply_column_mapping(
+        self,
         data: Sequence[Mapping[str, Any]],
-        column_mapping: Mapping[str, str],
+        column_mapping: Optional[Mapping[str, str]],
         max_lines: Optional[int],
     ) -> Sequence[Mapping[str, str]]:
+
+        resolved_column_mapping: Mapping[str, str] = self._resolve_column_mapping(column_mapping)
+        resolved_column_mapping.update(self._generate_defaults_for_column_mapping())
+        return self._apply_column_mapping_to_lines(data, resolved_column_mapping, max_lines)
+
+    def _resolve_column_mapping(
+        self,
+        column_mapping: Optional[Mapping[str, str]],
+    ) -> Mapping[str, str]:
+        parameters = inspect.signature(self._func).parameters
+        default_column_mapping: Dict[str, str] = {
+            name: f"${{data.{name}}}"
+            for name, value in parameters.items()
+            if name not in ["self", "cls", "args", "kwargs"]
+        }
+        resolved_mapping: Dict[str, str] = default_column_mapping.copy()
+
+        for name, value in parameters.items():
+            if value and value.default is not inspect.Parameter.empty:
+                resolved_mapping.pop(name)
+
+        resolved_mapping.update(column_mapping or {})
+        return resolved_mapping
+
+    def _generate_defaults_for_column_mapping(self) -> Mapping[Literal["$defaults$"], Any]:
+
+        return {
+            DEFAULTS_KEY: {
+                name: value.default
+                for name, value in inspect.signature(self._func).parameters.items()
+                if value.default is not inspect.Parameter.empty
+            }
+        }
+
+    @staticmethod
+    def _apply_column_mapping_to_lines(
+        data: Sequence[Mapping[str, Any]],
+        column_mapping: Mapping[str, str],
+        max_lines: Optional[int],
+    ) -> Sequence[Mapping[str, Any]]:
         data = data[:max_lines] if max_lines else data
 
         inputs: Sequence[Mapping[str, Any]] = []
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run.py
index fad4245d13ef..dce0e1f35399 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run.py
@@ -58,7 +58,7 @@ def __init__(
         dynamic_callable: Callable,
         name_prefix: Optional[str],
         inputs: Sequence[Mapping[str, Any]],
-        column_mapping: Mapping[str, str],
+        column_mapping: Optional[Mapping[str, str]] = None,
         created_on: Optional[datetime] = None,
         run: Optional["Run"] = None,
     ):
@@ -70,7 +70,7 @@ def __init__(
         self.dynamic_callable = dynamic_callable
         self.name = self._generate_run_name(name_prefix, self._created_on)
         self.inputs = inputs
-        self.column_mapping = column_mapping
+        self.column_mapping: Optional[Mapping[str, str]] = column_mapping
         self.result: Optional[BatchResult] = None
         self.metrics: Mapping[str, Any] = {}
         self._run = run
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
index 8b5186271515..9933ca91aab3 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
@@ -46,11 +46,6 @@ async def submit(
         **kwargs,
     ) -> Run:
 
-        # if the column mappings are not provided, generate them based on the arguments to the
-        # flow function.
-        if column_mapping is None:
-            column_mapping = self._generate_column_mapping(dynamic_callable)
-
         # The old code always spun up two threads here using a ThreadPoolExecutor:
         # 1. One thread essentially did nothing of value (since tracing was disabled, and we
         #    don't care about checking for the latest PromptFlow version number now)
@@ -173,31 +168,16 @@ async def _submit_bulk_run(self, run: Run, local_storage: AbstractRunStorage, **
             run.metrics = system_metrics
             run.result = batch_result
 
-    @staticmethod
-    def _generate_column_mapping(function: Callable) -> Mapping[str, Any]:
-        args = inspect.signature(function).parameters
-        default_values: Dict[str, Any] = {}
-        mapping: Dict[str, Any] = {}
-        for key, value in args.items():
-            if key in ["self", "cls"] or value.kind in [value.VAR_POSITIONAL, value.VAR_KEYWORD]:
-                continue
-
-            mapping[key] = f"${{data.{key}}}"
-            if value.default != inspect.Parameter.empty:
-                default_values[key] = value.default
-
-        return {
-            **mapping,
-            DEFAULTS_KEY: default_values,
-        }
-
     @staticmethod
     def _validate_inputs(run: Run):
         if not run.inputs and not run.previous_run:
             raise BatchEngineValidationError("Either data, or a previous run must be specified for the evaluation run.")
 
     @staticmethod
-    def _validate_column_mapping(column_mapping: Mapping[str, str]):
+    def _validate_column_mapping(column_mapping: Optional[Mapping[str, str]]):
+        if not column_mapping:
+            return
+
         if not isinstance(column_mapping, Mapping):
             raise BatchEngineValidationError(f"Column mapping must be a dict, got {type(column_mapping)}.")
 

From 1e3d7f5db80998f94761440ef499bccc7f5e62ce Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Fri, 18 Jul 2025 10:22:23 -0400
Subject: [PATCH 05/26] refactor: Don't shadow `value` variable in
 apply_column_mapping_to_lines

---
 .../azure/ai/evaluation/_legacy/_batch_engine/_engine.py    | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index 63934c1d6a68..8564104f4b4d 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -198,12 +198,12 @@ def _apply_column_mapping_to_lines(
                     continue
 
                 dict_path = match.group(1)
-                found, value = get_value_from_path(dict_path, input)
+                found, mapped_value = get_value_from_path(dict_path, input)
                 if not found:  # try default value
-                    found, value = get_value_from_path(dict_path, defaults)
+                    found, mapped_value = get_value_from_path(dict_path, defaults)
 
                 if found:
-                    mapped[key] = value
+                    mapped[key] = mapped_value
                 else:
                     missing_inputs.add(dict_path)
 

From 1d1882a347a6d53d329275e6f97c5406ac52e7cd Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Fri, 18 Jul 2025 13:57:54 -0400
Subject: [PATCH 06/26] feat: Add support for running aggregations in
 RunSubmitterClient

---
 .../_batch_run/_run_submitter_client.py       | 30 +++++++++++++++++--
 1 file changed, 28 insertions(+), 2 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
index ffd169119c62..f7badb3b0632 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
@@ -16,8 +16,9 @@
 from ..._legacy._batch_engine._config import BatchEngineConfig
 from ..._legacy._batch_engine._run import Run
 from ..._legacy._adapters._constants import LINE_NUMBER
+from ..._legacy._adapters.types import AttrDict
 from ..._legacy._common._thread_pool_executor_with_context import ThreadPoolExecutorWithContext
-
+from ..._evaluate._utils import _has_aggregator
 
 LOGGER = logging.getLogger(__name__)
 
@@ -97,7 +98,32 @@ def _update(prefix: str, items: Sequence[Mapping[str, Any]]) -> None:
 
     def get_metrics(self, client_run: BatchClientRun) -> Dict[str, Any]:
         run = self._get_run(client_run)
-        return dict(run.metrics)
+        return {**run.metrics, **self._get_aggregated_metrics(client_run)}
+
+    def _get_aggregated_metrics(self, client_run: BatchClientRun) -> Dict[str, Any]:
+        aggregated_metrics = None
+        run = self._get_run(client_run)
+        try:
+            if _has_aggregator(run.dynamic_callable):
+                result_df = pd.DataFrame(run.outputs)
+                if len(result_df.columns) == 1 and result_df.columns[0] == "output":
+                    aggregate_input = result_df["output"].tolist()
+                else:
+                    aggregate_input = [AttrDict(item) for item in result_df.to_dict("records")]
+
+                aggr_func = getattr(run.dynamic_callable, "__aggregate__")
+                aggregated_metrics = aggr_func(aggregate_input)
+
+        except Exception as ex:  # pylint: disable=broad-exception-caught
+            LOGGER.warning("Error calculating aggregations for evaluator, failed with error %s", ex)
+
+        if not isinstance(aggregated_metrics, dict):
+            LOGGER.warning(
+                "Aggregated metrics for evaluator is not a dictionary will not be logged as metrics",
+            )
+            return {}
+
+        return aggregated_metrics
 
     def get_run_summary(self, client_run: BatchClientRun) -> Dict[str, Any]:
         run = self._get_run(client_run)

From 53f6a3ed774942660de77f68d9abfd8e27d2945c Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Tue, 22 Jul 2025 14:01:15 -0400
Subject: [PATCH 07/26] tests,fix: Don't log duration as a metric

    Breaks a tests that checks for strict equality of metrics
---
 .../azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
index 9933ca91aab3..6fb5f9cea9f4 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
@@ -158,7 +158,7 @@ async def _submit_bulk_run(self, run: Run, local_storage: AbstractRunStorage, **
                 system_metrics.update(dataclasses.asdict(batch_result.tokens))  # token related
                 system_metrics.update(
                     {
-                        "duration": batch_result.duration.total_seconds(),
+                        # "duration": batch_result.duration.total_seconds(),
                         # "__pf__.lines.completed": batch_result.total_lines - batch_result.failed_lines,
                         # "__pf__.lines.failed": batch_result.failed_lines,
                     }

From 40f7452595e0b5d30f67a8f7f31f1b9d9282cf90 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Wed, 23 Jul 2025 21:48:24 -0400
Subject: [PATCH 08/26] refactor: Rewrite RunSubmitterClient.get_details

---
 .../_batch_run/_run_submitter_client.py       | 45 ++++++++++---------
 1 file changed, 25 insertions(+), 20 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
index f7badb3b0632..7b0ca71ff8c9 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
@@ -6,6 +6,7 @@
 import logging
 import pandas as pd
 import sys
+import itertools
 from collections import defaultdict
 from concurrent.futures import Future
 from os import PathLike
@@ -19,6 +20,7 @@
 from ..._legacy._adapters.types import AttrDict
 from ..._legacy._common._thread_pool_executor_with_context import ThreadPoolExecutorWithContext
 from ..._evaluate._utils import _has_aggregator
+from ..._constants import Prefixes
 
 LOGGER = logging.getLogger(__name__)
 
@@ -73,28 +75,31 @@ def run(
         return run_future
 
     def get_details(self, client_run: BatchClientRun, all_results: bool = False) -> pd.DataFrame:
-
         run = self._get_run(client_run)
 
-        data: Dict[str, List[Any]] = defaultdict(list)
-        stop_at: Final[int] = self._config.default_num_results if not all_results else sys.maxsize
-
-        def _update(prefix: str, items: Sequence[Mapping[str, Any]]) -> None:
-            for i, line in enumerate(items):
-                if i >= stop_at:
-                    break
-                for k, value in line.items():
-                    key = f"{prefix}.{k}"
-                    data[key].append(value)
-
-        # Go from a list of dictionaries (i.e. a row view of the data) to a dictionary of lists
-        # (i.e. a column view of the data)
-        _update("inputs", run.inputs)
-        _update("inputs", [{LINE_NUMBER: i} for i in range(len(run.inputs))])
-        _update("outputs", run.outputs)
-
-        df = pd.DataFrame(data).reindex(columns=[k for k in data.keys()])
-        return df
+        def concat(*dataframes: pd.DataFrame) -> pd.DataFrame:
+            return pd.concat(dataframes, axis=1, verify_integrity=True)
+
+        def to_dataframe(items: Sequence[Mapping[str, Any]], *, max_length: Optional[int] = None) -> pd.DataFrame:
+            """Convert a sequence of dictionaries to a DataFrame.
+
+            :param items: Sequence of dictionaries to convert.
+            :type items: Sequence[Mapping[str, Any]]
+            :param max_length: Maximum number of items to include in the DataFrame. If None, include all items.
+            :type max_length: Optional[int]
+            :return: DataFrame containing the items.
+            :rtype: pd.DataFrame
+            """
+            max_length = None if all_results else self._config.default_num_results
+            return pd.DataFrame(data=items if all_results else itertools.islice(items, max_length))
+
+        inputs = concat(
+            to_dataframe(run.inputs), to_dataframe([{LINE_NUMBER: i} for i in range(len(run.inputs))])
+        ).add_prefix(Prefixes.INPUTS)
+
+        outputs = to_dataframe(run.outputs).add_prefix(Prefixes.OUTPUTS)
+
+        return concat(inputs, outputs)
 
     def get_metrics(self, client_run: BatchClientRun) -> Dict[str, Any]:
         run = self._get_run(client_run)

From 00d8b2f002d707ef6251ca8af0d136f16f986a58 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Fri, 25 Jul 2025 08:45:59 -0400
Subject: [PATCH 09/26] fix: Correct the typing of is_onedp_project

---
 .../azure/ai/evaluation/_common/utils.py             | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/utils.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/utils.py
index 9de0bc9c34d2..2c45f37bc15a 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/utils.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_common/utils.py
@@ -6,11 +6,11 @@
 import re
 import math
 import threading
-from typing import Any, List, Literal, Mapping, Type, TypeVar, Tuple, Union, cast, get_args, get_origin
+from typing import Any, List, Literal, Mapping, Optional, Type, TypeVar, Tuple, Union, cast, get_args, get_origin
 
 import nltk
 from azure.storage.blob import ContainerClient
-from typing_extensions import NotRequired, Required, TypeGuard
+from typing_extensions import NotRequired, Required, TypeGuard, TypeIs
 from azure.ai.evaluation._legacy._adapters._errors import MissingRequiredPackage
 from azure.ai.evaluation._constants import AZURE_OPENAI_TYPE, OPENAI_TYPE
 from azure.ai.evaluation._exceptions import ErrorMessage, ErrorBlame, ErrorCategory, ErrorTarget, EvaluationException
@@ -127,17 +127,15 @@ def construct_prompty_model_config(
     return prompty_model_config
 
 
-def is_onedp_project(azure_ai_project: AzureAIProject) -> bool:
+def is_onedp_project(azure_ai_project: Optional[Union[str, AzureAIProject]]) -> TypeIs[str]:
     """Check if the Azure AI project is an OneDP project.
 
     :param azure_ai_project: The scope of the Azure AI project.
-    :type azure_ai_project: ~azure.ai.evaluation.AzureAIProject
+    :type azure_ai_project: Optional[Union[str,~azure.ai.evaluation.AzureAIProject]]
     :return: True if the Azure AI project is an OneDP project, False otherwise.
     :rtype: bool
     """
-    if isinstance(azure_ai_project, str):
-        return True
-    return False
+    return isinstance(azure_ai_project, str)
 
 
 def validate_azure_ai_project(o: object) -> AzureAIProject:

From 0322d0b3ec01a45aec53f9e905e2d099b892010d Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Fri, 25 Jul 2025 08:48:35 -0400
Subject: [PATCH 10/26] fix,tests: Don't log tokens as metrics

        Removing to match the behavior of the other clients
---
 .../azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
index 6fb5f9cea9f4..61b0ff58cb32 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
@@ -155,7 +155,7 @@ async def _submit_bulk_run(self, run: Run, local_storage: AbstractRunStorage, **
             # system metrics
             system_metrics = {}
             if batch_result:
-                system_metrics.update(dataclasses.asdict(batch_result.tokens))  # token related
+                # system_metrics.update(dataclasses.asdict(batch_result.tokens))  # token related
                 system_metrics.update(
                     {
                         # "duration": batch_result.duration.total_seconds(),

From d58be1cb10bbf65e9ebcdda47e5109244214774e Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Fri, 25 Jul 2025 10:38:30 -0400
Subject: [PATCH 11/26] fix: Set error message without depending on run storage

        Promptflow surfaces exceptions by reading them from their
        "Storage" abstraction. That has not been ported to this SDK.
---
 .../_legacy/_batch_engine/_run_submitter.py   | 20 ++++++++++++++++++-
 1 file changed, 19 insertions(+), 1 deletion(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
index 61b0ff58cb32..fa612e248a0a 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
@@ -5,6 +5,7 @@
 import dataclasses
 import inspect
 import sys
+import traceback
 
 from concurrent.futures import Executor
 from datetime import datetime, timezone
@@ -212,7 +213,24 @@ def stream_run(run: Run, storage: AbstractRunStorage, raise_on_error: bool) -> N
 
         if run.status == RunStatus.FAILED or run.status == RunStatus.CANCELED:
             if run.status == RunStatus.FAILED:
-                error_message = storage.load_exception().get("message", "Run fails with unknown error.")
+                # Get the first error message from the results, or use a default one
+                if run.result and run.result.error:
+                    error_message = "".join(
+                        traceback.format_exception(
+                            type(run.result.error), run.result.error, run.result.error.__traceback__
+                        )
+                    )
+                elif run.result and run.result.details:
+                    err = next((r.error for r in run.result.details if r.error), None)
+                    if err and err.exception:
+                        error_message = "".join(
+                            traceback.format_exception(type(err.exception), err.exception, err.exception.__traceback__)
+                        )
+                    elif err and err.details:
+                        error_message = err.details
+
+                if not error_message:
+                    error_message = "Run fails with unknown error."
             else:
                 error_message = "Run is canceled."
             if raise_on_error:

From 5bb01a7d65bb9f8c762e725e634cb9bb775e3c21 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Fri, 25 Jul 2025 18:15:47 -0400
Subject: [PATCH 12/26] tests,fix: Fix test_evaluate_invalid_column_mapping

    PR 40556 accidentally indented the assertion in
    test_evaluate_invalid_column_mapping into the `pytest.raises`
    block.

    This inadvertently made the test useless, since the `evaluate`
    call would always raise an exception which skips over the
    assertion as the exception unwinds the stack.

    This commit unindents the assertion so that it runs.

    Additionally, PR 41919 updated our validation logic to allow
    column mapping reference of arbitrary length e.g.
    `${target.foo.bar.baz`}`.

    So this commit also removes the test case that was explicitly
    guarding against this `${target.response.one}`
---
 .../azure-ai-evaluation/tests/unittests/test_evaluate.py | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
index 089c2917256e..cce9a105a44c 100644
--- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
+++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
@@ -410,7 +410,6 @@ def test_apply_column_mapping_target(self, json_data, inputs_mapping, response):
             {"query": "${foo.query}"},
             {"query": "${data.query"},
             {"query": "data.query", "response": "target.response"},
-            {"query": "${data.query}", "response": "${target.response.one}"},
         ],
     )
     def test_evaluate_invalid_column_mapping(self, mock_model_config, evaluate_test_data_jsonl_file, column_mapping):
@@ -426,10 +425,10 @@ def test_evaluate_invalid_column_mapping(self, mock_model_config, evaluate_test_
                 },
             )
 
-            assert (
-                "Unexpected references detected in 'column_mapping'. Ensure only ${target.} and ${data.} are used."
-                in exc_info.value.args[0]
-            )
+        assert (
+            "Unexpected references detected in 'column_mapping'. Ensure only ${target.} and ${data.} are used."
+            in exc_info.value.args[0]
+        )
 
     def test_evaluate_valid_column_mapping_with_numeric_chars(self, mock_model_config, evaluate_test_data_alphanumeric):
         # Valid column mappings that include numeric characters

From 150d63b3f123eabcdb2a8af5bd0deac7f4839f4e Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Sat, 26 Jul 2025 14:52:59 -0400
Subject: [PATCH 13/26] fix,tests: Force PFClient specific tests to use
 PFClient

---
 .../tests/unittests/test_evaluate.py                 | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
index cce9a105a44c..6f08f79909d8 100644
--- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
+++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
@@ -855,20 +855,20 @@ def custom_aggregator(values: List[float]) -> float:
             return sum(values) + 1
 
         os.environ["AI_EVALS_BATCH_USE_ASYNC"] = use_async
-        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators)
+        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators, _use_pf_client=True)
         counting_eval._set_conversation_aggregation_type(_AggregationType.MIN)
-        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators)
+        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators, _use_pf_client=True)
         counting_eval._set_conversation_aggregation_type(_AggregationType.SUM)
-        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators)
+        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators, _use_pf_client=True)
         counting_eval._set_conversation_aggregation_type(_AggregationType.MAX)
-        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators)
+        _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators, _use_pf_client=True)
         if use_async == "true":
             counting_eval._set_conversation_aggregator(custom_aggregator)
-            _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators)
+            _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators, _use_pf_client=True)
         else:
             with pytest.raises(EvaluationException) as exc_info:
                 counting_eval._set_conversation_aggregator(custom_aggregator)
-                _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators)
+                _ = evaluate(data=evaluate_test_data_conversion_jsonl_file, evaluators=evaluators, _use_pf_client=True)
             assert "TestEvaluate.test_aggregation_serialization.<locals>.custom_aggregator" in exc_info.value.args[0]
 
     def test_unsupported_file_inputs(self, mock_model_config, unsupported_file_type):

From 089e568f880ea78f3c4a9f1b497512e807c9cf5e Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Sat, 26 Jul 2025 14:53:35 -0400
Subject: [PATCH 14/26] fix,tests: Force CodeClient specific tests to use
 CodeClient

---
 .../azure-ai-evaluation/tests/unittests/test_evaluate.py    | 6 ++++++
 .../tests/unittests/test_evaluate_performance.py            | 5 ++---
 2 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
index 6f08f79909d8..3d2e306d0d01 100644
--- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
+++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate.py
@@ -700,6 +700,7 @@ def test_optional_inputs_with_data(self, questions_file, questions_answers_basic
                 "no": NoInputEval(),
             },
             _use_pf_client=False,
+            _use_run_submitter_client=False,
         )  # type: ignore
 
         first_row = results["rows"][0]
@@ -715,6 +716,7 @@ def test_optional_inputs_with_data(self, questions_file, questions_answers_basic
                     "non": NonOptionalEval(),
                 },
                 _use_pf_client=False,
+                _use_run_submitter_client=False,
             )  # type: ignore
 
         expected_message = "Some evaluators are missing required inputs:\n" "- non: ['response']\n"
@@ -725,6 +727,7 @@ def test_optional_inputs_with_data(self, questions_file, questions_answers_basic
             data=questions_file,
             evaluators={"half": HalfOptionalEval(), "opt": OptionalEval(), "no": NoInputEval()},
             _use_pf_client=False,
+            _use_run_submitter_client=False,
         )  # type: ignore
 
         first_row_2 = only_question_results["rows"][0]
@@ -741,6 +744,7 @@ def test_optional_inputs_with_target(self, questions_file, questions_answers_bas
             target=_new_answer_target,
             evaluators={"echo": EchoEval()},
             _use_pf_client=False,
+            _use_run_submitter_client=False,
         )  # type: ignore
 
         assert target_answer_results["rows"][0]["outputs.echo.echo_query"] == "How long is flight from Earth to LV-426?"
@@ -753,6 +757,7 @@ def test_optional_inputs_with_target(self, questions_file, questions_answers_bas
             target=_question_override_target,
             evaluators={"echo": EchoEval()},
             _use_pf_client=False,
+            _use_run_submitter_client=False,
         )  # type: ignore
 
         assert question_override_results["rows"][0]["outputs.echo.echo_query"] == "new query"
@@ -764,6 +769,7 @@ def test_optional_inputs_with_target(self, questions_file, questions_answers_bas
             target=_question_answer_override_target,
             evaluators={"echo": EchoEval()},
             _use_pf_client=False,
+            _use_run_submitter_client=False,
         )  # type: ignore
         assert double_override_results["rows"][0]["outputs.echo.echo_query"] == "new query"
         assert double_override_results["rows"][0]["outputs.echo.echo_response"] == "new response"
diff --git a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate_performance.py b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate_performance.py
index 515fd6508f1c..c05967f9c68a 100644
--- a/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate_performance.py
+++ b/sdk/evaluation/azure-ai-evaluation/tests/unittests/test_evaluate_performance.py
@@ -39,6 +39,7 @@ def test_bulk_evaluate(self, big_f1_data_file):
             data=big_f1_data_file,
             evaluators={"f1": f1_score_eval},
             _use_pf_client=False,
+            _use_run_submitter_client=False,
         )
         end = time.perf_counter()
         diff = end - start  # diff stored as variable just to make sure pytest output
@@ -61,9 +62,7 @@ def test_evaluate_parallelism(self, ten_queries_file):
         # run the evaluation with targets
         start = time.perf_counter()
         result = evaluate(
-            data=ten_queries_file,
-            evaluators={"slow": slow_eval},
-            _use_pf_client=False,
+            data=ten_queries_file, evaluators={"slow": slow_eval}, _use_pf_client=False, _use_run_submitter_client=False
         )
         end = time.perf_counter()
         # Time duration is stored as variable just to make sure pytest output

From d1f09378f4d5269553f9604bda3b37fc33e31b13 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Sat, 26 Jul 2025 15:06:53 -0400
Subject: [PATCH 15/26] fix: Improve the ergonomics for picking which client is
 used

        Except for the CodeClient, you only need to use
        at most 1 of `_use_pf_client` and `_use_run_submitter_client`
---
 .../ai/evaluation/_evaluate/_evaluate.py      | 42 +++++++++++++++++--
 1 file changed, 38 insertions(+), 4 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py
index fea768740277..63a4ae5d8dbc 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py
@@ -9,7 +9,7 @@
 import re
 import tempfile
 import json
-from typing import Any, Callable, Dict, List, Optional, Set, Tuple, TypedDict, Union, cast
+from typing import Any, Callable, Dict, List, Literal, Optional, Set, Tuple, TypedDict, Union, cast
 
 from openai import OpenAI, AzureOpenAI
 from azure.ai.evaluation._legacy._adapters._constants import LINE_NUMBER
@@ -1016,15 +1016,49 @@ def _preprocess_data(
     batch_run_client: BatchClient
     batch_run_data: Union[str, os.PathLike, pd.DataFrame] = data
 
-    if kwargs.pop("_use_run_submitter_client", False):
+    def get_client_type(evaluate_kwargs: Dict[str, Any]) -> Literal["run_submitter", "pf_client", "code_client"]:
+        """Determines the BatchClient to use from provided kwargs (_use_run_submitter_client and _use_pf_client)"""
+        _use_run_submitter_client = cast(Optional[bool], kwargs.pop("_use_run_submitter_client", None))
+        _use_pf_client = cast(Optional[bool], kwargs.pop("_use_pf_client", None))
+
+        if _use_run_submitter_client is None and _use_pf_client is None:
+            # If both are unset, return default
+            return "run_submitter"
+
+        if _use_run_submitter_client and _use_pf_client:
+            raise EvaluationException(
+                message="Only one of _use_pf_client and _use_run_submitter_client should be set to True.",
+                target=ErrorTarget.EVALUATE,
+                category=ErrorCategory.INVALID_VALUE,
+                blame=ErrorBlame.USER_ERROR,
+            )
+
+        if _use_run_submitter_client == False and _use_pf_client == False:
+            return "code_client"
+
+        if _use_run_submitter_client:
+            return "run_submitter"
+        if _use_pf_client:
+            return "pf_client"
+
+        if _use_run_submitter_client is None and _use_pf_client == False:
+            return "run_submitter"
+        if _use_run_submitter_client == False and _use_pf_client is None:
+            return "pf_client"
+
+        assert False, "This should be impossible"
+
+    client_type: Literal["run_submitter", "pf_client", "code_client"] = get_client_type(kwargs)
+
+    if client_type == "run_submitter":
         batch_run_client = RunSubmitterClient()
         batch_run_data = input_data_df
-    elif kwargs.pop("_use_pf_client", True):
+    elif client_type == "pf_client":
         batch_run_client = ProxyClient(user_agent=UserAgentSingleton().value)
         # Ensure the absolute path is passed to pf.run, as relative path doesn't work with
         # multiple evaluators. If the path is already absolute, abspath will return the original path.
         batch_run_data = os.path.abspath(data)
-    else:
+    elif client_type == "code_client":
         batch_run_client = CodeClient()
         batch_run_data = input_data_df
 

From 06e05c4ae07277d1b016ffff78ca1d223c6cd739 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Sat, 26 Jul 2025 15:28:10 -0400
Subject: [PATCH 16/26] feat: Show exception message in run logs

---
 .../_legacy/_batch_engine/_engine.py          | 34 +++++++++++++++++--
 .../_legacy/_batch_engine/_result.py          |  2 ++
 2 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index 8564104f4b4d..0447a7ac872a 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -26,6 +26,7 @@
     Dict,
     Final,
     Generator,
+    List,
     Mapping,
     MutableMapping,
     Optional,
@@ -41,7 +42,7 @@
 from ._status import BatchStatus
 from ._result import BatchResult, BatchRunDetails, BatchRunError, TokenMetrics
 from ._run_storage import AbstractRunStorage, NoOpRunStorage
-from .._common._logging import log_progress, NodeLogManager
+from .._common._logging import log_progress, logger, NodeLogManager
 from ..._exceptions import ErrorBlame
 from ._exceptions import (
     BatchEngineCanceledError,
@@ -262,10 +263,12 @@ async def _exec_in_task(
                     end_time=None,
                     tokens=TokenMetrics(0, 0, 0),
                     error=BatchRunError("The line run is not completed.", None),
+                    index=i,
                 )
             )
             for i in range(len(batch_inputs))
         ]
+        self.handle_line_failures(result_details)
 
         for line_result in result_details:
             # Indicate the worst status of the batch run. This works because
@@ -279,9 +282,15 @@ async def _exec_in_task(
                 metrics.total_tokens += line_result.tokens.total_tokens
 
         if failed_lines and not error:
-            error = BatchEngineRunFailedError(
-                str(floor(failed_lines / len(batch_inputs) * 100)) + f"% of the batch run failed."
+            error_message = f"{floor(failed_lines / len(batch_inputs) * 100)}% of the batch run failed."
+            first_exception: Optional[Exception] = next(
+                (result for result in result_details if result.error and result.error.exception),
+                None,
             )
+            if first_exception is not None:
+                error_message += f" {first_exception}"
+
+            error = BatchEngineRunFailedError(error_message)
 
         return BatchResult(
             status=status,
@@ -355,6 +364,7 @@ async def _exec_line_async(
                 end_time=None,
                 tokens=TokenMetrics(0, 0, 0),
                 error=None,
+                index=index,
             )
 
             try:
@@ -399,6 +409,24 @@ async def _exec_line_async(
 
         return index, details
 
+    @staticmethod
+    def handle_line_failures(run_infos: List[BatchRunDetails], raise_on_line_failure: bool = False):
+        """Handle line failures in batch run"""
+        failed_run_infos: List[BatchRunDetails] = [r for r in run_infos if r.status == BatchStatus.Failed]
+        failed_msg: Optional[str] = None
+        if len(failed_run_infos) > 0:
+            failed_indexes = ",".join([str(r.index) for r in failed_run_infos])
+            first_fail_exception: str = failed_run_infos[0].error.details
+            if raise_on_line_failure:
+                failed_msg = "Flow run failed due to the error: " + first_fail_exception
+                raise Exception(failed_msg)
+
+            failed_msg = (
+                f"{len(failed_run_infos)}/{len(run_infos)} flow run failed, indexes: [{failed_indexes}],"
+                f" exception of index {failed_run_infos[0].index}: {first_fail_exception}"
+            )
+            logger.error(failed_msg)
+
     def _persist_run_info(self, line_results: Sequence[BatchRunDetails]):
         # TODO ralphe: implement?
         pass
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_result.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_result.py
index d1aa043aff37..9adb67640fc3 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_result.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_result.py
@@ -55,6 +55,8 @@ class BatchRunDetails:
     """The token metrics of the line run."""
     error: Optional[BatchRunError]
     """The error of the line run. This will only be set if the status is Failed."""
+    index: int
+    """The line run index."""
 
     @property
     def duration(self) -> timedelta:

From 07296fc207c8f45343b53c18bd5698e641235974 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Sat, 26 Jul 2025 15:29:06 -0400
Subject: [PATCH 17/26] fix: For safety evaluation to use codeclient as
 originally intended

---
 .../azure/ai/evaluation/_safety_evaluation/_safety_evaluation.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_safety_evaluation/_safety_evaluation.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_safety_evaluation/_safety_evaluation.py
index 0fd827fe1c34..3984b5b07a74 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_safety_evaluation/_safety_evaluation.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_safety_evaluation/_safety_evaluation.py
@@ -903,6 +903,7 @@ async def __call__(
                     evaluation_name=evaluation_name,
                     output_path=output_path if output_path else f"{output_prefix}{strategy}{RESULTS_EXT}",
                     _use_pf_client=False,  # TODO: Remove this once eval logic for red team agent is moved to red team agent
+                    _use_run_submitter_client=False,  # TODO: Remove this once eval logic for red team agent is moved to red team agent
                 )
                 evaluation_results[strategy] = evaluate_outputs
             return evaluation_results

From 6c2d8c09f80fb01cdc719dc31469e0336350f57a Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 10:23:42 -0400
Subject: [PATCH 18/26] fix: Don't wrap EvaluationExcpeiton in BatchEngineError

---
 .../azure/ai/evaluation/_legacy/_batch_engine/_engine.py      | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index 0447a7ac872a..19ca4f136958 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -43,7 +43,7 @@
 from ._result import BatchResult, BatchRunDetails, BatchRunError, TokenMetrics
 from ._run_storage import AbstractRunStorage, NoOpRunStorage
 from .._common._logging import log_progress, logger, NodeLogManager
-from ..._exceptions import ErrorBlame
+from ..._exceptions import ErrorBlame, EvaluationException
 from ._exceptions import (
     BatchEngineCanceledError,
     BatchEngineError,
@@ -118,6 +118,8 @@ async def run(
             id = id or str(uuid4())
             result: BatchResult = await self._exec_in_task(id, batch_inputs, start_time)
             return result
+        except EvaluationException:
+            raise
         except Exception as ex:
             raise BatchEngineError(
                 "Unexpected error while running the batch run.", blame=ErrorBlame.SYSTEM_ERROR

From 92a4644db27f22a98a6f5404517fc3ac84ee4bc4 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 10:36:57 -0400
Subject: [PATCH 19/26] refactor: Refactor BatchConfig

---
 .../_batch_run/_run_submitter_client.py       | 23 +++++++++++++++----
 .../_legacy/_batch_engine/_config.py          |  6 ++---
 .../_legacy/_batch_engine/_engine.py          | 18 ++++++---------
 .../_legacy/_batch_engine/_run_submitter.py   |  4 +---
 4 files changed, 30 insertions(+), 21 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
index 7b0ca71ff8c9..e71747918bcc 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
@@ -20,14 +20,29 @@
 from ..._legacy._adapters.types import AttrDict
 from ..._legacy._common._thread_pool_executor_with_context import ThreadPoolExecutorWithContext
 from ..._evaluate._utils import _has_aggregator
-from ..._constants import Prefixes
+from ..._constants import Prefixes, PF_BATCH_TIMEOUT_SEC
 
-LOGGER = logging.getLogger(__name__)
+from .._utils import get_int_env_var as get_int
+
+
+LOGGER = logging.getLogger("run")
+MISSING_VALUE: Final[int] = sys.maxsize
 
 
 class RunSubmitterClient:
-    def __init__(self, config: Optional[BatchEngineConfig] = None) -> None:
-        self._config = config or BatchEngineConfig(LOGGER, use_async=True)
+    def __init__(self, *, config: Optional[BatchEngineConfig] = None) -> None:
+        if config:
+            self._config = config
+        else:
+            # Generate default config and apply any overrides to the configuration from environment variables
+            self._config = BatchEngineConfig(LOGGER, use_async=True)
+            if (val := get_int(PF_BATCH_TIMEOUT_SEC, MISSING_VALUE)) != MISSING_VALUE:
+                self._config.batch_timeout_seconds = val
+            if (val := get_int("PF_LINE_TIMEOUT_SEC", MISSING_VALUE)) != MISSING_VALUE:
+                self._config.line_timeout_seconds = val
+            if (val := get_int("PF_WORKER_COUNT", MISSING_VALUE)) != MISSING_VALUE:
+                self._config.max_concurrency = val
+
         self._thread_pool = ThreadPoolExecutorWithContext(
             thread_name_prefix="evaluators_thread", max_workers=self._config.max_concurrency
         )
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py
index 749f92f5ead4..208286dc3ff1 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py
@@ -19,7 +19,7 @@ class BatchEngineConfig:
     batch_timeout_seconds: int = PF_BATCH_TIMEOUT_SEC_DEFAULT
     """The maximum amount of time to wait for all evaluations in the batch to complete."""
 
-    run_timeout_seconds: int = 600
+    line_timeout_seconds: int = 600
     """The maximum amount of time to wait for an evaluation to run against a single entry
     in the data input to complete."""
 
@@ -37,8 +37,8 @@ def __post_init__(self):
             raise ValueError("logger cannot be None")
         if self.batch_timeout_seconds <= 0:
             raise ValueError("batch_timeout_seconds must be greater than 0")
-        if self.run_timeout_seconds <= 0:
-            raise ValueError("run_timeout_seconds must be greater than 0")
+        if self.line_timeout_seconds <= 0:
+            raise ValueError("line_timeout_seconds must be greater than 0")
         if self.max_concurrency <= 0:
             raise ValueError("max_concurrency must be greater than 0")
         if self.default_num_results <= 0:
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index 19ca4f136958..7d9b91e2f5fd 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -38,6 +38,7 @@
 )
 from uuid import uuid4
 
+from ._config import BatchEngineConfig
 from ._utils import DEFAULTS_KEY, get_int_env_var, get_value_from_path, is_async_callable
 from ._status import BatchStatus
 from ._result import BatchResult, BatchRunDetails, BatchRunError, TokenMetrics
@@ -69,30 +70,25 @@ def __init__(
         self,
         func: Callable,
         *,
+        config: BatchEngineConfig,
         storage: Optional[AbstractRunStorage] = None,
-        batch_timeout_sec: Optional[int] = None,
-        line_timeout_sec: Optional[int] = None,
-        max_worker_count: Optional[int] = None,
         executor: Optional[Executor] = None,
     ):
         """Create a new batch engine instance
 
         :param Callable func: The function to run the flow
+        :param BatchEngineConfig config: The configuration for the batch engine
         :param Optional[AbstractRunStorage] storage: The storage to store execution results
-        :param Optional[int] batch_timeout_sec: The timeout of batch run in seconds
-        :param Optional[int] line_timeout_sec: The timeout of each line in seconds
-        :param Optional[int] max_worker_count: The concurrency limit of batch run
         :param Optional[Executor] executor: The executor to run the flow (if needed)
         """
 
         self._func: Callable = func
+        self._config: BatchEngineConfig = config
         self._storage: AbstractRunStorage = storage or NoOpRunStorage()
 
-        # TODO ralphe: Consume these from the batch context/config instead of from
-        #              kwargs or (even worse) environment variables
-        self._batch_timeout_sec = batch_timeout_sec or get_int_env_var("PF_BATCH_TIMEOUT_SEC")
-        self._line_timeout_sec = line_timeout_sec or get_int_env_var("PF_LINE_TIMEOUT_SEC", 600)
-        self._max_worker_count = max_worker_count or get_int_env_var("PF_WORKER_COUNT") or MAX_WORKER_COUNT
+        self._batch_timeout_sec = self._config.batch_timeout_seconds
+        self._line_timeout_sec = self._config.line_timeout_seconds
+        self._max_worker_count = self._config.max_concurrency
 
         self._executor: Optional[Executor] = executor
         self._is_canceled: bool = False
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
index fa612e248a0a..c150170abcbb 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
@@ -121,10 +121,8 @@ async def _submit_bulk_run(self, run: Run, local_storage: AbstractRunStorage, **
         try:
             batch_engine = BatchEngine(
                 run.dynamic_callable,
+                config=self._config,
                 storage=local_storage,
-                batch_timeout_sec=self._config.batch_timeout_seconds,
-                line_timeout_sec=self._config.run_timeout_seconds,
-                max_worker_count=self._config.max_concurrency,
                 executor=self._executor,
             )
 

From 3af4f9dfd870094852f83159281cd75c90a3e662 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 10:38:00 -0400
Subject: [PATCH 20/26] fix: Make raising on error configurable for
 runsubmitterclient

---
 .../evaluation/_evaluate/_batch_run/_run_submitter_client.py  | 4 +++-
 .../azure/ai/evaluation/_evaluate/_evaluate.py                | 4 +++-
 .../azure/ai/evaluation/_legacy/_batch_engine/_config.py      | 3 +++
 .../ai/evaluation/_legacy/_batch_engine/_run_submitter.py     | 2 +-
 4 files changed, 10 insertions(+), 3 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
index e71747918bcc..ae7d84181702 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
@@ -30,7 +30,7 @@
 
 
 class RunSubmitterClient:
-    def __init__(self, *, config: Optional[BatchEngineConfig] = None) -> None:
+    def __init__(self, *, raise_on_errors: bool = False, config: Optional[BatchEngineConfig] = None) -> None:
         if config:
             self._config = config
         else:
@@ -43,6 +43,8 @@ def __init__(self, *, config: Optional[BatchEngineConfig] = None) -> None:
             if (val := get_int("PF_WORKER_COUNT", MISSING_VALUE)) != MISSING_VALUE:
                 self._config.max_concurrency = val
 
+        self._config.raise_on_error = raise_on_errors
+
         self._thread_pool = ThreadPoolExecutorWithContext(
             thread_name_prefix="evaluators_thread", max_workers=self._config.max_concurrency
         )
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py
index 63a4ae5d8dbc..861e1acbad11 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_evaluate.py
@@ -876,6 +876,7 @@ def _evaluate(  # pylint: disable=too-many-locals,too-many-statements
         output_path=output_path,
         azure_ai_project=azure_ai_project,
         evaluation_name=evaluation_name,
+        fail_on_evaluator_errors=fail_on_evaluator_errors,
         **kwargs,
     )
 
@@ -983,6 +984,7 @@ def _preprocess_data(
     output_path: Optional[Union[str, os.PathLike]] = None,
     azure_ai_project: Optional[Union[str, AzureAIProject]] = None,
     evaluation_name: Optional[str] = None,
+    fail_on_evaluator_errors: bool = False,
     **kwargs,
 ) -> __ValidatedData:
     # Process evaluator config to replace ${target.} with ${data.}
@@ -1051,7 +1053,7 @@ def get_client_type(evaluate_kwargs: Dict[str, Any]) -> Literal["run_submitter",
     client_type: Literal["run_submitter", "pf_client", "code_client"] = get_client_type(kwargs)
 
     if client_type == "run_submitter":
-        batch_run_client = RunSubmitterClient()
+        batch_run_client = RunSubmitterClient(raise_on_errors=fail_on_evaluator_errors)
         batch_run_data = input_data_df
     elif client_type == "pf_client":
         batch_run_client = ProxyClient(user_agent=UserAgentSingleton().value)
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py
index 208286dc3ff1..ebbe9c056bdf 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_config.py
@@ -32,6 +32,9 @@ class BatchEngineConfig:
     default_num_results: int = 100
     """The default number of results to return if you don't ask for all results."""
 
+    raise_on_error: bool = True
+    """Whether to raise an error if an evaluation fails."""
+
     def __post_init__(self):
         if self.logger is None:
             raise ValueError("logger cannot be None")
diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
index c150170abcbb..3af8b011a9d4 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
@@ -80,7 +80,7 @@ async def submit(
             # unnecessary Flow loading code was removed here. Instead do direct calls to _submit_bulk_run
             await self._submit_bulk_run(run=run, local_storage=local_storage, **kwargs)
 
-        self.stream_run(run=run, storage=local_storage, raise_on_error=True)
+        self.stream_run(run=run, storage=local_storage, raise_on_error=self._config.raise_on_error)
         return run
 
     async def _submit_bulk_run(self, run: Run, local_storage: AbstractRunStorage, **kwargs) -> None:

From b9cad3309957554351bd514bfa5c7e3d29abbe2a Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 12:12:37 -0400
Subject: [PATCH 21/26] chore: Update changelog

---
 sdk/evaluation/azure-ai-evaluation/CHANGELOG.md | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md b/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md
index eece18881253..4488e01a41ff 100644
--- a/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md
+++ b/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md
@@ -21,6 +21,12 @@ tolerance for harmful responses).
 - Significant improvements to Relevance evaluator. New version has more concrete rubrics and has less variance, is much faster and consumes fewer tokens.
 
 
+### Other Changes
+
+- The default engine for evaluation was changed from `promptflow` (PFClient) to an in-SDK Batch Engine
+  - In the event users need to fallback to the `promptflow` implementation, this can temporarily be done by setting the `_use_pf_client=True` when invoking `evaluate()`
+      - Note: This is due to be removed in a future release.
+
 ## 1.9.0 (2025-07-02)
 
 ### Features Added

From 0e2a56009c33f7b2daddb446c32e0cf8615ebd66 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 12:14:48 -0400
Subject: [PATCH 22/26] fix: Uncomment log_path

---
 .../ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
index ae7d84181702..bcb7d567cdee 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_evaluate/_batch_run/_run_submitter_client.py
@@ -158,7 +158,7 @@ def get_run_summary(self, client_run: BatchClientRun) -> Dict[str, Any]:
             "duration": str(run.duration),
             "completed_lines": total_lines - failed_lines,
             "failed_lines": failed_lines,
-            # "log_path": "",
+            "log_path": None,
         }
 
     @staticmethod

From 5d4968efa47ee060fd94103978b7b1cd7ee8e3ae Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 12:16:59 -0400
Subject: [PATCH 23/26] chore: Add promptflow to dev-requirements.txt

    Some tests have explicit dependencies on the promptflow implementation.
    Since we aren't removing the code path yet, allow them to run by installing
    promptflow for those tests.
---
 sdk/evaluation/azure-ai-evaluation/dev_requirements.txt | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt b/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt
index 1ccece0d2952..47edecbe4793 100644
--- a/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt
+++ b/sdk/evaluation/azure-ai-evaluation/dev_requirements.txt
@@ -10,4 +10,6 @@ azure-ai-inference>=1.0.0b4
 azure-ai-projects<=1.0.0b10
 aiohttp
 filelock
+promptflow-core>=1.17.1
+promptflow-devkit>=1.17.1
 -e ../azure-ai-evaluation

From b79f3a0f0b555db3a1e7ac9370931f617bc2a5fc Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 16:18:39 -0400
Subject: [PATCH 24/26] fix: Initialize error_message

---
 .../azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py  | 1 +
 1 file changed, 1 insertion(+)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
index 3af8b011a9d4..131b36df3610 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_run_submitter.py
@@ -200,6 +200,7 @@ def stream_run(run: Run, storage: AbstractRunStorage, raise_on_error: bool) -> N
             return
 
         file_handler = sys.stdout
+        error_message: Optional[str] = None
         try:
             printed = 0
             available_logs = storage.logger.get_logs()

From 9a7c190cd6088abaecc6d6b4e45ab29f07b3d6ab Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Mon, 28 Jul 2025 16:20:31 -0400
Subject: [PATCH 25/26] fix: Get exception instead of batchrunresult

---
 .../azure/ai/evaluation/_legacy/_batch_engine/_engine.py        | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
index 7d9b91e2f5fd..f9f255d5f695 100644
--- a/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
+++ b/sdk/evaluation/azure-ai-evaluation/azure/ai/evaluation/_legacy/_batch_engine/_engine.py
@@ -282,7 +282,7 @@ async def _exec_in_task(
         if failed_lines and not error:
             error_message = f"{floor(failed_lines / len(batch_inputs) * 100)}% of the batch run failed."
             first_exception: Optional[Exception] = next(
-                (result for result in result_details if result.error and result.error.exception),
+                (result.error.exception for result in result_details if result.error and result.error.exception),
                 None,
             )
             if first_exception is not None:

From be8285d552055c962093684e89d896659c4a1fd3 Mon Sep 17 00:00:00 2001
From: kdestin <101366538+kdestin@users.noreply.github.com>
Date: Tue, 29 Jul 2025 15:41:53 -0400
Subject: [PATCH 26/26] chore,docs: Clarify changelog

---
 sdk/evaluation/azure-ai-evaluation/CHANGELOG.md | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md b/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md
index 4488e01a41ff..fd208bfd5268 100644
--- a/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md
+++ b/sdk/evaluation/azure-ai-evaluation/CHANGELOG.md
@@ -23,9 +23,10 @@ tolerance for harmful responses).
 
 ### Other Changes
 
-- The default engine for evaluation was changed from `promptflow` (PFClient) to an in-SDK Batch Engine
-  - In the event users need to fallback to the `promptflow` implementation, this can temporarily be done by setting the `_use_pf_client=True` when invoking `evaluate()`
-      - Note: This is due to be removed in a future release.
+- The default engine for evaluation was changed from `promptflow` (PFClient) to an in-SDK batch client (RunSubmitterClient)
+  - Note: We've temporarily kept an escape hatch to fall back to the legacy `promptflow` implementation by setting `_use_pf_client=True` when invoking `evaluate()`.
+    This is due to be removed in a future release.
+
 
 ## 1.9.0 (2025-07-02)