Fix is_failed guard by storing exception

Author: andystaples

durabletask/task.py

@@ -552,6 +552,7 @@ def __init__(self, tasks: list[Task[T]]):
         super().__init__(cast(list[Task[Any]], tasks))
         self._completed_tasks = 0
         self._failed_tasks = 0
+        self._pending_exception: TaskFailedError | None = None
 
     @property
     def pending_tasks(self) -> int:
@@ -564,15 +565,21 @@ def on_child_completed(self, task: Task[Any]) -> None:
         self._completed_tasks += 1
         if task.is_failed:
             self._failed_tasks += 1
-            if self._exception is None:
-                self._exception = task.get_exception()
+            if self._pending_exception is None:
+                # Stage the first failure but do NOT expose it via `_exception`
+                # yet. Exposing it now would make `is_failed` return True while
+                # `is_complete` is still False, diverging from .NET's
+                # Task.WhenAll (which does not fault until all children finish).
+                self._pending_exception = task.get_exception()
         if self._completed_tasks == len(self._tasks):
             # Only complete once every child task has completed. This matches the
             # semantics of .NET's Task.WhenAll: the composite task waits for all
             # children to finish and, if any failed, surfaces the first failure
             # rather than failing fast on the first error.
             self._is_complete = True
-            if self._exception is None:
+            if self._pending_exception is not None:
+                self._exception = self._pending_exception
+            else:
                 # The order of the result MUST match the order of the tasks
                 # provided to the constructor.
                 self._result = [child.get_result() for child in self._tasks]

tests/durabletask/test_orchestration_executor.py

@@ -1646,6 +1646,40 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     assert str(ex) in complete_action.failureDetails.errorMessage
 
 
+def test_when_all_defers_failure_until_all_children_complete():
+    """when_all must not report is_failed until every child task completes.
+
+    This mirrors .NET's Task.WhenAll, where the returned task does not fault
+    (failed => complete) until all children have finished. A composite that
+    exposed is_failed while still incomplete would surprise consumers that
+    assume failed implies complete.
+    """
+    t1 = task.CompletableTask()
+    t2 = task.CompletableTask()
+    t3 = task.CompletableTask()
+    when_all = task.when_all([t1, t2, t3])
+
+    assert not when_all.is_complete
+    assert not when_all.is_failed
+
+    # First child fails: the composite must stay pending and must NOT report
+    # failure yet.
+    t1.fail("boom", Exception("boom"))
+    assert not when_all.is_complete
+    assert not when_all.is_failed
+
+    # A later child completes successfully: still pending, still not failed.
+    t2.complete("ok")
+    assert not when_all.is_complete
+    assert not when_all.is_failed
+
+    # Once the final child completes, the composite completes and only then
+    # surfaces the first failure.
+    t3.complete("ok")
+    assert when_all.is_complete
+    assert when_all.is_failed
+
+
 def test_when_any():
     """Tests that a when_any pattern works correctly"""
     def hello(_, name: str):