fix(workflow): Prevent silent drain of routed nodes in wait_for_output

A route-only node parked WAITING because completion handlers only checked
output is None. Added a check for route is None to satisfy the BaseNode
docstring contract that transitions to COMPLETED upon yielding output or route.
Includes corresponding unit test.

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

Author: DeanChensj

src/google/adk/workflow/_dynamic_node_scheduler.py

@@ -512,7 +512,11 @@ def _record_result(
     elif child_ctx.actions.transfer_to_agent:
       state.status = NodeStatus.COMPLETED
       run.transfer_to_agent = child_ctx.actions.transfer_to_agent
-    elif node.wait_for_output and child_ctx.output is None:
+    elif (
+        node.wait_for_output
+        and child_ctx.output is None
+        and child_ctx.route is None
+    ):
       state.status = NodeStatus.WAITING
     else:
       state.status = NodeStatus.COMPLETED

src/google/adk/workflow/_workflow.py

@@ -653,7 +653,11 @@ def _handle_completion(
       loop_state.interrupt_ids.update(child_ctx.interrupt_ids)
       return
 
-    if node.wait_for_output and child_ctx.output is None:
+    if (
+        node.wait_for_output
+        and child_ctx.output is None
+        and child_ctx.route is None
+    ):
       node_state.status = NodeStatus.WAITING
       return
 

tests/unittests/workflow/test_workflow_routes.py

@@ -15,7 +15,6 @@
 """Testings for the Workflow routes."""
 
 from typing import Any
-from typing import Dict
 
 from google.adk.agents.context import Context
 from google.adk.apps.app import App
@@ -28,7 +27,6 @@
 import pytest
 
 from .. import testing_utils
-from .workflow_testing_utils import create_parent_invocation_context
 from .workflow_testing_utils import simplify_events_with_node
 from .workflow_testing_utils import TestingNode
 
@@ -37,7 +35,7 @@
 async def test_run_async_with_edge_routes(request: pytest.FixtureRequest):
   route_holder = {'route': 'route_b'}
 
-  def dynamic_router(ctx: Context, node_input: Any):
+  def dynamic_router(_ctx: Context, _node_input: Any):
     return route_holder['route']
 
   node_a = TestingNode(name='NodeA', output='A', route=dynamic_router)
@@ -140,6 +138,39 @@ async def test_output_route_bool(request: pytest.FixtureRequest):
   ]
 
 
+@pytest.mark.asyncio
+async def test_wait_for_output_with_route_only_completes_successfully(
+    request: pytest.FixtureRequest,
+):
+  """A node with wait_for_output=True that yields only a route should complete and continue the workflow."""
+  node_a = TestingNode(name='NodeA', route='go_next', wait_for_output=True)
+  node_b = TestingNode(name='NodeB', output='B_done')
+
+  agent = Workflow(
+      name='test_wait_for_output_route',
+      edges=[
+          (START, node_a),
+          (node_a, {'go_next': node_b}),
+      ],
+  )
+  app = App(name=request.function.__name__, root_agent=agent)
+  runner = testing_utils.InMemoryRunner(app=app)
+
+  # NodeA should yield no output, and NodeB should yield its output.
+  events = await runner.run_async(testing_utils.get_user_content('start'))
+
+  assert simplify_events_with_node(events) == [
+      (
+          'test_wait_for_output_route@1/NodeA@1',
+          {'output': None},
+      ),
+      (
+          'test_wait_for_output_route@1/NodeB@1',
+          {'output': 'B_done'},
+      ),
+  ]
+
+
 @pytest.mark.asyncio
 async def test_output_route_no_data(request: pytest.FixtureRequest):
   node_a = TestingNode(name='NodeA', route='route_b')