Document coroutine orchestrators and add Managed compatibility tests

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: andystaples

CHANGELOG.md

@@ -7,6 +7,23 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- Added `async def` orchestrators with an automatically supplied
+`AsyncOrchestrationContext`, cold coroutine operations, durable asyncio
+timers, and supported task-group and gather composition. Generator
+orchestrators retain their eager durable handles. `as_task_context()` and
+`Task.wait()` enable explicit interoperability with those handles.
+- Async event waits support owned cancellation and timeouts, withdrawing
+abandoned subscriptions so they cannot consume later events.
+
+> [!WARNING]
+> Async cleanup can run at every dispatch, including suspension; it is not
+> durable workflow compensation. Converting running orchestrations or changing
+> Python versions may alter replay-visible action order. See the
+> [async contract](docs/async-orchestrators.md) for supported operations and
+> replay and upgrade guidance.
+
 FIXED
 
 - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe

README.md

@@ -32,6 +32,53 @@ The Azure Functions Durable provider requires Python 3.13 or later.
 
 ## Optional Features
 
+### Async Orchestrators
+
+All three packages support `async def` orchestrators and nested async helpers.
+Existing generator orchestrators continue to work.
+
+```python
+import asyncio
+
+from durabletask import task
+
+
+async def sequence(ctx: task.AsyncOrchestrationContext, value: int) -> list[int]:
+    first = await ctx.call_activity("plus_one", input=value, return_type=int)
+    async with asyncio.TaskGroup() as group:
+        left = group.create_task(ctx.call_activity("plus_one", input=first, return_type=int))
+        right = group.create_task(ctx.call_activity("plus_one", input=value, return_type=int))
+    return [left.result(), right.result()]
+```
+
+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(...)`. Async orchestrators automatically
+receive cold coroutine operations accepted by `create_task` and task groups.
+Generator orchestrators keep their eager, reusable durable handles. For
+interoperability, `ctx.as_task_context()` exposes the eager context and
+`handle.wait()` creates an independently cancellable borrower.
+
+> [!WARNING]
+> This is replay-driven await syntax, not general-purpose asyncio execution.
+> Compose durable work with `asyncio.gather`, local tasks/task groups, and
+> durable `asyncio.sleep` / timeouts within the
+> [supported contract](docs/async-orchestrators.md).
+> 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.
+> Python `finally` blocks and async context-manager exits can run at ordinary
+> dispatch suspension; they are not workflow-lifetime compensation scopes.
+
+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.
+See the contract for cancellation, per-dispatch async cleanup, unsupported
+ordering-sensitive APIs and framework adapters.
+
 ### Task Hub Query Filters
 
 Set `OrchestrationQuery.task_hub_names` to filter orchestration queries by one or

azure-functions-durable/CHANGELOG.md

@@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ADDED
 
+- With the corresponding core SDK, native and v1-style `async def`
+orchestrators automatically receive async contexts with cold durable operations
+and supported asyncio composition. The v1-style context preserves compatibility
+argument names and provides async `task_all` and `task_any`. Generator
+orchestrators retain their eager behavior. See the
+[async contract](../docs/async-orchestrators.md) for dispatch cleanup and
+replay/upgrade precautions.
+- The `orchestrator_generator_wrapper` test helper accepts direct-await
+coroutines. Tests involving asyncio scheduling still require a replay executor
+or backend.
 - Added opt-in `HttpRetryOptions` on `context.call_http()` for durable initial
 request retries with configurable timing and status selection. Empty or omitted
 status lists select all non-2xx responses; exhausted retries fail the call.
@@ -59,6 +69,7 @@ readers before enabling new writers; existing headerless blobs still require
 their original compression setting. See the
 [core compatibility guidance](../docs/features.md#cross-sdk-compatibility).
 
+
 ## v2.0.0rc4
 
 ADDED

azure-functions-durable/MIGRATION_GUIDE.md

@@ -102,10 +102,18 @@ The single-argument orchestrator receives the compatibility
 Use this state to validate the infrastructure upgrade before changing
 application APIs.
 
+In 2.x, a single-argument `async def` orchestrator instead receives
+`AsyncDurableOrchestrationContext`, with the same v1-style argument names and
+awaitable durable operations. Generator orchestrators retain their existing
+eager task-handle behavior. See the
+[async authoring contract](../docs/async-orchestrators.md) before converting
+running orchestrations.
+
 ## Migrate to native orchestration APIs
 
 Native orchestrators accept the durabletask context and input as separate
-arguments:
+arguments. Generator functions receive `task.OrchestrationContext`;
+`async def` functions receive `task.AsyncOrchestrationContext`:
 
 ```python
 from typing import Any

azure-functions-durable/README.md

@@ -48,6 +48,30 @@ their existing page sizes.
 so they can be used as dictionary keys or set elements. Construct a new ID
 instead of assigning or deleting attributes on an existing ID.
 
+## Async orchestrators
+
+Both authoring styles accept `async def` and `await` without an adapter loop.
+Await durable operations and compose branches with `asyncio.gather` or task
+groups; `asyncio.sleep` and timeouts use durable timers. Synchronous generator
+orchestrators remain supported, including mixed-style sub-orchestrations.
+
+Async orchestrators automatically receive cold coroutine operations accepted by
+`create_task` and task groups. Native orchestrators receive
+`task.AsyncOrchestrationContext`; single-argument orchestrators receive
+`df.AsyncDurableOrchestrationContext`, retaining v1-style argument names and
+`await context.task_all(...)` / `await context.task_any(...)`.
+`wait_for_external_event()` owns its registration and optional durable timeout,
+so an abandoned receive cannot consume the next event. Generator contexts
+retain eager, reusable durable handles; async code can access those through
+`context.as_task_context()` and use `handle.wait()` to create a borrowed waiter.
+
+The [shared async contract](../docs/async-orchestrators.md) defines the supported
+API subset, cancellation, per-dispatch `finally`/async-generator cleanup,
+replay/migration rules, and framework adapter responsibilities. Arbitrary I/O,
+threads, subprocesses, `asyncio.wait`/`as_completed`, and automatic recording of
+ordinary async helpers are not supported. Put nondeterministic work in ordinary
+activities, not orchestrators.
+
 ## Durable HTTP failure retries
 
 Opt in to durable retries for the **initial** `context.call_http()` request with
@@ -93,6 +117,7 @@ rather than returning the last selected response. Transport and other activity
 failures also use the policy regardless of the status list.
 
 The policy and status list are snapshotted when the request is scheduled.
+For async orchestrators, scheduling occurs when the coroutine starts.
 Omitting `retry_options` preserves the existing request payload and behavior.
 Automatic 202/Location polling remains separate: subsequent GET polls inherit
 neither the failure retry policy nor its status classification. Poll HTTP

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,16 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- With the corresponding core SDK, `DurableTaskSchedulerWorker` accepts
+`async def` orchestrators with an automatically supplied async context, cold
+durable operations, and supported asyncio composition. Generator orchestrators
+retain their existing behavior. Event waits support owned cancellation and
+timeouts; remote operations are not cancelled by local waiter cancellation.
+See the [async contract](../docs/async-orchestrators.md) for dispatch cleanup
+and replay/upgrade precautions.
+
 FIXED
 
 - With the corresponding core SDK update, activity and sub-orchestration

tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py

@@ -0,0 +1,102 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Exercise both workers over gRPC and optionally against the DTS emulator."""
+
+import json
+import asyncio
+import os
+import socket
+from collections.abc import Generator
+from datetime import timedelta
+from typing import Any
+
+import pytest
+
+from durabletask import client, task, worker
+from durabletask.azuremanaged.client import DurableTaskSchedulerClient
+from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker
+from durabletask.testing import create_test_backend
+
+
+@pytest.mark.parametrize("transport", [
+    "core",
+    "azuremanaged",
+    pytest.param("dts", marks=pytest.mark.dts),
+])
+@pytest.mark.parametrize("coroutine_parent", [False, True], ids=["generator-parent", "coroutine-parent"])
+@pytest.mark.parametrize("borrowed", [False, True], ids=["coroutines", "borrowed-handles"])
+def test_mixed_orchestrators_over_grpc(transport: str, coroutine_parent: bool, borrowed: bool) -> None:
+    def increment(ctx: task.ActivityContext, value: int) -> int:
+        return value + 1
+
+    async def coroutine_child(ctx: task.AsyncOrchestrationContext, value: int) -> int:
+        if not borrowed:
+            await ctx.create_timer(timedelta(milliseconds=10))
+            return await ctx.call_activity(increment, input=value, return_type=int)
+        await asyncio.sleep(0)
+        await asyncio.sleep(0.01)
+        return await ctx.as_task_context().call_activity(increment, input=value, return_type=int)
+
+    def generator_child(ctx: task.OrchestrationContext, value: int) -> Generator[task.Task[Any], Any, int]:
+        return (yield ctx.call_activity(increment, input=value))
+
+    async def coroutine(ctx: task.AsyncOrchestrationContext, value: int) -> list[Any]:
+        if not borrowed:
+            async with asyncio.TaskGroup() as group:
+                first = group.create_task(ctx.call_sub_orchestrator(coroutine_child, input=value))
+                second = group.create_task(ctx.call_sub_orchestrator(generator_child, input=value + 1))
+            return [first.result(), second.result(), await ctx.wait_for_external_event("go")]
+        handles = ctx.as_task_context()
+        values = list(await asyncio.gather(
+            handles.call_sub_orchestrator(coroutine_child, input=value),
+            handles.call_sub_orchestrator(generator_child, input=value + 1),
+        ))
+        event = await ctx.wait_for_external_event("go")
+        return values + [event]
+
+    def generator(ctx: task.OrchestrationContext, value: int) -> Generator[task.Task[Any], Any, list[Any]]:
+        values = yield task.when_all([
+            ctx.call_sub_orchestrator(coroutine_child, input=value),
+            ctx.call_sub_orchestrator(generator_child, input=value + 1),
+        ])
+        event = yield ctx.wait_for_external_event("go")
+        return values + [event]
+
+    if transport == "dts":
+        backend = None
+        host = os.environ["ENDPOINT"]
+        taskhub = os.environ["TASKHUB"]
+    else:
+        with socket.socket() as listener:
+            listener.bind(("127.0.0.1", 0))
+            port = listener.getsockname()[1]
+        backend = create_test_backend(port=port)
+        host = f"localhost:{port}"
+        taskhub = "test"
+    if transport != "core":
+        w = DurableTaskSchedulerWorker(
+            host_address=host, secure_channel=False, taskhub=taskhub, token_credential=None)
+        c = DurableTaskSchedulerClient(
+            host_address=host, secure_channel=False, taskhub=taskhub, token_credential=None)
+    else:
+        w = worker.TaskHubGrpcWorker(host_address=host)
+        c = client.TaskHubGrpcClient(host_address=host)
+    try:
+        with w, c:
+            parent = coroutine if coroutine_parent else generator
+            for fn in (parent, coroutine_child, generator_child):
+                w.add_orchestrator(fn)
+            w.add_activity(increment)
+            w.start()
+            instance_id = c.schedule_new_orchestration(parent, input=3)
+            c.raise_orchestration_event(instance_id, "go", data="approved")
+            state = c.wait_for_orchestration_completion(instance_id, timeout=15)
+            assert state is not None
+            assert state.runtime_status == client.OrchestrationStatus.COMPLETED, state.failure_details
+            assert state.serialized_output is not None
+            assert json.loads(state.serialized_output) == [4, 5, "approved"]
+    finally:
+        if backend is not None:
+            backend.stop()
+            backend.reset()