diff --git a/src/promptflow-azure/promptflow/azure/_cli/entry.py b/src/promptflow-azure/promptflow/azure/_cli/entry.py
index b98edfe182b..338a6d4d9b4 100644
--- a/src/promptflow-azure/promptflow/azure/_cli/entry.py
+++ b/src/promptflow-azure/promptflow/azure/_cli/entry.py
@@ -2,7 +2,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 # pylint: disable=wrong-import-position
-import json
 import time
 
 from promptflow._cli._pf.help import show_privacy_statement, show_welcome_message
@@ -17,14 +16,7 @@
 import logging  # noqa: E402
 import sys  # noqa: E402
 
-from promptflow._sdk._utils import (  # noqa: E402
-    get_promptflow_azure_version,
-    get_promptflow_core_version,
-    get_promptflow_devkit_version,
-    get_promptflow_sdk_version,
-    get_promptflow_tracing_version,
-    print_pf_version,
-)
+from promptflow._sdk._utils import print_pf_version, print_promptflow_version_dict_string  # noqa: E402
 from promptflow._utils.logger_utils import get_cli_sdk_logger  # noqa: E402
 from promptflow._utils.user_agent_utils import setup_user_agent_to_operation_context  # noqa: E402
 from promptflow.azure._cli._flow import add_parser_flow, dispatch_flow_commands  # noqa: E402
@@ -47,7 +39,7 @@ def run_command(args):
             for handler in logger.handlers:
                 handler.setLevel(logging.DEBUG)
         if args.version:
-            print_pf_version(with_azure=True)
+            print_pf_version(with_azure=True, ignore_none=True)
         elif args.action == "run":
             dispatch_run_commands(args)
         elif args.action == "flow":
@@ -123,24 +115,8 @@ def main():
     """Entrance of pf CLI."""
     command_args = sys.argv[1:]
     if len(command_args) == 1 and command_args[0] == "version":
-        version_dict = {"promptflow": get_promptflow_sdk_version()}
-        # check tracing version
-        version_tracing = get_promptflow_tracing_version()
-        if version_tracing:
-            version_dict["promptflow-tracing"] = version_tracing
-        # check azure version
-        version_azure = get_promptflow_azure_version()
-        if version_azure:
-            version_dict["promptflow-azure"] = version_azure
-        # check core version
-        version_core = get_promptflow_core_version()
-        if version_core:
-            version_dict["promptflow-core"] = version_core
-        # check devkit version
-        version_devkit = get_promptflow_devkit_version()
-        if version_devkit:
-            version_dict["promptflow-devkit"] = version_devkit
-        return json.dumps(version_dict, ensure_ascii=False, indent=2, sort_keys=True, separators=(",", ": ")) + "\n"
+        print_promptflow_version_dict_string(with_azure=True, ignore_none=True)
+        return
     if len(command_args) == 0:
         # print privacy statement & welcome message like azure-cli
         show_privacy_statement()
diff --git a/src/promptflow-azure/promptflow/azure/operations/_flow_operations.py b/src/promptflow-azure/promptflow/azure/operations/_flow_operations.py
index 1dd6e625aad..956978f01af 100644
--- a/src/promptflow-azure/promptflow/azure/operations/_flow_operations.py
+++ b/src/promptflow-azure/promptflow/azure/operations/_flow_operations.py
@@ -25,7 +25,6 @@
 from azure.core.exceptions import HttpResponseError
 
 from promptflow._constants import FlowType as FlowYamlType
-from promptflow._proxy import ProxyFactory
 from promptflow._sdk._constants import (
     CLIENT_FLOW_TYPE_2_SERVICE_FLOW_TYPE,
     MAX_LIST_CLI_RESULTS,
@@ -476,6 +475,8 @@ def _resolve_arm_id_or_upload_dependencies(self, flow: Flow, ignore_tools_json=F
 
     @classmethod
     def _try_resolve_code_for_flow(cls, flow: Flow, ops: OperationOrchestrator, ignore_tools_json=False) -> None:
+        from promptflow._proxy import ProxyFactory
+
         if flow.path:
             # remote path
             if flow.path.startswith("azureml://datastores"):
diff --git a/src/promptflow-core/promptflow/_utils/version_hint_utils.py b/src/promptflow-core/promptflow/_utils/version_hint_utils.py
index e976e316778..1b5dd232381 100644
--- a/src/promptflow-core/promptflow/_utils/version_hint_utils.py
+++ b/src/promptflow-core/promptflow/_utils/version_hint_utils.py
@@ -107,7 +107,9 @@ def hint_for_update():
         if LATEST_VERSION in cached_versions:
             from packaging.version import parse
 
-            if parse(cached_versions[CURRENT_VERSION]) < parse(cached_versions[LATEST_VERSION]):
+            if cached_versions[CURRENT_VERSION] is None or parse(cached_versions[CURRENT_VERSION]) < parse(
+                cached_versions[LATEST_VERSION]
+            ):
                 cached_versions[LAST_HINT_TIME] = str(datetime.datetime.now())
                 message = (
                     f"New prompt flow version available: promptflow-{cached_versions[LATEST_VERSION]}. Running "
diff --git a/src/promptflow-core/promptflow/executor/_service/apis/batch.py b/src/promptflow-core/promptflow/executor/_service/apis/batch.py
index 3ee3227bbfc..bce6e6b0cc7 100644
--- a/src/promptflow-core/promptflow/executor/_service/apis/batch.py
+++ b/src/promptflow-core/promptflow/executor/_service/apis/batch.py
@@ -32,10 +32,11 @@ def initialize(request: InitializationRequest):
         set_environment_variables(request)
         # init batch coordinator to validate flow and create process pool
         batch_coordinator = BatchCoordinator(
-            request.working_dir,
-            request.flow_file,
-            request.output_dir,
-            request.connections,
+            working_dir=request.working_dir,
+            flow_file=request.flow_file,
+            output_dir=request.output_dir,
+            flow_name=request.flow_name,
+            connections=request.connections,
             worker_count=request.worker_count,
             line_timeout_sec=request.line_timeout_sec,
         )
diff --git a/src/promptflow-core/promptflow/executor/_service/apis/execution.py b/src/promptflow-core/promptflow/executor/_service/apis/execution.py
index 778dcdd55a4..63b5c2ff584 100644
--- a/src/promptflow-core/promptflow/executor/_service/apis/execution.py
+++ b/src/promptflow-core/promptflow/executor/_service/apis/execution.py
@@ -94,6 +94,7 @@ def flow_test(request: FlowExecutionRequest):
             request.inputs,
             run_id=request.run_id,
             storage=storage,
+            name=request.flow_name,
         )
 
 
diff --git a/src/promptflow-core/promptflow/executor/_service/contracts/execution_request.py b/src/promptflow-core/promptflow/executor/_service/contracts/execution_request.py
index fc7f286694a..7945ced0aee 100644
--- a/src/promptflow-core/promptflow/executor/_service/contracts/execution_request.py
+++ b/src/promptflow-core/promptflow/executor/_service/contracts/execution_request.py
@@ -20,6 +20,7 @@ class BaseExecutionRequest(BaseRequest):
     log_path: Optional[str] = None
     connections: Optional[Mapping[str, Any]] = None
     environment_variables: Optional[Mapping[str, Any]] = None
+    flow_name: str = None
 
     def get_run_mode(self):
         raise NotImplementedError(f"Request type {self.__class__.__name__} is not implemented.")
diff --git a/src/promptflow-core/promptflow/executor/_service/utils/batch_coordinator.py b/src/promptflow-core/promptflow/executor/_service/utils/batch_coordinator.py
index dd112bb70d2..dd09e05c7a5 100644
--- a/src/promptflow-core/promptflow/executor/_service/utils/batch_coordinator.py
+++ b/src/promptflow-core/promptflow/executor/_service/utils/batch_coordinator.py
@@ -28,6 +28,7 @@ def __init__(
         working_dir: Path,
         flow_file: Path,
         output_dir: Path,
+        flow_name: str = None,
         connections: Optional[Mapping[str, Any]] = None,
         worker_count: Optional[int] = None,
         line_timeout_sec: Optional[int] = None,
@@ -46,7 +47,7 @@ def __init__(
         # So we pass DummyRunStorage to FlowExecutor because we don't need to
         # persist the run infos during execution in server mode.
         self._flow_executor = FlowExecutor.create(
-            flow_file, connections, working_dir, storage=DummyRunStorage(), raise_ex=False
+            flow_file, connections, working_dir, storage=DummyRunStorage(), raise_ex=False, name=flow_name
         )
 
         # Init line execution process pool and set serialize_multimedia_during_execution to True
diff --git a/src/promptflow-devkit/promptflow/_cli/_pf/_service.py b/src/promptflow-devkit/promptflow/_cli/_pf/_service.py
index 8bc484d2a7b..3a9d4c5b935 100644
--- a/src/promptflow-devkit/promptflow/_cli/_pf/_service.py
+++ b/src/promptflow-devkit/promptflow/_cli/_pf/_service.py
@@ -25,13 +25,13 @@
     check_pfs_service_status,
     dump_port_to_config,
     get_current_env_pfs_file,
+    get_pfs_version,
     get_port_from_config,
     get_started_service_info,
     is_port_in_use,
     is_run_from_built_binary,
     kill_exist_service,
 )
-from promptflow._sdk._utils import get_promptflow_sdk_version
 from promptflow._utils.logger_utils import get_cli_sdk_logger  # noqa: E402
 from promptflow.exceptions import UserErrorException
 
@@ -193,7 +193,7 @@ def validate_port(port, force_start):
                 app.logger.setLevel(logging.DEBUG)
             else:
                 app.logger.setLevel(logging.INFO)
-            message = f"Start Prompt Flow Service on {port}, version: {get_promptflow_sdk_version()}."
+            message = f"Start Prompt Flow Service on {port}, version: {get_pfs_version()}."
             app.logger.info(message)
             print(message)
             sys.stdout.flush()
@@ -259,20 +259,20 @@ def validate_port(port, force_start):
             subprocess.Popen(cmd, stdout=subprocess.DEVNULL, start_new_session=True)
         is_healthy = check_pfs_service_status(port)
         if is_healthy:
-            message = f"Start Prompt Flow Service on port {port}, version: {get_promptflow_sdk_version()}."
+            message = f"Start Promptflow Service on port {port}, version: {get_pfs_version()}."
             print(message)
             logger.info(message)
         else:
-            logger.warning(f"Pfs service start failed in {port}.")
+            logger.warning(f"Promptflow service start failed in {port}.")
 
 
 def stop_service():
     port = get_port_from_config()
     if port is not None and is_port_in_use(port):
         kill_exist_service(port)
-        message = f"Pfs service stop in {port}."
+        message = f"Promptflow service stop in {port}."
     else:
-        message = "Pfs service is not started."
+        message = "Promptflow service is not started."
     logger.debug(message)
     print(message)
 
diff --git a/src/promptflow-devkit/promptflow/_cli/_pf/_upgrade.py b/src/promptflow-devkit/promptflow/_cli/_pf/_upgrade.py
index c0c0d06e646..8b434e97aa1 100644
--- a/src/promptflow-devkit/promptflow/_cli/_pf/_upgrade.py
+++ b/src/promptflow-devkit/promptflow/_cli/_pf/_upgrade.py
@@ -40,17 +40,18 @@ def upgrade_version(args):
     from packaging.version import parse
 
     from promptflow._constants import _ENV_PF_INSTALLER, CLI_PACKAGE_NAME
+    from promptflow._sdk._utils import get_promptflow_sdk_version
     from promptflow._utils.version_hint_utils import get_latest_version
-    from promptflow._version import VERSION as local_version
 
     installer = os.getenv(_ENV_PF_INSTALLER) or ""
     installer = installer.upper()
     print(f"installer: {installer}")
     latest_version = get_latest_version(CLI_PACKAGE_NAME, installer=installer)
+    local_version = get_promptflow_sdk_version()
     if not latest_version:
         logger.warning("Failed to get the latest prompt flow version.")
         return
-    elif parse(latest_version) <= parse(local_version):
+    elif local_version and parse(latest_version) <= parse(local_version):
         logger.warning("You already have the latest prompt flow version: %s", local_version)
         return
 
@@ -109,6 +110,8 @@ def upgrade_version(args):
     importlib.reload(json)
 
     version_result = subprocess.check_output(["pf", "version"], shell=platform.system() == "Windows")
+    # Remove ANSI codes which control color and format of text in the console output.
+    version_result = version_result.decode().replace("\x1b[0m", "").strip()
     version_json = json.loads(version_result)
     new_version = version_json["promptflow"]
 
diff --git a/src/promptflow-devkit/promptflow/_cli/_pf/entry.py b/src/promptflow-devkit/promptflow/_cli/_pf/entry.py
index 83a8fbd4342..52b95c81260 100644
--- a/src/promptflow-devkit/promptflow/_cli/_pf/entry.py
+++ b/src/promptflow-devkit/promptflow/_cli/_pf/entry.py
@@ -2,7 +2,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # ---------------------------------------------------------
 # pylint: disable=wrong-import-position
-import json
 import time
 
 from promptflow._cli._pf._experiment import add_experiment_parser, dispatch_experiment_commands
@@ -28,13 +27,7 @@
 from promptflow._cli._pf._upgrade import add_upgrade_parser, upgrade_version  # noqa: E402
 from promptflow._cli._pf.help import show_privacy_statement, show_welcome_message  # noqa: E402
 from promptflow._cli._user_agent import USER_AGENT  # noqa: E402
-from promptflow._sdk._utils import (  # noqa: E402
-    get_promptflow_core_version,
-    get_promptflow_devkit_version,
-    get_promptflow_sdk_version,
-    get_promptflow_tracing_version,
-    print_pf_version,
-)
+from promptflow._sdk._utils import print_pf_version, print_promptflow_version_dict_string  # noqa: E402
 from promptflow._utils.logger_utils import get_cli_sdk_logger  # noqa: E402
 from promptflow._utils.user_agent_utils import setup_user_agent_to_operation_context  # noqa: E402
 
@@ -138,24 +131,7 @@ def main():
     """Entrance of pf CLI."""
     command_args = sys.argv[1:]
     if len(command_args) == 1 and command_args[0] == "version":
-        version_dict = {"promptflow": get_promptflow_sdk_version()}
-        # check tracing version
-        version_tracing = get_promptflow_tracing_version()
-        if version_tracing:
-            version_dict["promptflow-tracing"] = version_tracing
-        # check core version
-        version_core = get_promptflow_core_version()
-        if version_core:
-            version_dict["promptflow-core"] = version_core
-        # check devkit version
-        version_devkit = get_promptflow_devkit_version()
-        if version_devkit:
-            version_dict["promptflow-devkit"] = version_devkit
-
-        version_dict_string = (
-            json.dumps(version_dict, ensure_ascii=False, indent=2, sort_keys=True, separators=(",", ": ")) + "\n"
-        )
-        print(version_dict_string)
+        print_promptflow_version_dict_string()
         return
     if len(command_args) == 0:
         # print privacy statement & welcome message like azure-cli
diff --git a/src/promptflow-devkit/promptflow/_proxy/_python_executor_proxy.py b/src/promptflow-devkit/promptflow/_proxy/_python_executor_proxy.py
index f09051ad908..0920c7397db 100644
--- a/src/promptflow-devkit/promptflow/_proxy/_python_executor_proxy.py
+++ b/src/promptflow-devkit/promptflow/_proxy/_python_executor_proxy.py
@@ -9,6 +9,7 @@
 from promptflow._core._errors import UnexpectedError
 from promptflow._core.run_tracker import RunTracker
 from promptflow._sdk._constants import FLOW_META_JSON_GEN_TIMEOUT, FLOW_TOOLS_JSON_GEN_TIMEOUT
+from promptflow._sdk._utils import can_accept_kwargs
 from promptflow._utils.flow_utils import resolve_entry_file
 from promptflow._utils.logger_utils import bulk_logger
 from promptflow._utils.yaml_utils import load_yaml
@@ -60,9 +61,15 @@ async def create(
         init_kwargs: Optional[Dict[str, Any]] = None,
         **kwargs,
     ) -> "PythonExecutorProxy":
-        flow_executor = FlowExecutor.create(
-            flow_file, connections, working_dir, storage=storage, raise_ex=False, init_kwargs=init_kwargs, **kwargs
-        )
+        # Check if the method accepts kwargs in case of customer using an outdated version of core package.
+        if can_accept_kwargs(FlowExecutor.create):
+            flow_executor = FlowExecutor.create(
+                flow_file, connections, working_dir, storage=storage, raise_ex=False, init_kwargs=init_kwargs, **kwargs
+            )
+        else:
+            flow_executor = FlowExecutor.create(
+                flow_file, connections, working_dir, storage=storage, raise_ex=False, init_kwargs=init_kwargs
+            )
         return cls(flow_executor)
 
     async def exec_aggregation_async(
diff --git a/src/promptflow-devkit/promptflow/_sdk/_service/app.py b/src/promptflow-devkit/promptflow/_sdk/_service/app.py
index c25914eedb9..1ce1e37fc2d 100644
--- a/src/promptflow-devkit/promptflow/_sdk/_service/app.py
+++ b/src/promptflow-devkit/promptflow/_sdk/_service/app.py
@@ -33,18 +33,19 @@
 from promptflow._sdk._service.utils.utils import (
     FormattedException,
     get_current_env_pfs_file,
+    get_pfs_version,
     get_port_from_config,
     is_run_from_built_binary,
     kill_exist_service,
 )
-from promptflow._sdk._utils import get_promptflow_sdk_version, overwrite_null_std_logger, read_write_by_user
+from promptflow._sdk._utils import overwrite_null_std_logger, read_write_by_user
 from promptflow._utils.thread_utils import ThreadWithContextVars
 
 overwrite_null_std_logger()
 
 
 def heartbeat():
-    response = {"promptflow": get_promptflow_sdk_version()}
+    response = {"promptflow": get_pfs_version()}
     return jsonify(response)
 
 
diff --git a/src/promptflow-devkit/promptflow/_sdk/_service/utils/utils.py b/src/promptflow-devkit/promptflow/_sdk/_service/utils/utils.py
index d4e9c0994e4..44e7edfff1a 100644
--- a/src/promptflow-devkit/promptflow/_sdk/_service/utils/utils.py
+++ b/src/promptflow-devkit/promptflow/_sdk/_service/utils/utils.py
@@ -28,7 +28,7 @@
     PF_SERVICE_PORT_FILE,
 )
 from promptflow._sdk._errors import ConnectionNotFoundError, RunNotFoundError
-from promptflow._sdk._utils import get_promptflow_sdk_version, read_write_by_user
+from promptflow._sdk._utils import get_promptflow_devkit_version, get_promptflow_sdk_version, read_write_by_user
 from promptflow._sdk._version import VERSION
 from promptflow._utils.logger_utils import get_cli_sdk_logger
 from promptflow._utils.yaml_utils import dump_yaml, load_yaml
@@ -155,6 +155,16 @@ def make_response_no_content():
     return make_response("", 204)
 
 
+def get_pfs_version():
+    """Promptflow service show promptflow version if installed from root, else devkit version"""
+    version_promptflow = get_promptflow_sdk_version()
+    if version_promptflow:
+        return version_promptflow
+    else:
+        version_devkit = get_promptflow_devkit_version()
+        return version_devkit
+
+
 def is_pfs_service_healthy(pfs_port) -> bool:
     """Check if pfs service is running and pfs version matches pf version."""
     try:
@@ -164,15 +174,16 @@ def is_pfs_service_healthy(pfs_port) -> bool:
             match = re.search(r'"promptflow":"(.*?)"', response.text)
             if match:
                 version = match.group(1)
-                is_healthy = version == get_promptflow_sdk_version()
+                local_version = get_pfs_version()
+                is_healthy = version == local_version
                 if not is_healthy:
                     logger.warning(
                         f"Promptflow service is running on port {pfs_port}, but the version is not the same as "
-                        f"promptflow sdk version {get_promptflow_sdk_version()}. The service version is {version}."
+                        f"local sdk version {local_version}. The service version is {version}."
                     )
             else:
                 is_healthy = False
-                logger.warning("/heartbeat response doesn't contain current pfs version.")
+                logger.warning("/heartbeat response doesn't contain current promptflow service version.")
             return is_healthy
     except Exception:  # pylint: disable=broad-except
         pass
diff --git a/src/promptflow-devkit/promptflow/_sdk/_utils.py b/src/promptflow-devkit/promptflow/_sdk/_utils.py
index a3758e6a453..794e5c36f90 100644
--- a/src/promptflow-devkit/promptflow/_sdk/_utils.py
+++ b/src/promptflow-devkit/promptflow/_sdk/_utils.py
@@ -5,6 +5,7 @@
 import datetime
 import hashlib
 import importlib
+import inspect
 import json
 import os
 import platform
@@ -31,7 +32,6 @@
 from keyring.errors import NoKeyringError
 from marshmallow import ValidationError
 
-import promptflow
 from promptflow._constants import ENABLE_MULTI_CONTAINER_KEY, EXTENSION_UA, FLOW_FLEX_YAML, FlowLanguage
 from promptflow._core.entry_meta_generator import generate_flow_meta as _generate_flow_meta
 from promptflow._sdk._constants import (
@@ -328,9 +328,11 @@ def incremental_print(log: str, printed: int, fileout) -> int:
 
 def get_promptflow_sdk_version() -> str:
     try:
+        import promptflow
+
         return promptflow.__version__
-    except ImportError:
-        # if promptflow is installed from source, it does not have __version__ attribute
+    except (ImportError, AttributeError):
+        # if promptflow is not installed from root, it does not have __version__ attribute
         return None
 
 
@@ -370,24 +372,36 @@ def get_promptflow_azure_version() -> Union[str, None]:
         return None
 
 
-def print_pf_version(with_azure: bool = False):
-    version_promptflow = get_promptflow_sdk_version()
-    if version_promptflow:
-        print("promptflow\t\t\t {}".format(version_promptflow))
+def print_promptflow_version_dict_string(with_azure: bool = False, ignore_none: bool = False):
+    version_dict = {"promptflow": get_promptflow_sdk_version()}
+    # check tracing version
     version_tracing = get_promptflow_tracing_version()
     if version_tracing:
-        print("promptflow-tracing\t\t {}".format(version_tracing))
+        version_dict["promptflow-tracing"] = version_tracing
+    # check core version
     version_core = get_promptflow_core_version()
     if version_core:
-        print("promptflow-core\t\t\t {}".format(version_core))
+        version_dict["promptflow-core"] = version_core
+    # check devkit version
     version_devkit = get_promptflow_devkit_version()
     if version_devkit:
-        print("promptflow-devkit\t\t {}".format(version_devkit))
+        version_dict["promptflow-devkit"] = version_devkit
+
     if with_azure:
+        # check azure version
         version_azure = get_promptflow_azure_version()
         if version_azure:
-            print("promptflow-azure\t\t {}".format(version_azure))
-    print()
+            version_dict["promptflow-azure"] = version_azure
+    if ignore_none:
+        version_dict = {k: v for k, v in version_dict.items() if v is not None}
+    version_dict_string = (
+        json.dumps(version_dict, ensure_ascii=False, indent=2, sort_keys=True, separators=(",", ": ")) + "\n"
+    )
+    print(version_dict_string)
+
+
+def print_pf_version(with_azure: bool = False, ignore_none: bool = False):
+    print_promptflow_version_dict_string(with_azure, ignore_none)
     print("Executable '{}'".format(os.path.abspath(sys.executable)))
     print("Python ({}) {}".format(platform.system(), sys.version))
 
@@ -1007,6 +1021,12 @@ def create_temp_flex_flow_yaml(entry: Union[str, PathLike, Callable], code: Path
                     logger.warning(f"Failed to delete generated: {flow_yaml_path.as_posix()}, error: {e}")
 
 
+def can_accept_kwargs(func):
+    sig = inspect.signature(func)
+    params = sig.parameters.values()
+    return any(param.kind == param.VAR_KEYWORD for param in params)
+
+
 def callable_to_entry_string(callable_obj: Callable) -> str:
     """Convert callable object to entry string."""
     if not isfunction(callable_obj):
diff --git a/src/promptflow/tests/executor/unittests/executor/_service/utils/test_batch_coordinator.py b/src/promptflow/tests/executor/unittests/executor/_service/utils/test_batch_coordinator.py
new file mode 100644
index 00000000000..1293b917c09
--- /dev/null
+++ b/src/promptflow/tests/executor/unittests/executor/_service/utils/test_batch_coordinator.py
@@ -0,0 +1,37 @@
+import pytest
+
+from promptflow.executor._service.utils.batch_coordinator import BatchCoordinator
+
+from .....utils import get_flow_folder
+
+
+@pytest.mark.unittest
+class TestBatchCoordinator:
+    @pytest.fixture(autouse=True)
+    def resource_setup_and_teardown(self):
+        # Setup code goes here
+
+        # Yield to the test function
+        yield
+
+        # Teardown code goes here
+        # Set _instance to None to avoid singleton pattern
+        # Seems like BatchCoordinator.get_instance().close() can't be called if we don't call start first.
+        BatchCoordinator._instance = None
+
+    @pytest.mark.parametrize(
+        "file_name, name_from_payload, expected_name",
+        [
+            ("yaml_with_name.yaml", "name_from_payload", "name_from_payload"),
+            ("yaml_with_name.yaml", None, "name_from_yaml"),
+            ("yaml_without_name.yaml", "name_from_payload", "name_from_payload"),
+            ("yaml_without_name.yaml", None, "flow_name"),
+        ],
+    )
+    @pytest.mark.asyncio
+    async def test_executor_flow_name(self, file_name, name_from_payload, expected_name):
+        flow_folder = get_flow_folder("flow_name")
+        coordinator = BatchCoordinator(
+            working_dir=flow_folder, flow_file=file_name, output_dir="", flow_name=name_from_payload
+        )
+        assert coordinator._flow_executor._flow.name == expected_name
diff --git a/src/promptflow/tests/sdk_cli_azure_test/unittests/test_cli.py b/src/promptflow/tests/sdk_cli_azure_test/unittests/test_cli.py
index 1763de37276..25ec32e7b06 100644
--- a/src/promptflow/tests/sdk_cli_azure_test/unittests/test_cli.py
+++ b/src/promptflow/tests/sdk_cli_azure_test/unittests/test_cli.py
@@ -48,7 +48,7 @@ class TestAzureCli:
     def test_pf_azure_version(self, capfd):
         run_pf_command("--version")
         out, err = capfd.readouterr()
-        assert "0.0.1\n" in out
+        assert "0.0.1" in out
 
     def test_run_show(self, mocker: MockFixture, operation_scope_args):
         from promptflow.azure.operations._run_operations import RunOperations
diff --git a/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py b/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py
index de60aa6693d..e6e670c9c8b 100644
--- a/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py
+++ b/src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py
@@ -69,7 +69,7 @@ class TestCli:
     def test_pf_version(self, capfd):
         run_pf_command("--version")
         out, _ = capfd.readouterr()
-        assert "0.0.1\n" in out
+        assert "0.0.1" in out
 
     def test_basic_flow_run(self, capfd) -> None:
         # fetch std out
diff --git a/src/promptflow/tests/sdk_pfs_test/e2etests/test_general_apis.py b/src/promptflow/tests/sdk_pfs_test/e2etests/test_general_apis.py
index 27fd14eba60..f03288477d8 100644
--- a/src/promptflow/tests/sdk_pfs_test/e2etests/test_general_apis.py
+++ b/src/promptflow/tests/sdk_pfs_test/e2etests/test_general_apis.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from promptflow._sdk._utils import get_promptflow_sdk_version
+from promptflow._sdk._service.utils.utils import get_pfs_version
 
 from ..utils import PFSOperations
 
@@ -17,4 +17,4 @@ def test_heartbeat(self, pfs_op: PFSOperations) -> None:
         response_json = response.json
         assert isinstance(response_json, dict)
         assert "promptflow" in response_json
-        assert response_json["promptflow"] == get_promptflow_sdk_version()
+        assert response_json["promptflow"] == get_pfs_version()