From 76b1681c6dc39349419ca75f965054b049224099 Mon Sep 17 00:00:00 2001 From: Goutham Annem Date: Mon, 15 Jun 2026 11:08:49 -0700 Subject: [PATCH] fix(workflow): exclude done tasks from get_dynamic_tasks After a ParallelWorker completes, its tasks remain in self.runs with done=True. get_dynamic_tasks() returned them all, causing _cleanup_all_tasks to log a spurious "cancelling N leftover tasks" warning on every clean run. Filter to only tasks that are still running. Closes #6082 --- .../adk/workflow/_dynamic_node_scheduler.py | 6 +++- .../workflow/test_dynamic_node_scheduler.py | 35 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/src/google/adk/workflow/_dynamic_node_scheduler.py b/src/google/adk/workflow/_dynamic_node_scheduler.py index 70afd67c25f..110117c015f 100644 --- a/src/google/adk/workflow/_dynamic_node_scheduler.py +++ b/src/google/adk/workflow/_dynamic_node_scheduler.py @@ -99,7 +99,11 @@ class DynamicNodeState: def get_dynamic_tasks(self) -> list[asyncio.Task[Context]]: """Get all active dynamic node tasks.""" - return [run.task for run in self.runs.values() if run.task] + return [ + run.task + for run in self.runs.values() + if run.task and not run.task.done() + ] class DynamicNodeScheduler(ScheduleDynamicNode): diff --git a/tests/unittests/workflow/test_dynamic_node_scheduler.py b/tests/unittests/workflow/test_dynamic_node_scheduler.py index 0e7e37f2b3c..82fb343fed2 100644 --- a/tests/unittests/workflow/test_dynamic_node_scheduler.py +++ b/tests/unittests/workflow/test_dynamic_node_scheduler.py @@ -566,6 +566,41 @@ async def _run_impl(self, *, ctx, node_input): ) +def test_get_dynamic_tasks_excludes_done_tasks(): + """get_dynamic_tasks should not return completed tasks (regression for #6082).""" + import asyncio + + loop = asyncio.new_event_loop() + try: + async def _done(): + return None + + done_task = loop.run_until_complete(asyncio.ensure_future(_done(), loop=loop)) + running_coro = asyncio.sleep(9999) + running_task = loop.create_task(running_coro) + + state = DynamicNodeState() + state.runs['path/done@r-1'] = DynamicNodeRun( + state=NodeState(run_id='r-1'), + task=done_task, + ) + state.runs['path/running@r-2'] = DynamicNodeRun( + state=NodeState(run_id='r-2'), + task=running_task, + ) + state.runs['path/no-task@r-3'] = DynamicNodeRun( + state=NodeState(run_id='r-3'), + task=None, + ) + + tasks = state.get_dynamic_tasks() + + assert tasks == [running_task] + running_task.cancel() + finally: + loop.close() + + class _ModelA(BaseModel): x: int