[SDK] Fix list error when run flow snapshot is incomplete (#1811)

# Description

This pull request includes changes to improve error handling and add
tests for specific scenarios. The most important changes include adding
error handling in the `__init__` method of the `LocalStorageOperations`
class and adding code to test the behavior of accessing operations and
converting a run object to a dictionary.

Error handling improvements:

*
[`src/promptflow/promptflow/_sdk/operations/_local_storage_operations.py`](diffhunk://#diff-1a5c2463c2ef997c5203a6b2e77b247d812961b7a6df5f0f45c02579b6656719R222-R227):
Added error handling in the `__init__` method of the
`LocalStorageOperations` class to handle cases where the `dag.yaml` file
does not exist.

Testing improvements:

*
[`src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py`](diffhunk://#diff-94a59a05643476869fa3c6bc45466f1582944a935488075e2e63b6a6a196958fR1268-R1285):
Added code to test the behavior of accessing operations and converting a
run object to a dictionary after the dag has been removed.
# All Promptflow Contribution checklist:
- [ ] **The pull request does not introduce [breaking changes].**
- [ ] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [ ] **I have read the [contribution guidelines](../CONTRIBUTING.md).**
- [ ] **Create an issue and link to the pull request to get dedicated
review from promptflow team. Learn more: [suggested
workflow](../CONTRIBUTING.md#suggested-workflow).**

## General Guidelines and Best Practices
- [ ] Title of the pull request is clear and informative.
- [ ] There are a small number of commits, each of which have an
informative message. This means that previously merged commits do not
appear in the history of the PR. For more information on cleaning up the
commits in your PR, [see this
page](https://github.com/Azure/azure-powershell/blob/master/documentation/development-docs/cleaning-up-commits.md).

### Testing Guidelines
- [ ] Pull request includes test coverage for the included changes.

src/promptflow/promptflow/_cli/_pf/_run.py

@@ -33,6 +33,7 @@
 from promptflow._sdk._load_functions import load_run
 from promptflow._sdk._pf_client import PFClient
 from promptflow._sdk._run_functions import _create_run
+from promptflow._sdk._utils import safe_parse_object_list
 from promptflow._sdk.entities import Run
 from promptflow.exceptions import UserErrorException
 
@@ -502,7 +503,12 @@ def list_runs(
         list_view_type=get_list_view_type(archived_only=archived_only, include_archived=include_archived),
     )
     # hide additional info and debug info in run list for better user experience
-    json_list = [run._to_dict(exclude_additional_info=True, exclude_debug_info=True) for run in runs]
+    parser = lambda run: run._to_dict(exclude_additional_info=True, exclude_debug_info=True)  # noqa: E731
+    json_list = safe_parse_object_list(
+        obj_list=runs,
+        parser=parser,
+        message_generator=lambda x: f"Error parsing run {x.name!r}, skipped.",
+    )
     _output_result_list_with_format(result_list=json_list, output_format=output)
     return runs
 

src/promptflow/promptflow/_sdk/operations/_local_storage_operations.py

@@ -215,9 +215,13 @@ def __init__(self, run: Run, stream=False, run_mode=RunMode.Test):
 
         self._dump_meta_file()
         if run.flow:
-            flow_obj = load_flow(source=run.flow)
-            # TODO(2898455): refine here, check if there's cases where dag.yaml not exist
-            self._eager_mode = isinstance(flow_obj, EagerFlow)
+            try:
+                flow_obj = load_flow(source=run.flow)
+                self._eager_mode = isinstance(flow_obj, EagerFlow)
+            except Exception as e:
+                # For run with incomplete flow snapshot, ignore load flow error to make sure it can still show.
+                logger.debug(f"Failed to load flow from {run.flow} due to {e}.")
+                self._eager_mode = False
         else:
             # TODO(2901279): support eager mode for run created from run folder
             self._eager_mode = False

src/promptflow/tests/sdk_cli_test/e2etests/test_cli.py

@@ -1948,3 +1948,13 @@ def test_batch_run_completed_within_the_required_time(self, local_client):
         p.start()
         p.join()
         assert p.exitcode == 0
+
+    def test_run_list(self, local_client):
+        from promptflow._sdk.entities import Run
+
+        with patch.object(Run, "_to_dict") as mock_to_dict:
+            mock_to_dict.side_effect = RuntimeError("mock exception")
+            run_pf_command(
+                "run",
+                "list",
+            )

src/promptflow/tests/sdk_cli_test/e2etests/test_flow_run.py

@@ -1265,3 +1265,21 @@ def test_eager_flow_test_invalid_cases(self, pf):
                 data=f"{DATAS_DIR}/simple_eager_flow_data.jsonl",
             )
         assert "'path': ['Missing data for required field.']" in str(e.value)
+
+    def test_get_incomplete_run(self, local_client, pf) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            shutil.copytree(f"{FLOWS_DIR}/print_env_var", f"{temp_dir}/print_env_var")
+
+            run = pf.run(
+                flow=f"{temp_dir}/print_env_var",
+                data=f"{DATAS_DIR}/env_var_names.jsonl",
+            )
+
+            # remove run dag
+            shutil.rmtree(f"{temp_dir}/print_env_var")
+
+            # can still get run operations
+            LocalStorageOperations(run=run)
+
+            # can to_dict
+            run._to_dict()