Repository: deepset-ai/haystack
Pull request: feat: Change `Pipeline.add_component` to fail when reusing `Component` instances
### Related Issues

- fixes #6431

### Proposed Changes:

Add `__haystack_pipeline_owned__` field in all `Component` instances. Defaults to `False`.

Change `Pipeline.add_component()` to fail if the `Component` instance is already used in another `Pipeline`. The method will also set `__haystack_pipeline_owned__` to `True`.

### How did you test it?

Added a unit test.

### Notes for the reviewer

N/A

### 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:`.
- 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:
Warn users when adding the same component instance to multiple pipelines
Adding the same component instance to different pipelines causes side effects that often result in unexpected behaviour.

Except for any operation done during the `warm_up` phase, components are stateless so there's no gain in adding the same instance to different pipelines, and our stance at this point in time is we don't support it.

We need to document this and add a meaningful error in `add_component` to detect if an instance is already part of another pipeline, and in case failing the operation explaining why.

Hints from issue discussion:

Commit message: Change Pipeline.add_component to fail when reusing Component instances

Patch to review:
diff --git a/haystack/core/component/component.py b/haystack/core/component/component.py
index ab264ce11a..223fb2f1b4 100644
--- a/haystack/core/component/component.py
+++ b/haystack/core/component/component.py
@@ -147,6 +147,12 @@ def __call__(cls, *args, **kwargs):
                 if run_signature.parameters[param].default != inspect.Parameter.empty:
                     socket_kwargs["default_value"] = run_signature.parameters[param].default
                 instance.__haystack_input__[param] = InputSocket(**socket_kwargs)
+
+        # Since a Component can't be used in multiple Pipelines at the same time
+        # we need to know if it's already owned by a Pipeline when adding it to one.
+        # We use this flag to check that.
+        instance.__haystack_pipeline_owned__ = False
+
         return instance
 
 
diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py
index 4339db9e03..72db999546 100644
--- a/haystack/core/pipeline/pipeline.py
+++ b/haystack/core/pipeline/pipeline.py
@@ -190,10 +190,19 @@ def add_component(self, name: str, instance: Component) -> None:
                 f"'{type(instance)}' doesn't seem to be a component. Is this class decorated with @component?"
             )
 
+        if getattr(instance, "__haystack_pipeline_owned__", False):
+            msg = (
+                "Component has already been added in another Pipeline. "
+                "Components can't be shared between Pipelines. Create a new instance instead."
+            )
+            raise PipelineError(msg)
+
         # Create the component's input and output sockets
         input_sockets = getattr(instance, "__haystack_input__", {})
         output_sockets = getattr(instance, "__haystack_output__", {})
 
+        setattr(instance, "__haystack_pipeline_owned__", True)
+
         # Add component to the graph, disconnected
         logger.debug("Adding component '%s' (%s)", name, instance)
         self.graph.add_node(
diff --git a/releasenotes/notes/components-reuse-ccc5f2d4b9d7f9ff.yaml b/releasenotes/notes/components-reuse-ccc5f2d4b9d7f9ff.yaml
new file mode 100644
index 0000000000..1ef0554411
--- /dev/null
+++ b/releasenotes/notes/components-reuse-ccc5f2d4b9d7f9ff.yaml
@@ -0,0 +1,4 @@
+---
+enhancements:
+  - |
+    Change `Pipeline.add_component()` to fail if the `Component` instance has already been added in another `Pipeline`.
diff --git a/test/core/pipeline/test_pipeline.py b/test/core/pipeline/test_pipeline.py
index 2a0d4f9b89..03104a17e8 100644
--- a/test/core/pipeline/test_pipeline.py
+++ b/test/core/pipeline/test_pipeline.py
@@ -7,14 +7,27 @@
 import pytest
 
 from haystack.core.component.sockets import InputSocket, OutputSocket
-from haystack.core.errors import PipelineError, PipelineMaxLoops, PipelineRuntimeError
+from haystack.core.errors import PipelineError, PipelineRuntimeError
 from haystack.core.pipeline import Pipeline
 from haystack.testing.factory import component_class
-from haystack.testing.sample_components import AddFixedValue, Double, Sum, Threshold
+from haystack.testing.sample_components import AddFixedValue, Double
 
 logging.basicConfig(level=logging.DEBUG)
 
 
+def test_add_component_to_different_pipelines():
+    first_pipe = Pipeline()
+    second_pipe = Pipeline()
+    some_component = component_class("Some")()
+
+    assert not some_component.__haystack_pipeline_owned__
+    first_pipe.add_component("some", some_component)
+    assert some_component.__haystack_pipeline_owned__
+
+    with pytest.raises(PipelineError):
+        second_pipe.add_component("some", some_component)
+
+
 def test_run_with_component_that_does_not_return_dict():
     BrokenComponent = component_class(
         "BrokenComponent", input_types={"a": int}, output_types={"b": int}, output=1  # type:ignore


Code review comments:
diff_hunk: @@ -147,6 +147,12 @@ def __call__(cls, *args, **kwargs):
                 if run_signature.parameters[param].default != inspect.Parameter.empty:
                     socket_kwargs["default_value"] = run_signature.parameters[param].default
                 instance.__haystack_input__[param] = InputSocket(**socket_kwargs)
+
+        # Since a Component can't be used in multiple Pipelines at the same time
+        # we need to know if it's already owned by a Pipeline when adding it to one.
+        # We use this flag to check that.
+        instance.__haystack_pipeline_owned__ = False
path: haystack/core/component/component.py
text: @user1:
nit: I would use a name with a simpler semantic, like `__haystack_added_to_pipeline__`

@author:
I was actually thinking of changing this to hold a reference to the `Pipeline`, it can be useful in later PRs. 🤔 
So something like `__haystack_owned_by_pipeline__`.

@user1:
Since we don't have any explicit concept of ownership in Haystack, it's the `owned` bit that bugs me 😅

@author:
Makes sense, I'll use added then.