Repository: deepset-ai/haystack
Pull request: feat: allow SuperComponent to Include outputs from non leaf pipeline components
### Related Issues

- fixes #9231 

### Proposed Changes:

- map nonleaf pipelines outputs to the supercomponenets output

### How did you test it?

- added unit test

### Notes for the reviewer

<!-- E.g. point out section where the reviewer  -->

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:` and added `!` in case the PR includes breaking changes.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
Allow SuperComponent to include outputs from already-connected components
Currently SuperComponent can only have outputs from Components that don't have their output connected. 

A workaround for this is to connect the component you would also like the output of to an isolated Joiner which you can then connect to the output. 

I believe a better solution would be to:
1) potentially expose the `include_outputs_from` parameter in the SuperComponent to allow outputs from already connected components
2) update the logic of output_mapping validation/creation to automatically populate `include_outputs_from` if we detect that the component already has its output connected. 

cc @deep-rloebbert

Hints from issue discussion:

Commit message: add rn

Patch to review:
diff --git a/haystack/core/super_component/super_component.py b/haystack/core/super_component/super_component.py
index 0cb5a80830..a47d6de7cb 100644
--- a/haystack/core/super_component/super_component.py
+++ b/haystack/core/super_component/super_component.py
@@ -129,7 +129,7 @@ def __init__(
         self._original_input_mapping = input_mapping
 
         # Set output types based on pipeline and mapping
-        pipeline_outputs = self.pipeline.outputs()
+        pipeline_outputs = self.pipeline.outputs(include_components_with_connected_outputs=True)
         resolved_output_mapping = (
             output_mapping if output_mapping is not None else self._create_output_mapping(pipeline_outputs)
         )
@@ -189,9 +189,14 @@ def run(self, **kwargs: Any) -> Dict[str, Any]:
         """
         filtered_inputs = {param: value for param, value in kwargs.items() if value != _delegate_default}
         pipeline_inputs = self._map_explicit_inputs(input_mapping=self.input_mapping, inputs=filtered_inputs)
-        pipeline_outputs = self.pipeline.run(data=pipeline_inputs)
+        include_outputs_from = self._get_include_outputs_from()
+        pipeline_outputs = self.pipeline.run(data=pipeline_inputs, include_outputs_from=include_outputs_from)
         return self._map_explicit_outputs(pipeline_outputs, self.output_mapping)
 
+    def _get_include_outputs_from(self) -> set[str]:
+        # Collecting the component names from output_mapping
+        return {self._split_component_path(path)[0] for path in self.output_mapping.keys()}
+
     async def run_async(self, **kwargs: Any) -> Dict[str, Any]:
         """
         Runs the wrapped pipeline with the provided inputs async.
diff --git a/releasenotes/notes/allow-non-leaf-outputs-in-supercomponents-outputs-adf29d68636c23ba.yaml b/releasenotes/notes/allow-non-leaf-outputs-in-supercomponents-outputs-adf29d68636c23ba.yaml
new file mode 100644
index 0000000000..2446f1f27a
--- /dev/null
+++ b/releasenotes/notes/allow-non-leaf-outputs-in-supercomponents-outputs-adf29d68636c23ba.yaml
@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    map nonleaf pipelines outputs to the supercomponents output.
diff --git a/test/core/super_component/test_super_component.py b/test/core/super_component/test_super_component.py
index 825e57fdcc..6f0dbed87d 100644
--- a/test/core/super_component/test_super_component.py
+++ b/test/core/super_component/test_super_component.py
@@ -278,3 +278,13 @@ def from_dict(cls, data):
 
         assert custom_serialized["type"] == "test_super_component.CustomSuperComponent"
         assert custom_super_component._to_super_component_dict() == serialized
+
+    def test_super_component_non_leaf_output(self, rag_pipeline):
+        # 'retriever' is not a leaf, but should now be allowed
+        output_mapping = {"retriever.documents": "retrieved_docs", "answer_builder.answers": "final_answers"}
+        wrapper = SuperComponent(pipeline=rag_pipeline, output_mapping=output_mapping)
+        wrapper.warm_up()
+        result = wrapper.run(query="What is the capital of France?")
+        assert "final_answers" in result  # leaf output
+        assert "retrieved_docs" in result  # non-leaf output
+        assert isinstance(result["retrieved_docs"][0], Document)


Code review comments:
diff_hunk: @@ -129,7 +129,7 @@ def __init__(
         self._original_input_mapping = input_mapping
 
         # Set output types based on pipeline and mapping
-        pipeline_outputs = self.pipeline.outputs()
+        pipeline_outputs = self.pipeline.outputs(include_components_with_connected_outputs=True)
path: haystack/core/super_component/super_component.py
text: @user1:
~~We should make `include_components_with_connected_outputs` configurable in the init of SuperComponent and set it to False as default.~~

@user1:
Okay this is a bit more nuanced than I thought. I think the code block should actually be something like
```python
# Set output types based on pipeline and mapping
all_possible_pipeline_outputs = self.pipeline.outputs(include_components_with_connected_outputs=True)
# HERE: In this step pass self.pipeline.outputs(), 
# we don't want to include all possible pipeline outputs by default 
# when auto creating output_mapping
resolved_output_mapping = (
    output_mapping if output_mapping is not None else self._create_output_mapping(self.pipeline.outputs())
)
self._validate_output_mapping(all_possible_pipeline_outputs, resolved_output_mapping)
output_types = self._resolve_output_types_from_mapping(all_possible_pipeline_outputs, resolved_output_mapping)
```

@user1:
Using the above code block should also fix the failing tests