diff --git a/dapr/ext/workflow/_durabletask/task.py b/dapr/ext/workflow/_durabletask/task.py
index 26813c5a..8149013a 100644
--- a/dapr/ext/workflow/_durabletask/task.py
+++ b/dapr/ext/workflow/_durabletask/task.py
@@ -364,10 +364,33 @@ class CompositeTask(Task[T]):
     _tasks: list[Task]
 
     def __init__(self, tasks: list[Task]):
+        """Adopts the child tasks and replays completions for any already-complete child.
+
+        Invokes the subclass's ``on_child_completed`` during construction, so
+        subclasses must set every attribute that override reads before calling
+        ``super().__init__()``, and must not reset counting state after it.
+
+        Raises:
+            ValueError: If the same task instance appears more than once in ``tasks``,
+                or if a still-pending task already belongs to another pending composite.
+        """
         super().__init__()
+        unique_task_count = len({id(task) for task in tasks})
+        if unique_task_count != len(tasks):
+            raise ValueError(
+                'the same task instance was passed to when_all/when_any more than once; '
+                'a task notifies its parent composite only once, so duplicates would hang'
+            )
+        for task in tasks:
+            has_live_parent = task._parent is not None and not task._parent.is_complete
+            if has_live_parent and not task.is_complete:
+                raise ValueError(
+                    'a pending task passed to when_all/when_any already belongs to another '
+                    'pending composite; a task notifies only its latest composite, so the '
+                    'earlier one would hang'
+                )
         self._tasks = tasks
         self._completed_tasks = 0
-        self._failed_tasks = 0
         for task in tasks:
             task._parent = self
             if task.is_complete:
@@ -385,11 +408,8 @@ class WhenAllTask(CompositeTask[list[T]]):
     """A task that completes when all of its child tasks complete."""
 
     def __init__(self, tasks: list[Task[T]]):
-        # Do not reset _completed_tasks / _failed_tasks after super().__init__.
-        # CompositeTask already initializes them and counts any children that are
-        # already complete via on_child_completed(). Resetting the counters here
-        # drops those completions, so deferred when_all(children) hangs forever
-        # when some (but not all) children finished before when_all was constructed.
+        # CompositeTask.__init__ already counted pre-completed children; do not reset
+        # _completed_tasks after it, or a deferred when_all over them hangs forever.
         super().__init__(tasks)
         # If there are no child tasks, this composite should complete immediately
         if len(self._tasks) == 0:
@@ -579,6 +599,7 @@ def __init__(
         timeout: Optional[Union[datetime, timedelta]] = None,
         on_timeout: Optional[Callable[[], None]] = None,
     ):
+        # Set before super().__init__(), which may replay completions into on_child_completed.
         self._event_task = event_task
         self._timer_task = timer_task
         self._event_name = event_name
diff --git a/tests/ext/workflow/durabletask/test_task.py b/tests/ext/workflow/durabletask/test_task.py
index 230d9b54..49e3cdfb 100644
--- a/tests/ext/workflow/durabletask/test_task.py
+++ b/tests/ext/workflow/durabletask/test_task.py
@@ -119,6 +119,65 @@ def test_when_any_of_when_all_with_precompleted_children():
     assert any_task.get_result() is all_task
 
 
+def test_when_all_rejects_duplicate_task_instances():
+    """A child notifies its parent composite only once, so a duplicated pending
+    child would leave when_all one notification short and hanging forever.
+    Composites reject duplicates upfront instead."""
+    child = task.CompletableTask()
+
+    with pytest.raises(ValueError, match='more than once'):
+        task.when_all([child, child])
+
+
+def test_when_all_rejects_duplicate_already_complete_task_instances():
+    """Duplicates are rejected regardless of completion state, so behavior does
+    not depend on whether the child finished before or after construction."""
+    child = task.CompletableTask()
+    child.complete('x')
+
+    with pytest.raises(ValueError, match='more than once'):
+        task.when_all([child, child])
+
+
+def test_when_any_rejects_duplicate_task_instances():
+    child = task.CompletableTask()
+
+    with pytest.raises(ValueError, match='more than once'):
+        task.when_any([child, child])
+
+
+def test_composite_rejects_pending_child_owned_by_live_composite():
+    """A task has a single parent slot, so two live composites over the same
+    pending child would leave the earlier one unnotified and hanging forever.
+    Constructing the second composite fails fast instead."""
+    t1 = task.CompletableTask()
+    t2 = task.CompletableTask()
+    t3 = task.CompletableTask()
+    task.when_all([t1, t2])
+
+    with pytest.raises(ValueError, match='already belongs'):
+        task.when_any([t1, t3])
+
+
+def test_sequential_composite_reuse_after_completion_is_allowed():
+    """Winner-then-gather must keep working: reusing a losing (still pending)
+    child is fine once the first composite completed, because completed
+    composites ignore late notifications. Only live overlap is rejected."""
+    t1 = task.CompletableTask()
+    t2 = task.CompletableTask()
+    any_task = task.when_any([t1, t2])
+    t1.complete('winner')
+    assert any_task.is_complete
+
+    all_task = task.when_all([t1, t2])
+    assert all_task.get_completed_tasks() == 1
+    assert not all_task.is_complete
+
+    t2.complete('loser')
+    assert all_task.is_complete
+    assert all_task.get_result() == ['winner', 'loser']
+
+
 def test_when_all_is_composable_with_when_any():
     c1 = task.CompletableTask()
     c2 = task.CompletableTask()