fix: allow multi-connection to Any-typed pipeline sockets

haystack/core/pipeline/base.py

@@ -1068,10 +1068,17 @@ def _make_socket_auto_variadic(
         A socket is automatically made lazy variadic when:
           - It already has at least one connected sender
           - It is not already variadic
-          - Its type is list, Optional[list], a union of list types
-
-        When auto-variadicity is applied, `wrap_input_in_list` is also set to False so that sender output types match
-        the receiver socket's declared list type directly.
+          - Its type is list, Optional[list], a union of list types, or ``Any`` (the type-erased case;
+            common for components like ``PromptBuilder`` that register template variables as ``Any``)
+
+        When auto-variadicity is applied:
+          - For a list-shaped receiver (``list`` / ``Optional[list]`` / union of lists), ``wrap_input_in_list``
+            is set to ``False`` so the framework flattens one level and the component receives a single
+            concatenated list, matching the declared socket type.
+          - For an ``Any``-typed receiver, ``wrap_input_in_list`` is left at ``True`` so the component
+            receives a list of the per-sender values (``[sender_1, sender_2, ...]``) and can decide how
+            to handle the type-erased inputs (a component that knows it always aggregates a list of
+            lists may flatten them in ``run()``).
 
         :param component_name:
             Name of the component owning the receiver socket, used in error messages.
@@ -1088,6 +1095,20 @@ def _make_socket_auto_variadic(
         if receiver_socket.is_variadic:
             return receiver_socket
 
+        # An ``Any``-typed socket can accept inputs from multiple senders. The framework cannot
+        # safely flatten one level (it does not know the element type), so the component receives
+        # ``[sender_1, sender_2, ...]`` via the existing ``wrap_input_in_list=True`` path. This is
+        # the gap left after #10783, whose commit message claimed ``Any`` support but only
+        # implemented ``list`` / ``Optional[list]`` / union-of-list handling. The most visible
+        # beneficiary is ``PromptBuilder``: every template variable is registered as ``Any``
+        # (``component.set_input_type(self, var, Any)``), so a multi-connection from two
+        # retrievers to ``prompt_builder.documents`` used to fail before this branch was added.
+        if receiver_socket.type is Any:
+            receiver_socket.is_lazy_variadic = True
+            # ``wrap_input_in_list`` stays at its current value (True by default). The component
+            # sees ``[sender_1, sender_2, ...]`` and can flatten in its own ``run()`` if needed.
+            return receiver_socket
+
         # Get receiver origin
         receiver_origin = _safe_get_origin(receiver_socket.type)
 
@@ -1110,7 +1131,7 @@ def _make_socket_auto_variadic(
         raise error_type(
             f"Component '{component_name}' cannot accept multiple inputs to '{receiver_socket.name}'. "
             f"It is already connected to component '{receiver_socket.senders[0]}', and it can only accept "
-            f"inputs from multiple senders if its type is list, Optional[list], or union of list types."
+            f"inputs from multiple senders if its type is list, Optional[list], union of list types, or Any."
         )
 
     def _prepare_component_input_data(self, data: dict[str, Any]) -> dict[str, dict[str, Any]]:

releasenotes/notes/fix-pipeline-any-typed-multi-connection-b3e7d2f8a1c90546.yaml

@@ -0,0 +1,14 @@
+---
+fixes:
+  - |
+    Fixed ``Pipeline`` ``connect`` rejecting a second connection to an ``Any``-typed
+    input socket (e.g. ``PromptBuilder.documents``) with a ``PipelineConnectError``.
+    The previous behaviour was the gap left after #10783 — its commit message claimed
+    ``Any``-type support, but ``_make_socket_auto_variadic`` only handled ``list``,
+    ``Optional[list]``, and union-of-list receivers. An ``Any``-typed receiver is now
+    made lazy variadic on the second connection so multiple senders can attach. The
+    framework cannot safely flatten one level for an ``Any``-typed socket (it has no
+    element type to enforce), so the receiver gets ``[sender_1, sender_2, ...]`` and
+    is responsible for any per-sender aggregation. This is the connection-eligibility
+    half of #10721; rendering-side aggregation (e.g. flattening inside
+    ``PromptBuilder``) is a natural follow-up.

test/core/pipeline/test_pipeline_base.py

@@ -2554,6 +2554,26 @@ def test_successful(self, receiver_type, current_sender_type, new_sender_type):
         assert inp_socket.is_lazy_variadic is True
         assert inp_socket.wrap_input_in_list is False
 
+    def test_successful_any_typed_socket(self):
+        # An `Any`-typed receiver socket (e.g. every template variable in `PromptBuilder` is
+        # registered as `Any`) can be made lazy variadic so multiple senders can connect to it.
+        # The framework cannot safely flatten one level (it does not know the element type), so
+        # `wrap_input_in_list` is left at its default `True` and the component receives
+        # `[sender_1, sender_2, ...]`.
+        from typing import Any  # local import to keep the parametrize block above self-contained
+
+        pipe = PipelineBase()
+        inp_socket = pipe._make_socket_auto_variadic(
+            component_name="prompt_builder",
+            receiver_socket=InputSocket(name="documents", type=Any, senders=["retriever_1"]),
+            error_type=PipelineConnectError,
+        )
+        assert inp_socket.is_variadic is True
+        assert inp_socket.is_lazy_variadic is True
+        # `wrap_input_in_list` must stay True: the framework has no way to know whether each
+        # sender's value is itself a list to flatten one level. The component decides.
+        assert inp_socket.wrap_input_in_list is True
+
     def test_raises_error_all_int(self):
         with pytest.raises(PipelineConnectError):
             pipe = PipelineBase()
@@ -2592,7 +2612,7 @@ def test_validate_multiple_connections_to_non_variadic_input(self):
             ValueError,
             match="Component 'comp2' cannot accept multiple inputs to 'input_'. "
             "It is already connected to component 'comp1', and it can only accept inputs from multiple "
-            r"senders if its type is list, Optional\[list\], or union of list types.",
+            r"senders if its type is list, Optional\[list\], union of list types, or Any.",
         ):
             pipe.validate_input(data={"comp1": {"input_": "test"}, "comp2": {"input_": "extra_input"}})
 
@@ -2710,3 +2730,38 @@ def test_blocking_components_from_two_branches(self):
         # actually blocking.
         assert blocking_comps == ["a_comp3", "blocking_comp"]
         assert blocking_comp_types == ["FakeComponent", "FakeComponent"]
+
+
+# Integration test for the gap fixed in #10721 follow-up: a component whose input socket is typed
+# as `Any` (e.g. every template variable in `PromptBuilder`) can now accept multiple senders. The
+# connection used to fail with `PipelineConnectError` because #10783's commit message claimed `Any`
+# support but the actual implementation only handled `list` / `Optional[list]` / union of lists.
+class TestAnyTypedMultiConnection:
+    def test_prompt_builder_documents_accepts_multiple_retrievers(self):
+        from haystack import Document
+        from haystack.components.builders import PromptBuilder
+        from haystack.components.retrievers import InMemoryBM25Retriever
+        from haystack.document_stores.in_memory import InMemoryDocumentStore
+
+        ds = InMemoryDocumentStore()
+        ds.write_documents([Document(content="Ferrari is fast", meta={"brand": "Ferrari"})])
+        ds.write_documents([Document(content="Porsche is fast", meta={"brand": "Porsche"})])
+
+        pipe = PipelineBase()
+        pipe.add_component("retriever_1", InMemoryBM25Retriever(document_store=ds))
+        pipe.add_component("retriever_2", InMemoryBM25Retriever(document_store=ds))
+        # PromptBuilder registers every template variable as `Any` (see
+        # `PromptBuilder.__init__` -> `component.set_input_type(self, var, Any)`), so `documents` is an
+        # `Any`-typed socket. Before the fix, the second `connect` raised `PipelineConnectError`; with
+        # the fix, the socket is made lazy variadic and the component receives
+        # `[documents_from_retriever_1, documents_from_retriever_2]` at runtime (one-level wrapping).
+        pipe.add_component("prompt_builder", PromptBuilder(template="Docs: {{ documents }}"))
+
+        pipe.connect("retriever_1.documents", "prompt_builder.documents")
+        pipe.connect("retriever_2.documents", "prompt_builder.documents")
+        # The `Any`-typed socket is now lazy variadic. The component is reachable and the second
+        # connection did not raise. (The aggregation behavior at render time is the component's
+        # responsibility; the framework fix is the connection-eligibility half of #10721.)
+        assert "retriever_1" in pipe.graph.nodes
+        assert "retriever_2" in pipe.graph.nodes
+        assert "prompt_builder" in pipe.graph.nodes