Fix when_all hang when some children complete before construction

dapr/ext/workflow/_durabletask/task.py

@@ -385,9 +385,12 @@ 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.
         super().__init__(tasks)
-        self._completed_tasks = 0
-        self._failed_tasks = 0
         # If there are no child tasks, this composite should complete immediately
         if len(self._tasks) == 0:
             self._result = []  # type: ignore[assignment]

tests/ext/workflow/durabletask/test_task.py

@@ -62,6 +62,63 @@ def test_when_all_happy_path_returns_ordered_results_and_completes_last():
     assert all_task.get_result() == ['one', 'two', 'three']
 
 
+def test_when_all_counts_already_complete_children():
+    """Deferred when_all must retain completions that happened before construction.
+
+    Pattern: schedule children, do other work (timer/activity), then when_all.
+    Children that finish during that gap must still be counted; otherwise
+    when_all hangs forever when some but not all children finished early.
+    """
+    children = [task.CompletableTask() for _ in range(5)]
+    for child in children[:3]:
+        child.complete(f'done-{id(child)}')
+
+    all_task = task.when_all(children)
+
+    assert not all_task.is_complete
+    assert all_task.get_completed_tasks() == 3
+
+    children[3].complete('four')
+    assert not all_task.is_complete
+
+    children[4].complete('five')
+    assert all_task.is_complete
+    assert all_task.get_completed_tasks() == 5
+    assert len(all_task.get_result()) == 5
+
+
+def test_when_all_all_children_already_complete():
+    """when_all of already-complete children should complete immediately."""
+    children = [task.CompletableTask() for _ in range(3)]
+    for i, child in enumerate(children):
+        child.complete(f'v{i}')
+
+    all_task = task.when_all(children)
+
+    assert all_task.is_complete
+    assert all_task.get_result() == ['v0', 'v1', 'v2']
+
+
+def test_when_any_of_when_all_with_precompleted_children():
+    """when_any([when_all(children), timeout]) must unblock when children finish early."""
+    children = [task.CompletableTask() for _ in range(3)]
+    children[0].complete('a')
+    children[1].complete('b')
+
+    timeout = task.CompletableTask()
+    all_task = task.when_all(children)
+    any_task = task.when_any([all_task, timeout])
+
+    assert not any_task.is_complete
+    assert all_task.get_completed_tasks() == 2
+
+    children[2].complete('c')
+
+    assert all_task.is_complete
+    assert any_task.is_complete
+    assert any_task.get_result() is all_task
+
+
 def test_when_all_is_composable_with_when_any():
     c1 = task.CompletableTask()
     c2 = task.CompletableTask()