fix: run show-details depends on inputs line number (#1298)

# Description

This pull request includes changes to improve the functionality and
maintainability of the codebase. The most important changes include
adding a new method to the `ExecutorProxy` class and modifying the
import statement in the `_csharp_executor_proxy.py` file, as well as
updating the `load_inputs_and_outputs` method in the
`_local_storage_operations.py` file to use a constant instead of a
string.

Main changes:

* <a
href="diffhunk://#diff-f8219170007d8a89eed104a166ab8c0c9d3af5b5c5e61225659d982f470aeb95R60-R81">`src/promptflow/promptflow/batch/_csharp_executor_proxy.py`</a>:
Added a new method `exec_line_async` to the `ExecutorProxy` class and
modified the import statement to include the `LineResult` class. <a
href="diffhunk://#diff-f8219170007d8a89eed104a166ab8c0c9d3af5b5c5e61225659d982f470aeb95R60-R81">[1]</a>
<a
href="diffhunk://#diff-f8219170007d8a89eed104a166ab8c0c9d3af5b5c5e61225659d982f470aeb95R10-R12">[2]</a>
* <a
href="diffhunk://#diff-1a5c2463c2ef997c5203a6b2e77b247d812961b7a6df5f0f45c02579b6656719L447-R447">`src/promptflow/promptflow/_sdk/operations/_local_storage_operations.py`</a>:
Updated the `load_inputs_and_outputs` method to use a constant
`LINE_NUMBER` instead of the string `&quot;line_number&quot;`.

# 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).**
- [x] **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
- [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/_sdk/operations/_local_storage_operations.py

@@ -444,7 +444,7 @@ def load_inputs_and_outputs(self) -> Tuple["DataFrame", "DataFrame"]:
                 outputs = pd.read_json(f, orient="records", lines=True)
                 # if all line runs are failed, no need to fill
                 if len(outputs) > 0:
-                    outputs = self._outputs_padding(outputs, inputs["line_number"].tolist())
+                    outputs = self._outputs_padding(outputs, inputs[LINE_NUMBER].tolist())
                     outputs.fillna(value="(Failed)", inplace=True)  # replace nan with explicit prompt
                     outputs = outputs.set_index(LINE_NUMBER)
         return inputs, outputs

src/promptflow/promptflow/batch/_csharp_executor_proxy.py

@@ -7,8 +7,9 @@
 from pathlib import Path
 from typing import Any, Mapping, Optional
 
+from promptflow._constants import LINE_NUMBER_KEY
 from promptflow.batch._base_executor_proxy import APIBasedExecutorProxy
-from promptflow.executor._result import AggregationResult
+from promptflow.executor._result import AggregationResult, LineResult
 from promptflow.storage._run_storage import AbstractRunStorage
 
 EXECUTOR_SERVICE_DOMAIN = "http://localhost:"
@@ -56,6 +57,28 @@ def create(
         process = subprocess.Popen(command)
         return cls(process, port)
 
+    async def exec_line_async(
+        self,
+        inputs: Mapping[str, Any],
+        index: Optional[int] = None,
+        run_id: Optional[str] = None,
+    ) -> LineResult:
+        line_result = await super().exec_line_async(inputs, index, run_id)
+        # TODO: check if we should ask C# executor to keep unmatched inputs, although it's not so straightforward
+        #   for executor service to do so.
+        # local_storage_operations.load_inputs_and_outputs now have an assumption that there is an extra
+        # line_number key in the inputs.
+        # This key will be appended to the inputs in below call stack:
+        # BatchEngine.run =>
+        # BatchInputsProcessor.process_batch_inputs =>
+        # ... =>
+        # BatchInputsProcessor._merge_input_dicts_by_line
+        # For python, it will be kept in the returned line_result.run_info.inputs
+        # For csharp, it will be dropped by executor service for now
+        # Append it here for now to make behavior consistent among ExecutorProxy.
+        line_result.run_info.inputs[LINE_NUMBER_KEY] = index
+        return line_result
+
     def destroy(self):
         """Destroy the executor"""
         if self._process and self._process.poll() is None: