fix: require messages input parameter in Agent component (#10734)

* fix: require messages in Agent component

* fix: default to empty messages list in LLM component

* docs: add release notes

* test: add case to prove agentic pipeline execution failure

* docs: update agent docstring to accurately represent requirement of
messages input

* tests: fix agent messages param

* docs: add explicit bug reference in new test case

Author: marc-mrt

haystack/components/agents/agent.py

@@ -212,6 +212,7 @@ def translate(
 
     # The template variables 'language' and 'document' become inputs to the run method
     result = agent.run(
+        messages=[],
         language="French",
         document="The weather is lovely today and the sun is shining.",
     )
@@ -467,7 +468,7 @@ def _create_agent_span(self) -> Any:
 
     def _initialize_fresh_execution(
         self,
-        messages: list[ChatMessage] | None,
+        messages: list[ChatMessage],
         streaming_callback: StreamingCallbackT | None,
         requires_async: bool,
         *,
@@ -497,12 +498,6 @@ def _initialize_fresh_execution(
         """
         user_prompt = user_prompt or self.user_prompt
         system_prompt = system_prompt or self.system_prompt
-        if messages is None and user_prompt is None and system_prompt is None:
-            raise ValueError(
-                "No messages provided to the Agent and neither user_prompt nor system_prompt is set. "
-                "Please provide at least one of these inputs."
-            )
-
         messages = messages or []
 
         if user_prompt is not None:
@@ -675,7 +670,7 @@ def _runtime_checks(self, break_point: AgentBreakpoint | None) -> None:
 
     def run(  # noqa: PLR0915
         self,
-        messages: list[ChatMessage] | None = None,
+        messages: list[ChatMessage],
         streaming_callback: StreamingCallbackT | None = None,
         *,
         generation_kwargs: dict[str, Any] | None = None,
@@ -905,7 +900,7 @@ def run(  # noqa: PLR0915
 
     async def run_async(  # noqa: PLR0915
         self,
-        messages: list[ChatMessage] | None = None,
+        messages: list[ChatMessage],
         streaming_callback: StreamingCallbackT | None = None,
         *,
         generation_kwargs: dict[str, Any] | None = None,

haystack/components/generators/chat/llm.py

@@ -133,7 +133,7 @@ def run(
             - "last_message": The last message exchanged during the LLM's run.
         """
         return super(LLM, self).run(  # noqa: UP008
-            messages=messages,
+            messages=messages or [],
             streaming_callback=streaming_callback,
             generation_kwargs=generation_kwargs,
             system_prompt=system_prompt,
@@ -170,7 +170,7 @@ async def run_async(
             - "last_message": The last message exchanged during the LLM's run.
         """
         return await super(LLM, self).run_async(  # noqa: UP008
-            messages=messages,
+            messages=messages or [],
             streaming_callback=streaming_callback,
             generation_kwargs=generation_kwargs,
             system_prompt=system_prompt,

releasenotes/notes/revert-optional-agent-messages-8bfcb31386c43fb4.yaml

@@ -0,0 +1,4 @@
+fixes:
+  - |
+    Reverts the change that made Agent messages optional as it caused issues with pipeline execution.
+    As a consequence, the LLM component now defaults to an empty messages list unless provided at runtime.

test/components/agents/test_agent.py

@@ -1380,13 +1380,6 @@ def _make_agent_with_user_prompt(
 
 
 class TestUserPromptInitialization:
-    def test_user_prompt_raises_when_no_messages_and_no_prompt(self, weather_tool):
-        agent = Agent(chat_generator=MockChatGenerator(), tools=[weather_tool])
-        with pytest.raises(
-            ValueError, match="No messages provided to the Agent and neither user_prompt nor system_prompt is set"
-        ):
-            agent.run()
-
     def test_user_prompt_conflict_with_state_schema_raises(self, weather_tool):
         with pytest.raises(ValueError, match="already defined in the state schema"):
             _make_agent_with_user_prompt(
@@ -1400,14 +1393,14 @@ def test_user_prompt_conflict_with_run_param_raises(self, weather_tool):
     def test_user_prompt_only_variables_forwarded_to_builder(self, weather_tool):
         agent = _make_agent_with_user_prompt(_user_msg("Question: {{question}}"), tools=[weather_tool])
         # 'irrelevant_kwarg' is not a template variable — must not raise
-        result = agent.run(question="Will it snow?", irrelevant_kwarg="unused")
+        result = agent.run(messages=[], question="Will it snow?", irrelevant_kwarg="unused")
         assert "messages" in result
 
 
 class TestUserPromptOnly:
     def test_simple_literal_user_prompt(self, weather_tool):
         agent = _make_agent_with_user_prompt(_user_msg("Tell me the weather."), tools=[weather_tool])
-        result = agent.run()
+        result = agent.run(messages=[])
         messages = result["messages"]
         # The rendered user_prompt should be the first (and only) non-system message
         user_messages = [m for m in messages if m.is_from(ChatRole.USER)]
@@ -1423,7 +1416,7 @@ def test_user_prompt_with_template_variables(self, weather_tool):
             ),
             tools=[weather_tool],
         )
-        result = agent.run(name="Alice", cities=["Berlin", "Paris", "Rome"], date="2024-01-15")
+        result = agent.run(messages=[], name="Alice", cities=["Berlin", "Paris", "Rome"], date="2024-01-15")
         user_messages = [m for m in result["messages"] if m.is_from(ChatRole.USER)]
         assert user_messages[0].text == "Hello ALICE, check weather for: Berlin, Paris, Rome on 2024-01-15?"
 
@@ -1438,7 +1431,7 @@ def test_user_prompt_with_system_prompt(self, weather_tool):
             tools=[weather_tool],
             system_prompt="You are a helpful weather assistant.",
         )
-        result = agent.run(city="Berlin")
+        result = agent.run(messages=[], city="Berlin")
         messages = result["messages"]
         assert messages[0].is_from(ChatRole.SYSTEM)
         assert messages[0].text == "You are a helpful weather assistant."
@@ -1455,15 +1448,15 @@ def test_user_prompt_with_documents_variable(self, weather_tool):
             tools=[weather_tool],
         )
         docs = [Document(content="Doc A"), Document(content="Doc B")]
-        result = agent.run(documents=docs, question="What is in the docs?")
+        result = agent.run(messages=[], documents=docs, question="What is in the docs?")
         user_messages = [m for m in result["messages"] if m.is_from(ChatRole.USER)]
         assert "Doc A" in user_messages[0].text
         assert "Doc B" in user_messages[0].text
         assert "What is in the docs?" in user_messages[0].text
 
     def test_runtime_user_prompt_overrides_init_prompt(self, weather_tool):
         agent = _make_agent_with_user_prompt(_user_msg("Default prompt for {{city}}."), tools=[weather_tool])
-        result = agent.run(user_prompt=_user_msg("Runtime prompt for {{city}}."), city="Berlin")
+        result = agent.run(messages=[], user_prompt=_user_msg("Runtime prompt for {{city}}."), city="Berlin")
         user_messages = [m for m in result["messages"] if m.is_from(ChatRole.USER)]
         assert user_messages[0].text == "Runtime prompt for Berlin."
 
@@ -1558,7 +1551,7 @@ def document_store_with_docs(self):
     def test_rag_pipeline_user_prompt_init_only(self, document_store_with_docs, weather_tool):
         pipeline = _make_rag_pipeline(document_store_with_docs, weather_tool)
         query = "Where is the Colosseum?"
-        result = pipeline.run(data={"retriever": {"query": query}, "agent": {"query": query}})
+        result = pipeline.run(data={"retriever": {"query": query}, "agent": {"messages": [], "query": query}})
         assert "agent" in result
         agent_output = result["agent"]
         assert "messages" in agent_output
@@ -1585,6 +1578,7 @@ def test_rag_pipeline_user_prompt_runtime_override(self, document_store_with_doc
             data={
                 "retriever": {"query": query},
                 "agent": {
+                    "messages": [],
                     "user_prompt": _user_msg(
                         "OVERRIDE: Using docs:\n"
                         "{% for doc in documents %}{{doc.content}}\n{% endfor %}"
@@ -1633,3 +1627,151 @@ def test_rag_pipeline_messages_plus_user_prompt(self, document_store_with_docs,
         assert "History:" in user_messages[0].text
         rendered = user_messages[1].text
         assert "Relevant docs:" in rendered
+
+
+class TestAgentPipelineStaticToolInput:
+    """
+    Regression test for the scheduling bug introduced by making the 'messages'
+    run parameter non-required in https://github.com/deepset-ai/haystack/pull/10638.
+
+        pipeline inputs:
+            query    →  history_parser          # feeds the messages chain
+            filters  →  agent.retrieval_filters # static, sender=None  ← the trigger
+            (files is optional / absent)
+
+        pipeline connections:
+            history_parser.messages  →  messages_joiner.values
+            files_processor.prompt   →  messages_joiner.values  # needs 'files' (mandatory)
+            messages_joiner.values   →  system_concat.messages
+            system_concat.output     →  agent.messages
+
+        agent.tools = [ComponentTool(inputs_from_state={"documents": "docs"})]
+
+    The bug
+    -------
+    When the optional 'files' pipeline input is NOT provided:
+    1. files_processor is BLOCKED (its mandatory 'files' input is absent).
+    2. messages_joiner stays DEFER_LAST.
+    3. system_concat is BLOCKED – cannot receive 'messages'.
+    4. agent.messages is therefore never delivered.
+
+    Meanwhile, 'filters' → agent.retrieval_filters (sender=None) fires the pipeline's
+    "user trigger" gate on the Agent's first visit.  Because none of the Agent's
+    sockets are mandatory, can_component_run() returns True and the Agent gets
+    DEFER priority instead of BLOCKED.
+
+    The scheduler eventually pops the Agent (DEFER) from the queue — the only
+    non-BLOCKED component left — and runs it.  _add_missing_input_defaults fills
+    messages=None, and Agent._initialize_fresh_execution raises:
+
+        ValueError("No messages provided to the Agent and neither
+                    user_prompt nor system_prompt is set.")
+    """
+
+    @pytest.fixture()
+    def search_tool(self):
+        return ComponentTool(
+            name="search",
+            description="Searches documents.",
+            component=PromptBuilder(template="{% for d in docs %}{{ d.content }}{% endfor %}"),
+            inputs_from_state={"documents": "docs"},
+        )
+
+    def _make_agent(self, search_tool):
+        chat_generator = MockChatGenerator()
+        agent = Agent(
+            chat_generator=chat_generator,
+            tools=[search_tool],
+            state_schema={"retrieval_filters": {"type": dict[str, Any]}, "documents": {"type": list[Document]}},
+        )
+        # Mock after __init__ so Agent sees the real 'tools' param in the signature.
+        chat_generator.run = MagicMock(return_value={"replies": [ChatMessage.from_assistant("done")]})
+        return agent
+
+    def test_agent_runs_prematurely_when_messages_predecessor_is_blocked(self, search_tool):
+        """
+        Demonstrates the bug: the Agent executes without 'messages' when its
+        messages-providing predecessor chain is permanently BLOCKED.
+
+        Pipeline shape:
+            query   →  history_parser             →  messages_joiner.values
+            files=[]→  files_processor             →  attachments_builder  →  messages_joiner.values
+            messages_joiner  →  system_concat  →  agent.messages
+            filters →  agent.retrieval_filters   (static, triggers the user gate)
+
+        Scheduling sequence that exposes the bug:
+        1. history_parser runs (query provided) → sends to messages_joiner.
+        2. files_processor runs with files=[] → returns {} (_NO_OUTPUT_PRODUCED).
+        3. attachments_builder receives _NO_OUTPUT_PRODUCED → BLOCKED (mandatory
+           processed_files socket never filled).
+        4. messages_joiner is DEFER_LAST (lazy-variadic; attachments_builder
+           has not executed yet → are_all_lazy_variadic_sockets_resolved=False).
+        5. system_concat is BLOCKED (mandatory messages from messages_joiner
+           never received).
+        6. agent is DEFER (static retrieval_filters triggered the user gate;
+           no mandatory sockets → can_component_run=True).
+
+        DEFER (priority=3) < DEFER_LAST (priority=4) → the scheduler picks the
+        Agent before messages_joiner gets a chance to run.  _add_missing_input_defaults
+        fills messages=None, and Agent._initialize_fresh_execution raises:
+            ValueError("No messages provided …")
+        """
+
+        @component
+        class HistoryParser:
+            @component.output_types(messages=list[ChatMessage])
+            def run(self, query: str) -> dict:
+                return {"messages": [ChatMessage.from_user(query)]}
+
+        @component
+        class FilesProcessor:
+            """Produces no output when given an empty file list."""
+
+            @component.output_types(processed_files=list[str])
+            def run(self, files: list[str]) -> dict:
+                if not files:
+                    return {}  # _NO_OUTPUT_PRODUCED → blocks AttachmentsBuilder
+                return {"processed_files": files}
+
+        @component
+        class AttachmentsBuilder:
+            """Builds attachment messages; mandatory processed_files from FilesProcessor."""
+
+            @component.output_types(prompt=list[ChatMessage])
+            def run(self, processed_files: list[str]) -> dict:
+                return {"prompt": [ChatMessage.from_user(f"Files: {processed_files}")]}
+
+        @component
+        class SystemConcat:
+            @component.output_types(output=list[ChatMessage])
+            def run(self, messages: list[ChatMessage]) -> dict:
+                return {"output": messages}
+
+        from haystack.components.joiners.list_joiner import ListJoiner
+
+        agent = self._make_agent(search_tool)
+
+        pipeline = Pipeline()
+        pipeline.add_component("history_parser", HistoryParser())
+        pipeline.add_component("files_processor", FilesProcessor())
+        pipeline.add_component("attachments_builder", AttachmentsBuilder())
+        pipeline.add_component("messages_joiner", ListJoiner(list[ChatMessage]))
+        pipeline.add_component("system_concat", SystemConcat())
+        pipeline.add_component("agent", agent)
+
+        pipeline.connect("history_parser.messages", "messages_joiner.values")
+        pipeline.connect("files_processor.processed_files", "attachments_builder.processed_files")
+        pipeline.connect("attachments_builder.prompt", "messages_joiner.values")
+        pipeline.connect("messages_joiner.values", "system_concat.messages")
+        pipeline.connect("system_concat.output", "agent.messages")
+
+        # files=[] → files_processor produces no output → attachments_builder BLOCKED
+        # → messages_joiner stays DEFER_LAST → system_concat BLOCKED
+        # → agent (DEFER) runs first without messages → ValueError
+        pipeline.run(
+            data={
+                "history_parser": {"query": "What case law applies?"},
+                "files_processor": {"files": []},  # empty → no output
+                "agent": {"retrieval_filters": {"field": "date", "value": "2024-01-01"}},
+            }
+        )