[Promptflow CLI] Write exception info to stderr when set env PROMPTFL…

…OW_STRUCTURE_EXCEPTION_OUTPUT (#331)

# Description

Write exception info and command to stderr when set environment variable
`PROMPTFLOW_STRUCTURE_EXCEPTION_OUTPUT=true`

![image](https://github.com/microsoft/promptflow/assets/17938940/8c3f21ce-b62d-46cb-b7a9-a2e6de3337b3)

promptflow error in stderr:
``` json
{
    "message":"Connection 'invalid_name' is not found.",
    "messageFormat":"",
    "messageParameters":{

    },
    "referenceCode":"Unknown",
    "code":"ConnectionNotFoundError",
    "innerError":null,
    "debugInfo":{
        "type":"ConnectionNotFoundError",
        "message":"Connection 'invalid_name' is not found.",
        "stackTrace":"Traceback (most recent call last):\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\promptflow\\_cli\\_utils.py\", line 354, in wrapper\n    return func(*args, **kwargs)\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\promptflow\\_cli\\_pf\\_connection.py\", line 187, in show_connection\n    connection = _client.connections.get(name)\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\promptflow\\_sdk\\operations\\_connection_operations.py\", line 47, in get\n    orm_connection = ORMConnection.get(name, raise_error)\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\promptflow\\_sdk\\_orm\\retry.py\", line 43, in f_retry\n    return f(*args, **kwargs)\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\promptflow\\_sdk\\_orm\\connection.py\", line 52, in get\n    raise ConnectionNotFoundError(f\"Connection {name!r} is not found.\")\n",
        "innerException":null
    },
    "command":"pf connection show --name invalid_name --format_exception"
}
```

system error in stderr:
``` json
{
    "code":"SystemError",
    "message":"mock exception",
    "messageFormat":"",
    "messageParameters":{

    },
    "innerError":{
        "code":"Exception",
        "innerError":null
    },
    "debugInfo":{
        "type":"Exception",
        "message":"mock exception",
        "stackTrace":"Traceback (most recent call last):\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\promptflow\\_cli\\_utils.py\", line 354, in wrapper\n    return func(*args, **kwargs)\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\promptflow\\_cli\\_pf\\_connection.py\", line 187, in show_connection\n    connection = _client.connections.get(name)\n  File \"C:\\Users\\zhrua\\Miniconda3\\envs\\prompt_github\\lib\\unittest\\mock.py\", line 1093, in __call__\n    return self._mock_call(*args, **kwargs)\n  File \"C:\\Users\\zhrua\\Miniconda3\\envs\\prompt_github\\lib\\unittest\\mock.py\", line 1097, in _mock_call\n    return self._execute_mock_call(*args, **kwargs)\n  File \"C:\\Users\\zhrua\\Miniconda3\\envs\\prompt_github\\lib\\unittest\\mock.py\", line 1158, in _execute_mock_call\n    result = effect(*args, **kwargs)\n  File \"D:\\Project\\github_promptflow\\promptflow\\src\\promptflow\\tests\\sdk_cli_test\\e2etests\\test_cli.py\", line 1123, in mocked_connection_get\n    raise Exception(\"mock exception\")\n",
        "innerException":null
    },
    "command":"pf connection show --name invalid_connection_name --format_exception"
}
```


# All Promptflow Contribution checklist:
- [x] **The pull request does not introduce [breaking changes]**
- [x] **CHANGELOG is updated for new features, bug fixes or other
significant changes.**
- [x] **I have read the [contribution guidelines](../CONTRIBUTING.md).**

## General Guidelines and Best Practices
- [x] Title of the pull request is clear and informative.
- [x] 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
- [x] Pull request includes test coverage for the included changes.

src/promptflow/promptflow/_cli/_utils.py

@@ -23,6 +23,7 @@
 
 from promptflow._sdk._utils import print_red_error, print_yellow_warning
 from promptflow._utils.utils import is_in_ci_pipeline
+from promptflow._utils.exception_utils import ExceptionPresenter
 from promptflow.exceptions import ErrorTarget, PromptflowException, UserErrorException
 
 AzureMLWorkspaceTriad = namedtuple("AzureMLWorkspace", ["subscription_id", "resource_group_name", "workspace_name"])
@@ -334,6 +335,12 @@ def pretty_print_dataframe_as_table(df: pd.DataFrame) -> None:
     print(tabulate(df, headers="keys", tablefmt="grid", maxcolwidths=column_widths, maxheadercolwidths=column_widths))
 
 
+def is_format_exception():
+    if os.environ.get("PROMPTFLOW_STRUCTURE_EXCEPTION_OUTPUT", "false").lower() == "true":
+        return True
+    return False
+
+
 def exception_handler(command: str):
     """Catch known cli exceptions."""
 
@@ -342,9 +349,18 @@ def decorator(func):
         def wrapper(*args, **kwargs):
             try:
                 return func(*args, **kwargs)
-            except PromptflowException as e:
-                print_red_error(f"{command} failed with {e.__class__.__name__}: {str(e)}")
-                exit(1)
+            except Exception as e:
+                if is_format_exception():
+                    # When the flag format_exception is set in command,
+                    # it will write a json with exception info and command to stderr.
+                    error_msg = ExceptionPresenter.create(e).to_dict(include_debug_info=True)
+                    error_msg["command"] = " ".join(sys.argv)
+                    sys.stderr.write(json.dumps(error_msg))
+                if isinstance(e, PromptflowException):
+                    print_red_error(f"{command} failed with {e.__class__.__name__}: {str(e)}")
+                    exit(1)
+                else:
+                    raise e
 
         return wrapper
 

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

@@ -1102,3 +1102,46 @@ def test_pf_run_no_stream_log(self):
         assert "Executing node print_val. node run id:" not in f.getvalue()
         # executor logs won't stream
         assert "Node print_val completes." not in f.getvalue()
+
+    def test_format_cli_exception(self, capsys):
+        from promptflow._sdk.operations._connection_operations import ConnectionOperations
+
+        with patch.dict(os.environ, {"PROMPTFLOW_STRUCTURE_EXCEPTION_OUTPUT": "true"}):
+            with pytest.raises(SystemExit):
+                run_pf_command(
+                    "connection",
+                    "show",
+                    "--name",
+                    "invalid_connection_name",
+                )
+            outerr = capsys.readouterr()
+            assert outerr.err
+            error_msg = json.loads(outerr.err)
+            assert error_msg["code"] == "ConnectionNotFoundError"
+
+            def mocked_connection_get(*args, **kwargs):
+                raise Exception("mock exception")
+
+            with patch.object(ConnectionOperations, "get") as mock_connection_get:
+                mock_connection_get.side_effect = mocked_connection_get
+                with pytest.raises(Exception):
+                    run_pf_command(
+                        "connection",
+                        "show",
+                        "--name",
+                        "invalid_connection_name",
+                    )
+                outerr = capsys.readouterr()
+                assert outerr.err
+                error_msg = json.loads(outerr.err)
+                assert error_msg["code"] == "SystemError"
+
+        with pytest.raises(SystemExit):
+            run_pf_command(
+                "connection",
+                "show",
+                "--name",
+                "invalid_connection_name",
+            )
+        outerr = capsys.readouterr()
+        assert not outerr.err