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.

PiperOrigin-RevId: 947349988

Author: google-genai-bot

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

@@ -198,43 +198,6 @@ 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]:
@@ -256,5 +219,4 @@ 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,10 +16,6 @@
 
 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
@@ -392,71 +388,3 @@ 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