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