@@ -13,6 +13,26 @@ ADDED
 
 FIXED
 
+- Fixed `when_all` hanging when pending tasks are shared between groups or
+appear multiple times in one group. Results preserve input order and duplicate
+positions without scheduling additional activities. Completed `when_any` groups
+stop observing losing children without disturbing other groups, and a throwing
+completion callback no longer prevents other observers from being notified.
+
+> [!WARNING]
+> **Replay-breaking bug fix:** Shared nested races can select a different direct
+> child during replay. For example, with `first = when_any([timer, work])` and
+> `second = when_any([first, timer])`, timer completion now notifies `first`,
+> allowing `second` to select `first` rather than `timer`. Existing downstream
+> history can then fail replay with `NonDeterminismError`. Disjoint task groups
+> are unchanged.
+>
+> Drain affected instances on the previous SDK, or keep their task hub on the
+> previous SDK and use a separate task hub for new instances. Already-stuck
+> instances may need deliberate recovery. Pin or lock the core `durabletask`
+> dependency, including when using a provider. Provider dependency minimums
+> will be coordinated in a dedicated release PR after the core fix is released.
+
 - Single-instance `purge_orchestration()` requests now explicitly target
 orchestrations rather than entities in both synchronous and asynchronous clients.
 - Timer callbacks no longer schedule additional long-timer chunks, retry
 
@@ -13,6 +13,11 @@ ADDED
 
 FIXED
 
+- With the corresponding core `durabletask` SDK fix, shared or duplicate pending
+tasks no longer strand native `when_all` or compatibility `context.task_all()`
+groups. This correction can change shared nested `when_any` / `context.task_any()`
+race winners during replay; see the [core changelog](../CHANGELOG.md) for upgrade
+guidance. Dependency minimums will be coordinated after the core fix is released.
 - With the corresponding core `durabletask` SDK fix, single-instance
 `purge_orchestration()` requests from `DurableFunctionsClient` and
 `SyncDurableFunctionsClient` now explicitly target orchestrations rather than
 
@@ -51,6 +51,31 @@ def orchestrator(ctx: task.OrchestrationContext, _):
 
 See the full [fan-out sample](../examples/fanout_fanin.py).
 
+Tasks can be shared between `when_all` and `when_any` groups without scheduling
+the activity again. `when_all` preserves input order and every duplicate position:
+`when_all([work, work])` returns the result twice, whether `work` is pending or
+already complete. If a child fails, `when_all` waits for every child before raising
+the first observed failure.
+
+`when_any` returns its winning direct child, including when that child is itself
+a composite. It stops observing losing children but does not cancel them or
+remove other groups' subscriptions. Completion notifications follow group
+subscription order, so shared nested races can select a composite rather than
+the leaf task that triggered it.
+
+> [!WARNING]
+> Corrected shared-task notifications can change race winners during replay of
+> existing instances. Drain affected instances on the previous SDK or isolate
+> new instances in a separate task hub before upgrading. See the
+> [core changelog](../CHANGELOG.md) for migration guidance.
+
+A canceled task still raises `TaskCancelledError` when its result is read.
+If collecting `when_all` results throws, that group does not notify its own
+parents; do not catch the error and reuse the group. Other groups observing
+the same child are still notified. A single completion callback error is
+re-raised unchanged; multiple callback errors are collected in one exception
+whose `errors` tuple retains each original exception.
+
 ## Human interaction and durable timers
 
 An orchestration can wait for a user-defined event, such as a human approval event, before proceeding
 
@@ -13,6 +13,11 @@ ADDED
 
 FIXED
 
+- With the corresponding core `durabletask` SDK fix, shared or duplicate pending
+tasks no longer leave `when_all` groups waiting after their children complete.
+This correction can change shared nested race winners during replay; see the
+[core changelog](../CHANGELOG.md) for upgrade guidance. Dependency minimums will
+be coordinated after the core fix is released.
 - With the corresponding core `durabletask` SDK fix, single-instance
 `purge_orchestration()` requests from synchronous and asynchronous Azure Managed
 clients now explicitly target orchestrations rather than entities.
 
@@ -608,17 +608,28 @@ class TaskCancelledError(Exception):
     """Exception type for cancelled orchestration tasks."""
 
 
+class _TaskCompletionError(Exception):
+    """Collect callback errors without requiring Python 3.11 ExceptionGroup."""
+
+    def __init__(self, errors: list[Exception]) -> None:
+        self.errors = tuple(errors)
+        super().__init__(
+            "Multiple task completion callbacks failed: "
+            + "; ".join(f"{type(error).__name__}: {error}" for error in errors))
+
+
 class Task(ABC, Generic[T]):
     """Abstract base class for asynchronous tasks in a durable orchestration."""
     _result: T
     _exception: TaskFailedError | None
-    _parent: CompositeTask[Any] | None
+    _parents: dict[int, CompositeTask[Any]]
 
     def __init__(self) -> None:
         super().__init__()
         self._is_complete = False
         self._exception = None
-        self._parent = None
+        # Dict insertion order preserves subscription order for shared races.
+        self._parents = {}
 
     @property
     def is_complete(self) -> bool:
@@ -650,8 +661,19 @@ def get_exception(self) -> TaskFailedError:
         return self._exception
 
     def _notify_parent(self) -> None:
-        if self._parent is not None:
-            self._parent.on_child_completed(self)
+        parents = tuple(self._parents.values())
+        self._parents.clear()
+        errors: list[Exception] = []
+        for parent in parents:
+            try:
+                parent.on_child_completed(self)
+            except Exception as error:
+                # A canceled result can throw without blocking other observers.
+                errors.append(error)
+        if len(errors) == 1:
+            raise errors[0]
+        if errors:
+            raise _TaskCompletionError(errors) from errors[0]
 
 
 class CompositeTask(Task[T]):
@@ -665,10 +687,24 @@ def __init__(self, tasks: list[Task[Any]]):
         self._tasks = tasks
         self._completed_tasks = 0
         self._failed_tasks = 0
+        # Track references by identity, including unhashable Task subclasses.
+        self._pending_children: dict[int, tuple[Task[Any], int]] = {}
         for task in tasks:
-            task._parent = self
+            _, count = self._pending_children.get(id(task), (task, 0))
+            self._pending_children[id(task)] = (task, count + 1)
+        for task, _ in tuple(self._pending_children.values()):
+            if self.is_complete:
+                break
             if task.is_complete:
                 self.on_child_completed(task)
+            else:
+                task._parents[id(self)] = self
+
+    def _notify_parent(self) -> None:
+        for task, _ in self._pending_children.values():
+            task._parents.pop(id(self), None)
+        self._pending_children.clear()
+        super()._notify_parent()
 
     def get_tasks(self) -> list[Task[Any]]:
         return self._tasks
@@ -698,10 +734,14 @@ def pending_tasks(self) -> int:
 
     def on_child_completed(self, task: Task[Any]) -> None:
         if self.is_complete:
-            raise ValueError('The task has already completed.')
-        self._completed_tasks += 1
+            return
+        pending = self._pending_children.pop(id(task), None)
+        if pending is None:
+            return
+        _, occurrences = pending
+        self._completed_tasks += occurrences
         if task.is_failed:
-            self._failed_tasks += 1
+            self._failed_tasks += occurrences
             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
@@ -871,12 +911,20 @@ def on_child_completed(self, task: Task[Any]) -> None:
 
 
 def when_all(tasks: list[Task[T]]) -> WhenAllTask[T]:
-    """Returns a task that completes when all of the provided tasks complete or when one of the tasks fail."""
+    """Wait for every provided task to complete or fail.
+
+    Results preserve input order and duplicate task references. Tasks may be
+    shared between groups without scheduling additional work. If any task fails,
+    the first observed failure is raised after every task has finished.
+    """
     return WhenAllTask(tasks)
 
 
 def when_any(tasks: Sequence[Task[T]]) -> WhenAnyTask[T]:
-    """Returns a task that completes when any of the provided tasks complete or fail."""
+    """Return the first direct child to complete or fail.
+
+    Tasks may be shared between groups. Losing tasks are not canceled.
+    """
     return WhenAnyTask(list(tasks))
 
 
 
@@ -6,6 +6,7 @@
 
 import pytest
 
+from durabletask import task
 from azure.durable_functions.internal.compat.orchestration_context import (
     DurableOrchestrationContext,
     accepts_two_positional_args,
@@ -29,6 +30,25 @@ def test_get_input_returns_stored_input():
     assert adapter.get_input() == {"x": 1}
 
 
+def test_task_helpers_support_shared_and_duplicate_pending_tasks():
+    adapter, _ = _adapter()
+    a = task.CompletableTask[str]()
+    b = task.CompletableTask[str]()
+    c = task.CompletableTask[str]()
+    left = adapter.task_all([a, b, a])
+    right = adapter.task_all([a, c])
+    race = adapter.task_any([a, c])
+    root = adapter.task_all([left, right])
+
+    a.complete("A")
+    b.complete("B")
+    c.complete("C")
+
+    assert root.is_complete
+    assert root.get_result() == [["A", "B", "A"], ["A", "C"]]
+    assert race.get_result() is a
+
+
 def test_property_delegation():
     adapter, fake = _adapter()
     assert adapter.instance_id == "iid"
 
@@ -88,6 +88,59 @@ def sequence(ctx: task.OrchestrationContext, start_val: int):
     assert state.serialized_custom_status is None
 
 
+@pytest.mark.parametrize("pattern, expected, invocations", [
+    ("shared", [["A", "B"], ["A", "C"]], ["A", "B", "C"]),
+    ("duplicate", ["V", "V"], ["V"]),
+    ("precompleted_duplicate", ["V", "V"], ["V"]),
+    ("distinct", ["A", "B", "C"], ["A", "B", "C"]),
+    ("nested_disjoint", [["A", "B"], ["C"]], ["A", "B", "C"]),
+])
+def test_task_composition_executes_each_activity_once(pattern, expected, invocations):
+    calls: list[str] = []
+    lock = threading.Lock()
+
+    def echo(_: task.ActivityContext, value: str) -> str:
+        with lock:
+            calls.append(value)
+        return value
+
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        if pattern in ("duplicate", "precompleted_duplicate"):
+            work = ctx.call_activity(echo, input="V")
+            if pattern == "precompleted_duplicate":
+                yield work
+            return (yield task.when_all([work, work]))
+
+        a = ctx.call_activity(echo, input="A")
+        b = ctx.call_activity(echo, input="B")
+        c = ctx.call_activity(echo, input="C")
+        if pattern == "shared":
+            left = task.when_all([a, b])
+            right = task.when_all([a, c])
+            return (yield task.when_all([left, right]))
+        if pattern == "nested_disjoint":
+            return (yield task.when_all([task.when_all([a, b]), task.when_all([c])]))
+        return (yield task.when_all([a, b, c]))
+
+    with worker.TaskHubGrpcWorker(host_address=HOST) as w:
+        w.add_orchestrator(orchestrator)
+        w.add_activity(echo)
+        w.start()
+        with client.TaskHubGrpcClient(host_address=HOST) as c:
+            instance_id = c.schedule_new_orchestration(orchestrator)
+            state = c.wait_for_orchestration_completion(instance_id, timeout=10)
+            committed_history = c.get_orchestration_history(instance_id)
+
+    assert state is not None
+    assert state.runtime_status == client.OrchestrationStatus.COMPLETED
+    assert state.serialized_output == json.dumps(expected)
+    assert sorted(calls) == invocations
+    assert sum(isinstance(event, history.TaskScheduledEvent)
+               for event in committed_history) == len(invocations)
+    assert sum(isinstance(event, history.TaskCompletedEvent)
+               for event in committed_history) == len(invocations)
+
+
 def test_activity_error_handling():
 
     def throw(_: task.ActivityContext, input: int) -> int:
 
@@ -3117,7 +3117,8 @@ def orchestrator(ctx: task.OrchestrationContext, _):
     assert complete_action.result.value == encoded_output
 
 
-def test_when_all_with_retry():
+@pytest.mark.parametrize("shared", [False, True])
+def test_when_all_with_retry(shared):
     """Tests that a when_all pattern works correctly with retries"""
     def dummy_activity(ctx, inp: str):
         if inp == "Tokyo":
@@ -3134,7 +3135,12 @@ def orchestrator(ctx: task.OrchestrationContext, _):
                                    retry_timeout=timedelta(seconds=50)),
                                input="Tokyo")
         t2 = ctx.call_activity(dummy_activity, input="Seattle")
-        results = yield task.when_all([t1, t2])
+        if shared:
+            left = task.when_all([t1, t2, t1])
+            right = task.when_all([t1, t2])
+            results = yield task.when_all([left, right])
+        else:
+            results = yield task.when_all([t1, t2])
         return results
 
     registry = worker._Registry()