fix(workflow): Prevent incorrect chat agent wiring in graphs

Add validation to raise a ValueError at graph compilation time when a chat-mode agent has an incoming edge from a non-START node. This prevents silent dropping of node inputs.

Fixes #5868

Change-Id: I575a7594a8912b4ea6eb4c1ed539c6c7f16eadd1

Author: boyangsvl

src/google/adk/workflow/_graph.py

@@ -517,6 +517,26 @@ def _validate_static_schemas(self) -> None:
               f" input schema {to_node.input_schema}."
           )
 
+  def _validate_chat_agent_wiring(self) -> None:
+    """Validates that chat-mode agents do not have incoming edges from non-START nodes."""
+    from ..agents.llm_agent import LlmAgent
+
+    for edge in self.edges:
+      to_node = edge.to_node
+      if (
+          isinstance(to_node, LlmAgent)
+          and getattr(to_node, "mode", None) == "chat"
+      ):
+        if edge.from_node.name != START.name:
+          raise ValueError(
+              f"The agent '{to_node.name}' has been added to the workflow with"
+              f" mode='chat' following node '{edge.from_node.name}'. This is"
+              " not supported because chat-mode agents rely on conversational"
+              " history (session events) and cannot consume direct node inputs"
+              " from preceding nodes. Please change the agent's mode to"
+              " 'single_turn'"
+          )
+
   def _compute_terminal_nodes(self) -> None:
     """Computes terminal nodes (no outgoing edges)."""
     from_names = {edge.from_node.name for edge in self.edges}
@@ -535,4 +555,5 @@ def validate_graph(self) -> None:
     self._validate_default_routes()
     self._detect_unconditional_cycles(node_names)
     self._validate_static_schemas()
+    self._validate_chat_agent_wiring()
     self._compute_terminal_nodes()

tests/unittests/workflow/test_graph.py

@@ -775,3 +775,19 @@ def test_get_next_pending_nodes() -> None:
       'NodeA', routes_to_match=['route1', 'unknown_route']
   )
   assert set(next_nodes) == {'NodeB', 'NodeC'}
+
+
+def test_chat_agent_wiring_validation_only_runs_on_llm_agent() -> None:
+  """Tests that _validate_chat_agent_wiring checks non-LlmAgent nodes safely."""
+  node_a = TestingNode(name='NodeA')
+  node_b = TestingNode(name='NodeB')
+  # Set mode='chat' on a non-LlmAgent node
+  object.__setattr__(node_b, 'mode', 'chat')
+
+  graph = Graph(
+      edges=[
+          Edge(from_node=START, to_node=node_a),
+          Edge(from_node=node_a, to_node=node_b),
+      ],
+  )
+  graph.validate_graph()  # Should not raise because node_b is a TestingNode, not LlmAgent

tests/unittests/workflow/test_llm_agent_as_node.py

@@ -1093,6 +1093,35 @@ async def mock_run_async(*args, **kwargs):
   assert events[0].output is None
 
 
+def test_chat_mode_agent_following_non_start_raises_validation_error():
+  """Wiring a chat-mode agent following a non-START node raises ValueError."""
+  agent = _make_v1_agent(mode='chat')
+  predecessor = TestingNode(name='pred', output='some output')
+
+  with pytest.raises(ValueError) as exc_info:
+    Workflow(
+        name='wf',
+        edges=[('START', predecessor), (predecessor, agent)],
+    )
+
+  assert (
+      "The agent 'test_v1_agent' has been added to the workflow with"
+      " mode='chat' following node 'pred'."
+      in str(exc_info.value)
+  )
+
+
+def test_chat_mode_agent_from_start_allowed():
+  """Wiring a chat-mode agent directly from START is allowed and validated without error."""
+  agent = _make_v1_agent(mode='chat')
+
+  wf = Workflow(
+      name='wf',
+      edges=[('START', agent)],
+  )
+  assert wf.graph is not None
+
+
 @pytest.mark.asyncio
 async def test_three_layer_llm_agent_transfer_round_trip(
     request: pytest.FixtureRequest,