feat: Validate that no old orchestrators are used inside Workflow graphs

Adds validation to `validate_graph` to raise an error if any of the deprecated
orchestrators (`SequentialAgent`, `ParallelAgent`, `LoopAgent`) are included
as nodes in a `Workflow` static graph. Also includes logic to recursively
unwrap wrapped nodes (e.g. `_ParallelWorker`) to ensure the underlying agent
types are validated.

Co-authored-by: Shangjie Chen <deanchen@google.com>
PiperOrigin-RevId: 947237776

Author: DeanChensj

src/google/adk/workflow/utils/_graph_validation.py

@@ -198,6 +198,43 @@ def _validate_chat_agent_wiring(edges: list[Edge]) -> None:
         )
 
 
+def _validate_no_old_orchestrators(nodes: list[BaseNode]) -> None:
+  """Validates that deprecated orchestrators are not used in the workflow graph."""
+  from ...agents.loop_agent import LoopAgent
+  from ...agents.parallel_agent import ParallelAgent
+  from ...agents.sequential_agent import SequentialAgent
+
+  for node in nodes:
+    target = node
+    visited = set()
+    while True:
+      if id(target) in visited:
+        break
+      visited.add(id(target))
+
+      next_target = None
+      for attr in ("node", "_node", "agent", "_agent"):
+        val = getattr(target, attr, None)
+        if isinstance(val, BaseNode):
+          next_target = val
+          break
+
+      if next_target is not None:
+        target = next_target
+      else:
+        break
+
+    if isinstance(target, (SequentialAgent, ParallelAgent, LoopAgent)):
+      agent_type_name = type(target).__name__
+      raise ValueError(
+          f"Graph validation failed. Node '{node.name}' uses deprecated"
+          f" orchestrator '{agent_type_name}'. Old orchestrators"
+          " (SequentialAgent, ParallelAgent, LoopAgent) cannot be used inside"
+          " a Workflow static graph. Please migrate the orchestration logic to"
+          " Workflow nodes and edges."
+      )
+
+
 def _compute_terminal_nodes(
     nodes: list[BaseNode], edges: list[Edge]
 ) -> set[str]:
@@ -219,4 +256,5 @@ def validate_graph(nodes: list[BaseNode], edges: list[Edge]) -> set[str]:
   _detect_unconditional_cycles(edges, node_names)
   _validate_static_schemas(edges)
   _validate_chat_agent_wiring(edges)
+  _validate_no_old_orchestrators(nodes)
   return _compute_terminal_nodes(nodes, edges)

tests/unittests/workflow/utils/test_graph_validation.py

@@ -16,6 +16,10 @@
 
 import logging
 
+from google.adk.agents.loop_agent import LoopAgent
+from google.adk.agents.parallel_agent import ParallelAgent
+from google.adk.agents.sequential_agent import SequentialAgent
+from google.adk.workflow import BaseNode
 from google.adk.workflow import Edge
 from google.adk.workflow import START
 from google.adk.workflow._graph import DEFAULT_ROUTE
@@ -388,3 +392,71 @@ def test_chat_agent_wiring_validation_only_runs_on_llm_agent() -> None:
   validate_graph(
       graph.nodes, graph.edges
   )  # Should not raise because node_b is a TestingNode, not LlmAgent
+
+
+@pytest.mark.parametrize(
+    'agent_class, name, expected_class_name',
+    [
+        (SequentialAgent, 'SeqAgent', 'SequentialAgent'),
+        (ParallelAgent, 'ParAgent', 'ParallelAgent'),
+        (LoopAgent, 'LoopAgent', 'LoopAgent'),
+    ],
+)
+def test_old_orchestrators_fail_validation(
+    agent_class: type[BaseNode],
+    name: str,
+    expected_class_name: str,
+) -> None:
+  """Tests that a graph containing deprecated orchestrators fails validation."""
+  agent = agent_class(name=name, sub_agents=[])
+  graph = Graph(
+      edges=[
+          Edge(from_node=START, to_node=agent),
+      ],
+  )
+  with pytest.raises(
+      ValueError,
+      match=(
+          rf"Graph validation failed\. Node '{name}' uses deprecated"
+          rf" orchestrator '{expected_class_name}'\."
+      ),
+  ):
+    validate_graph(graph.nodes, graph.edges)
+
+
+def test_old_orchestrator_wrapped_in_parallel_worker_fails_validation() -> None:
+  """Tests that a graph with old orchestrators in ParallelWorker fails validation."""
+  from google.adk.agents.sequential_agent import SequentialAgent
+  from google.adk.workflow._parallel_worker import _ParallelWorker
+
+  seq_agent = SequentialAgent(name='SeqAgent', sub_agents=[])
+  worker_node = _ParallelWorker(node=seq_agent)
+  graph = Graph(
+      edges=[
+          Edge(from_node=START, to_node=worker_node),
+      ],
+  )
+  with pytest.raises(
+      ValueError,
+      match=(
+          r"Graph validation failed\. Node 'SeqAgent' uses deprecated"
+          r" orchestrator 'SequentialAgent'\."
+      ),
+  ):
+    validate_graph(graph.nodes, graph.edges)
+
+
+def test_self_referential_wrapped_node_safely_terminates() -> None:
+  """Tests that a self-referential wrapped node safely breaks without infinite loop."""
+  node_a = TestingNode(name='NodeA')
+  # Mock a self-referential node property
+  object.__setattr__(node_a, 'node', node_a)
+
+  graph = Graph(
+      edges=[
+          Edge(from_node=START, to_node=node_a),
+      ],
+  )
+  validate_graph(
+      graph.nodes, graph.edges
+  )  # Should not spin forever and should pass