Prototype async/await orchestrators across all three Python packages

CHANGELOG.md

@@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- Prototype support for `async def` orchestrators that `await` durable tasks,
+including activities, timers, external events, and `when_all` / `when_any`.
+Generator orchestrators remain supported. This is replay-driven syntax, not
+support for asyncio I/O inside orchestrators.
+
 ## v1.12.0
 
 ADDED

README.md

@@ -27,6 +27,44 @@ and [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/
 
 ## Optional Features
 
+### Async Orchestrators (Prototype)
+
+This branch prototypes `async def` orchestrators across all three packages.
+Existing generator orchestrators continue to work.
+
+```python
+from durabletask import task
+
+
+async def sequence(ctx: task.OrchestrationContext, value: int) -> list[int]:
+    first = await ctx.call_activity("plus_one", input=value, return_type=int)
+    pending: list[task.Task[int]] = [
+        ctx.call_activity("plus_one", input=first, return_type=int),
+        ctx.call_activity("plus_one", input=value, return_type=int),
+    ]
+    return await task.when_all(pending)
+```
+
+Register the function with `TaskHubGrpcWorker`, `DurableTaskSchedulerWorker`,
+or the Azure Functions `orchestration_trigger` decorator as usual. Functions
+also supports single-argument `async def` orchestrators using the compatibility
+context and `await context.task_all(...)`.
+
+> [!WARNING]
+> This is replay-driven await syntax, not general-purpose asyncio execution.
+> Await only durable tasks or async helpers that themselves await durable tasks.
+> Use durable timers instead of `asyncio.sleep`, and `task.when_all` /
+> `task.when_any` instead of `asyncio.gather`, `create_task`, or `TaskGroup`.
+> Perform network, storage, and other nondeterministic I/O in activities.
+> Completed foreign awaitables and synchronous side effects cannot reliably be
+> detected by the replay engine.
+
+An async helper is not a durable sub-orchestration: use
+`await ctx.call_sub_orchestrator(...)` for a separate orchestration instance.
+An `async def` function containing `yield` is an async generator, not a supported
+orchestrator. Converting an existing orchestrator must preserve durable action
+order and payloads for replay compatibility.
+
 ### Large Payload Externalization
 
 Install the `azure-blob-payloads` extra to automatically offload oversized orchestration payloads to

azure-functions-durable/CHANGELOG.md

@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+ADDED
+
+- With the corresponding core SDK prototype, native and v1-style orchestrators
+can use `async def` and await durable tasks. The
+`orchestrator_generator_wrapper` test helper also accepts coroutines.
+Requires a coordinated release after the core SDK is published; dependency
+minimums are intentionally unchanged in this spike.
+
 ## v2.0.0rc3
 
 ADDED

azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py

@@ -1,16 +1,17 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-"""Replay simulation helper for unit-testing orchestrator generators."""
+"""Replay simulation helper for unit-testing generator or coroutine orchestrators."""
 
-from typing import Any, Generator
+from collections.abc import Coroutine, Generator
+from typing import Any
 
 
 def orchestrator_generator_wrapper(
-        generator: Generator[Any, Any, Any]) -> Generator[Any, None, None]:
-    """Drive a user orchestrator generator the way the Durable replay engine does.
+        generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any]) -> Generator[Any, None, None]:
+    """Drive a user orchestrator the way the Durable replay engine does.
 
-    A Durable Functions orchestrator is a generator that yields task objects and
+    A Durable Functions orchestrator yields or awaits durable task objects and
     is resumed with each task's result until it returns the orchestration
     output. In a real execution the Durable Task runtime performs this drive
     loop during replay. This wrapper reproduces that loop in-process so an
@@ -22,8 +23,8 @@ def orchestrator_generator_wrapper(
 
     Parameters
     ----------
-    generator: Generator[Any, Any, Any]
-        The user orchestrator generator. It is expected to yield task objects
+    generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any]
+        The user orchestrator generator or coroutine. It must suspend on tasks
         that expose a ``result`` attribute and to be resumed with those results
         until it returns the orchestration output.
 
@@ -34,13 +35,12 @@ def orchestrator_generator_wrapper(
         orchestrator's final return value.
     """
     try:
-        previous = next(generator)
+        previous = generator.send(None)
     except StopIteration as e:
         # An orchestrator that returns immediately without yielding any task
         # completes right away (mirroring the runtime, which finishes the
         # orchestration on the first return). Surface its output as the final
-        # value and stop -- ``next`` raising here otherwise leaks as an
-        # unhandled ``StopIteration``.
+        # value and stop -- letting StopIteration escape would become RuntimeError.
         yield e.value
         return
     yield previous

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- With the corresponding core SDK prototype, `DurableTaskSchedulerWorker`
+accepts `async def` orchestrators that await durable tasks alongside generator
+orchestrators. Requires a coordinated release after the core SDK is published;
+dependency minimums are intentionally unchanged in this spike.
+
 ## v1.12.0
 
 ADDED

durabletask/task.py

@@ -7,7 +7,7 @@
 import logging
 import math
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Generator, Sequence
+from collections.abc import Callable, Coroutine, Generator, Sequence
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Generic, TypeAlias, TypeVar, cast, overload
@@ -631,6 +631,16 @@ def __init__(self) -> None:
         # Dict insertion order preserves subscription order for shared races.
         self._parents = {}
 
+    def __await__(self) -> Generator[Task[T], Any, T]:
+        """Wait for this task in a replay-driven ``async def`` orchestrator.
+
+        Durable tasks are not asyncio futures. Only the orchestration replay
+        engine may drive a coroutine that awaits them.
+        """
+        if not self.is_complete:
+            yield self
+        return self.get_result()
+
     @property
     def is_complete(self) -> bool:
         """Returns True if the task has completed, False otherwise."""
@@ -961,8 +971,11 @@ def task_id(self) -> int:
         return self._task_id
 
 
-# Orchestrators are generators that yield tasks, receive any type, and return TOutput
-Orchestrator: TypeAlias = Callable[[OrchestrationContext, TInput], Generator[Task[Any], Any, TOutput] | TOutput]
+# Orchestrators yield or await durable tasks, or return an immediate result.
+Orchestrator: TypeAlias = Callable[
+    [OrchestrationContext, TInput],
+    Generator[Task[Any], Any, TOutput] | Coroutine[Any, Any, TOutput] | TOutput,
+]
 
 # Activities are simple functions that can be scheduled by orchestrators
 Activity: TypeAlias = Callable[[ActivityContext, TInput], TOutput]

durabletask/worker.py

@@ -8,12 +8,12 @@
 import logging
 import os
 import time
-from collections.abc import Callable, Generator, Sequence
+from collections.abc import Callable, Coroutine, Generator, Sequence
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from threading import Event, Lock, Thread
-from types import GeneratorType
+from types import CoroutineType, GeneratorType
 from enum import Enum
 from typing import Any, TypeVar, cast, overload
 import uuid
@@ -1503,7 +1503,7 @@ def _cancel_entity_batch(
 
 
 class _RuntimeOrchestrationContext(task.OrchestrationContext):
-    _generator: Generator[task.Task[Any], Any, Any] | None
+    _generator: Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any] | None
     _previous_task: task.Task[Any] | None
 
     def __init__(self,
@@ -1549,21 +1549,26 @@ def __init__(self,
         self._exception_properties_provider = exception_properties_provider
         self._logger = logger
 
-    def run(self, generator: Generator[task.Task[Any], Any, Any]) -> None:
+    def run(self, generator: Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any]) -> None:
         self._generator = generator
-        # TODO: Do something with this task
-        task = next(generator)  # this starts the generator
-        # TODO: Check if the task is null?
-        self._previous_task = task
+        self._set_next_task(generator.send(None))
+
+    def _set_next_task(self, next_task: Any) -> None:
+        if not isinstance(next_task, task.Task):
+            raise TypeError(
+                "Orchestrators must only yield or await durabletask.Task objects. "
+                "Use durable timers and task.when_all/when_any instead of asyncio APIs."
+            )
+        self._previous_task = next_task
 
     def resume(self) -> None:
         if self._generator is None:
             # This is never expected unless maybe there's an issue with the history
             raise TypeError(
-                "The orchestrator generator is not initialized! Was the orchestration history corrupted?"
+                "The orchestrator is not initialized! Was the orchestration history corrupted?"
             )
 
-        # We can resume the generator only if the previously yielded task
+        # We can resume the generator or coroutine only if the previous task
         # has reached a completed state. The only time this won't be the
         # case is if the user yielded on a WhenAll task and there are still
         # outstanding child tasks that need to be completed.
@@ -1578,9 +1583,7 @@ def resume(self) -> None:
                 # This will either return a Task or raise StopIteration if it's done.
                 next_task = self._generator.send(self._previous_task.get_result())
 
-            if not isinstance(next_task, task.Task):
-                raise TypeError("The orchestrator generator yielded a non-Task object")
-            self._previous_task = next_task
+            self._set_next_task(next_task)
 
     def set_complete(
             self,
@@ -2521,10 +2524,15 @@ def process_event(
 
                 result = fn(
                     ctx, input
-                )  # this does not execute the generator, only creates it
-                if isinstance(result, GeneratorType):
-                    # Start the orchestrator's generator function
-                    ctx.run(cast(Generator[task.Task[Any], Any, Any], result))
+                )  # this creates the generator or coroutine without executing it
+                if isinstance(result, (GeneratorType, CoroutineType)):
+                    # Both protocols suspend on durable tasks and finish with StopIteration.
+                    ctx.run(cast(Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any], result))
+                elif inspect.isasyncgen(result):
+                    raise TypeError(
+                        "Async generator orchestrators are not supported. "
+                        "Use async def with await, not yield."
+                    )
                 else:
                     # This is an orchestrator that doesn't schedule any tasks
                     ctx.set_complete(result, pb.ORCHESTRATION_STATUS_COMPLETED)

tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py

@@ -35,6 +35,15 @@ def activity_chain(ctx: task.OrchestrationContext, _: Any):
     return [first, second, third]
 
 
+@bp.orchestration_trigger(context_name="context")
+async def async_await_probe(ctx: task.OrchestrationContext, _: Any) -> dict[str, Any]:
+    greeting = await ctx.call_activity("say_hello", input="Tokyo", return_type=str)
+    values = await task.when_all([ctx.call_activity("square", input=i, return_type=int) for i in (2, 3)])
+    await ctx.create_timer(timedelta(milliseconds=10))
+    child: list[str] = await ctx.call_sub_orchestrator("activity_chain")
+    return {"greeting": greeting, "total": sum(values), "child": child}
+
+
 @bp.orchestration_trigger(context_name="context")
 def logging_filtered(ctx: task.OrchestrationContext, _: Any):
     _LOGGER.info("worker-info-filter-anchor %s", ctx.instance_id)

tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py

@@ -13,6 +13,7 @@
 """
 
 from datetime import timedelta
+from typing import Any
 
 import azure.durable_functions as df
 
@@ -31,6 +32,15 @@ def activity_chain(context: df.DurableOrchestrationContext):
     return [first, second, third]
 
 
+@bp.orchestration_trigger(context_name="context")
+async def async_await_probe(context: df.DurableOrchestrationContext) -> dict[str, Any]:
+    greeting = await context.call_activity("say_hello", "Tokyo")
+    values = await context.task_all([context.call_activity("square", i) for i in (2, 3)])
+    await context.create_timer(context.current_utc_datetime + timedelta(milliseconds=10))
+    child = await context.call_sub_orchestrator("activity_chain")
+    return {"greeting": greeting, "total": sum(values), "child": child}
+
+
 @bp.orchestration_trigger(context_name="context")
 def fan_out_fan_in(context: df.DurableOrchestrationContext):
     count = context.get_input() or 5

tests/azure-functions-durable/e2e/test_coroutine_e2e.py

@@ -0,0 +1,24 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Real-host smoke coverage for both async orchestrator authoring styles."""
+
+import pytest
+
+from ._harness import FunctionApp
+
+
+pytestmark = pytest.mark.functions_e2e
+
+
+@pytest.mark.parametrize("app_fixture", ["dtask_app", "v1_app"])
+def test_async_await_orchestrator(request: pytest.FixtureRequest, app_fixture: str) -> None:
+    app: FunctionApp = request.getfixturevalue(app_fixture)
+    instance_id = app.start_orchestration("async_await_probe")
+    status = app.wait_for_completion(instance_id)
+    assert status["runtimeStatus"].upper() == "COMPLETED", status
+    assert status["output"] == {
+        "greeting": "Hello Tokyo!",
+        "total": 13,
+        "child": ["Hello Tokyo!", "Hello Seattle!", "Hello London!"],
+    }