diff --git a/CHANGELOG.md b/CHANGELOG.md
index 40aa2e88..d830fa92 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,25 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+ADDED
+
+- Added `async def` orchestrators. They receive an `AsyncOrchestrationContext`
+whose durable operations are cold coroutines, support durable asyncio timers
+and ordered composition such as `asyncio.gather`, `asyncio.TaskGroup`, and
+`asyncio.timeout`, and offer event waits with owned cancellation and timeouts.
+Generator orchestrators are unchanged; `as_task_context()` and `Task.wait()`
+interoperate with their eager durable handles. Pass `return_type` to type and
+reconstruct async sub-orchestration results. `asyncio.wait` and
+`asyncio.as_completed` are not supported.
+
+> [!WARNING]
+> Async cleanup can run at every dispatch, including suspension; it is not
+> durable workflow compensation. Converting running orchestrations can alter
+> replay-visible action order, and recorded async histories depend on the
+> Python minor version, so keep it fixed for in-flight instances. 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
diff --git a/README.md b/README.md
index fb947a98..8bd214fd 100644
--- a/README.md
+++ b/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. All three
+packages use this two-argument shape for async orchestrators. 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
diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md
index 4dd19e87..8c0feec1 100644
--- a/azure-functions-durable/CHANGELOG.md
+++ b/azure-functions-durable/CHANGELOG.md
@@ -11,7 +11,14 @@ ADDED
 
 - A best-effort background check warns when a Function App's requirements allow
 a future major version, which may introduce replay-breaking changes.
-
+- Added `async def` orchestrators with the core two-argument shape,
+`async def orchestrator(ctx: task.AsyncOrchestrationContext, input)`, and
+`df.call_http()` for durable HTTP requests from durabletask orchestration
+contexts. Single-parameter `async def` orchestrators are not supported and
+raise `TypeError` when registered. Generator orchestrators are unchanged.
+Requires a `durabletask` release with async orchestrator support. See the
+[async contract](../docs/async-orchestrators.md) for dispatch cleanup and
+replay/upgrade precautions.
 - 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.
diff --git a/azure-functions-durable/MIGRATION_GUIDE.md b/azure-functions-durable/MIGRATION_GUIDE.md
index 20d992da..c9c1f34d 100644
--- a/azure-functions-durable/MIGRATION_GUIDE.md
+++ b/azure-functions-durable/MIGRATION_GUIDE.md
@@ -102,10 +102,16 @@ The single-argument orchestrator receives the compatibility
 Use this state to validate the infrastructure upgrade before changing
 application APIs.
 
+Single-argument orchestrators must be generator or regular functions. To write
+an `async def` orchestrator, use the native two-argument shape described next;
+the `orchestration_trigger` decorator raises `TypeError` for a single-argument
+`async def` orchestrator.
+
 ## 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
@@ -148,16 +154,58 @@ def say_hello(name: str) -> str:
 | `context.call_activity_with_retry(...)` | `ctx.call_activity(..., retry_policy=policy)` |
 | `context.call_sub_orchestrator(name, input_)` | `ctx.call_sub_orchestrator(name, input=input_)` |
 | `call_sub_orchestrator_with_retry` | `ctx.call_sub_orchestrator(..., retry_policy=...)` |
-| `context.task_all(tasks)` | `task.when_all(tasks)` |
-| `context.task_any(tasks)` | `task.when_any(tasks)` |
+| `context.task_all(tasks)` | `task.when_all(tasks)` (eager `task.Task` handles only) |
+| `context.task_any(tasks)` | `task.when_any(tasks)` (eager `task.Task` handles only) |
 | `context.new_guid()` | `ctx.new_uuid()` |
+| `context.call_http(method, uri, ...)` | `df.call_http(ctx, method, uri, ...)` |
 | `df.EntityId(name, key)` | `entities.EntityInstanceId(name, key)` |
 
 Methods including `create_timer`, `wait_for_external_event`,
 `continue_as_new`, `set_custom_status`, `call_entity`, and `signal_entity`
 retain their names. Activity, sub-orchestration, and entity calls use keyword
 `input` instead of the 1.x `input_` or `operationInput` names.
-`continue_as_new` uses `new_input`.
+`continue_as_new` uses `new_input`. `df.call_http` takes the request options
+as keyword arguments.
+
+### Compose async operations
+
+The grouping mappings above apply to generator orchestrators and explicit eager
+durable handles, not the cold coroutines returned by
+`task.AsyncOrchestrationContext`. Do not pass `ctx.call_activity(...)` coroutines
+directly to `task.when_all` or `task.when_any`.
+
+For fan-out in an async orchestrator, use ordered `asyncio.gather` or
+`asyncio.TaskGroup`:
+
+```python
+async def greetings(ctx: task.AsyncOrchestrationContext, names: list[str]) -> list[str]:
+    return list(await asyncio.gather(*(
+        ctx.call_activity("say_hello", input=name, return_type=str)
+        for name in names
+    )))
+```
+
+These are not interchangeable with the generator `task.when_all` join, which
+waits for every task before raising the first observed failure: default
+`gather` reports the first failure without cancelling siblings, while task
+groups cancel and drain local siblings on failure. Local cancellation does not
+stop already-issued remote work.
+
+For races, obtain eager handles through `ctx.as_task_context()` and pass them to
+`task.when_any`, which returns the identical winning handle:
+
+```python
+async def approval(ctx: task.AsyncOrchestrationContext, _: Any) -> bool:
+    handles = ctx.as_task_context()
+    approved = handles.wait_for_external_event("approval", data_type=bool)
+    deadline = handles.create_timer(timedelta(hours=1))
+    winner = await task.when_any([approved, deadline])
+    return winner is approved and approved.get_result()
+```
+
+Losing handles are not cancelled; retain and join them when their completion
+matters. See the [async authoring contract](../docs/async-orchestrators.md)
+before changing joins or cancellation behavior in running orchestrations.
 
 ### Retry policy mapping
 
diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md
index 29da298a..dc5c0852 100644
--- a/azure-functions-durable/README.md
+++ b/azure-functions-durable/README.md
@@ -108,9 +108,85 @@ These compatibility paths remain supported.
 > Changing typing modes can change values returned during orchestration replay.
 > Validate existing payloads and drain affected orchestrations before switching.
 
+## Async orchestrators
+
+Async orchestrators use the same two-argument shape as the core SDK and Durable
+Task Scheduler provider. They receive `task.AsyncOrchestrationContext` and the
+orchestration input:
+
+```python
+import asyncio
+from datetime import timedelta
+from typing import Any
+
+import azure.durable_functions as df
+from durabletask import task
+
+app = df.DFApp()
+
+
+@app.orchestration_trigger(context_name="context")
+async def fan_out(ctx: task.AsyncOrchestrationContext, names: list[str]) -> list[str]:
+    return list(await asyncio.gather(*(
+        ctx.call_activity("say_hello", input=name, return_type=str)
+        for name in names
+    )))
+
+
+@app.orchestration_trigger(context_name="context")
+async def approval(ctx: task.AsyncOrchestrationContext, _: Any) -> bool:
+    handles = ctx.as_task_context()
+    approved = handles.wait_for_external_event("approval", data_type=bool)
+    deadline = handles.create_timer(timedelta(hours=1))
+    winner = await task.when_any([approved, deadline])
+    return winner is approved and approved.get_result()
+```
+
+Durable operations on the async context are cold coroutines. Compose branches
+with `asyncio.gather` or `asyncio.TaskGroup`; `asyncio.sleep` and timeouts use
+durable timers. For races, schedule eager handles through
+`ctx.as_task_context()` and pass them to `task.when_any`, which returns the
+identical winning handle. `handle.wait()` creates an independently cancellable
+waiter for an eager handle. Generator orchestrators remain supported, including
+mixed-style sub-orchestrations.
+
+Single-argument `async def` orchestrators are not supported. The
+`orchestration_trigger` decorator raises `TypeError` for them. The
+single-argument `df.DurableOrchestrationContext` remains available for
+generator orchestrators.
+
+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 calls
+
+`df.call_http` makes durable HTTP requests from durabletask orchestration
+contexts. With `task.AsyncOrchestrationContext`, it returns a cold coroutine to
+`await`. With a generator's `task.OrchestrationContext`, it schedules the
+request immediately and returns a task to `yield`. The result is a
+`df.DurableHttpResponse`:
+
+```python
+@app.orchestration_trigger(context_name="context")
+async def check_status(ctx: task.AsyncOrchestrationContext, url: str) -> int:
+    response = await df.call_http(ctx, "GET", url)
+    return response.status_code
+```
+
+`df.call_http` accepts the same `content`, `headers`, `token_source`,
+`is_raw_str`, and `retry_options` options as `context.call_http()` on
+`df.DurableOrchestrationContext`, as keyword arguments. Both follow `202
+Accepted` responses with a `Location` header until the operation completes and
+return unselected non-2xx responses rather than raising.
+
 ## Durable HTTP failure retries
 
-Opt in to durable retries for the **initial** `context.call_http()` request with
+Opt in to durable retries for the **initial** `context.call_http()` or
+`df.call_http()` request with
 `HttpRetryOptions`. It uses the core activity retry engine and durable timers,
 not an in-process HTTP retry loop:
 
@@ -153,6 +229,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
@@ -297,8 +374,9 @@ agree on the store and reference encoding; Functions references are JSON strings
 > storage error can therefore become a terminal orchestration failure.
 
 Registered orchestration and entity handlers await the store's async methods
-before and after execution. Orchestrators remain synchronous generators, and
-orchestration replay and entity code run on execution threads with their
+before and after execution. Orchestrators may be synchronous generators or
+native `async def` functions; both use synchronous replay execution.
+Orchestration replay and entity code run on execution threads with their
 invocation logging context preserved. Their payload downloads and uploads do
 not occupy those threads, and serialization does not access storage during
 replay. Custom stores must implement genuinely nonblocking async methods to
diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md
index 8de9b64d..55794d71 100644
--- a/docs/async-orchestrators.md
+++ b/docs/async-orchestrators.md
@@ -4,8 +4,8 @@ All three packages accept native `async def` orchestrators alongside existing
 synchronous generator orchestrators. Both styles use the same history, action,
 retry, payload, and entity engine. Register the function normally with
 `TaskHubGrpcWorker`, `DurableTaskSchedulerWorker`, or the Functions
-`orchestration_trigger` decorator. Functions supports both native two-argument
-and compatibility single-argument orchestrators.
+`orchestration_trigger` decorator. Async orchestrators use the same
+two-argument `(ctx, input)` shape in all three packages.
 
 ```python
 import asyncio
@@ -156,47 +156,47 @@ unmarked synchronous wrapper returning a coroutine is rejected because it would
 have received the generator context. Cyclic `__wrapped__` chains are rejected at
 registration. Existing registration naming requirements still apply.
 
-## Azure Functions compatibility
+## Azure Functions
 
-Single-argument async Functions orchestrators receive
-`df.AsyncDurableOrchestrationContext`, with the v1-style method and argument
-names. Two-argument native orchestrators receive `task.AsyncOrchestrationContext`.
+Functions async orchestrators use the two-argument shape and receive
+`task.AsyncOrchestrationContext`, as in the other packages. The second argument
+is the orchestration input. Single-argument `async def` orchestrators are not
+supported: the `orchestration_trigger` decorator raises `TypeError` for them.
+The single-argument `df.DurableOrchestrationContext` remains available for
+generator orchestrators.
+
+Durable HTTP requests use `df.call_http`, which accepts the async context and
+returns a cold coroutine:
 
 ```python
+import asyncio
+
 import azure.durable_functions as df
+from durabletask import task
 
 app = df.DFApp()
 
 
 @app.orchestration_trigger(context_name="context")
-async def greetings(context: df.AsyncDurableOrchestrationContext) -> list[str]:
-    names = context.get_input()
-    return await context.task_all([
-        context.call_activity("say_hello", name) for name in names
-    ])
+async def greetings(
+        ctx: task.AsyncOrchestrationContext, names: list[str]) -> dict[str, object]:
+    messages = await asyncio.gather(*(
+        ctx.call_activity("say_hello", input=name, return_type=str)
+        for name in names
+    ))
+    response = await df.call_http(ctx, "GET", "https://example.com/status")
+    return {"greetings": messages, "status": response.status_code}
 ```
 
-`task_all` starts its inputs, waits for all of them, and returns results in
-input order, including duplicate references. If an operation fails, the first
-observed failure is raised after all have finished. `task_any` returns the
-winning `asyncio.Future`, whether it succeeded, failed, or was cancelled,
-without cancelling the losers. Await the winner to obtain its result or failure.
-An existing future keeps its identity; a coroutine is wrapped in a new task.
-Create tasks yourself when you need to identify the winner or later await the
-losers. Repeated references within a group are scheduled only once.
-
-On group cancellation, coroutines started by that group are cancelled and
-drained, while supplied futures are borrowed and continue running. Remote work
-already issued by either kind of operation is not cancelled. The compatibility
-context's `as_task_context()` returns its eager v1-style counterpart when
-explicit durable handles are needed.
+Compose branches with `asyncio.gather` or `asyncio.TaskGroup`, and use
+`ctx.as_task_context()` handles with `task.when_any` for races, as described
+above.
 
 ## Supported authoring contract
 
 Await activities, timers, external events, sub-orchestrations, durable
 operations via eager `task.when_all` / `task.when_any`, and entity calls/locks
-where supported by the provider. Functions compatibility contexts also provide
-async `task_all` / `task_any`. Generator and coroutine parents/children may call
+where supported by the provider. Generator and coroutine parents/children may call
 one another through durable sub-orchestrations.
 
 For sub-orchestration results, pass `return_type=Result` to reconstruct a custom
@@ -409,8 +409,7 @@ instances and validate histories before upgrades. All workers processing an
 instance must use that pinned environment; see the Python version warning
 above.
 
-The Functions `orchestrator_generator_wrapper` helper supports direct durable
-awaits in synchronous and async mock-based unit tests; it does not simulate the
-asyncio scheduler.
-Use the real replay executor/provider or the in-memory backend to test task
-groups, gather, and durable timeouts.
+The Functions `orchestrator_generator_wrapper` helper drives generator
+orchestrators only. Use the real replay executor/provider or the in-memory
+backend to test async orchestrators, including task groups, gather, and durable
+timeouts.
diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md
index 1b327beb..54f86a1c 100644
--- a/durabletask-azuremanaged/CHANGELOG.md
+++ b/durabletask-azuremanaged/CHANGELOG.md
@@ -7,6 +7,17 @@ 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. Requires a `durabletask` release with async
+orchestrator support.
+
 FIXED
 
 - With the corresponding core SDK update, activity and sub-orchestration
diff --git a/eng/templates/build.yml b/eng/templates/build.yml
index 655833d8..67fd7e1c 100644
--- a/eng/templates/build.yml
+++ b/eng/templates/build.yml
@@ -110,9 +110,12 @@ jobs:
         displayName: "pytest: durabletask (unit tests, no emulators)"
         workingDirectory: tests/durabletask
 
-      # `tests/durabletask-azuremanaged/` is  NOT exercised here.
-      # Those tests require the Durable Task Scheduler emulator which is 
-      # not available in the network-isolated pools.
+      # Exercise the in-memory gRPC compatibility cases without an emulator.
+      # DTS-backed cases remain in the GitHub Actions emulator matrix.
+      - script: pytest test_dts_coroutine_orchestrators.py -m "not dts" --verbose
+        displayName: "pytest: durabletask-azuremanaged (coroutines, no emulator)"
+        workingDirectory: tests/durabletask-azuremanaged
+
       - script: |
           set -e
           python -P -c "import durabletask.azuremanaged; from durabletask.azuremanaged.client import DurableTaskSchedulerClient; from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker; print('durabletask.azuremanaged smoke import OK')"
diff --git a/noxfile.py b/noxfile.py
index 15cab4be..929ee5c4 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -457,6 +457,13 @@ def azuremanaged_tests(session: nox.Session) -> None:
             "pytest", *_pytest_warning_arguments(session),
             *arguments, "-m", "dts", "--verbose",
         )
+        if not session.posargs:
+            # In-memory gRPC cases need no emulator but still gate PRs here.
+            session.run(
+                "pytest", *_pytest_warning_arguments(session),
+                "tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py",
+                "-m", "not dts", "--verbose",
+            )
     finally:
         _stop_dts_emulator(container_name)
 
diff --git a/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py
new file mode 100644
index 00000000..c2bbca90
--- /dev/null
+++ b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py
@@ -0,0 +1,128 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Exercise both workers over gRPC and optionally against the DTS emulator."""
+
+import asyncio
+import importlib.util
+import json
+import os
+from collections.abc import Callable, Generator
+from datetime import timedelta
+from pathlib import Path
+from typing import Any, cast
+
+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
+
+
+def _load_find_free_port() -> Callable[[], int]:
+    # pytest imports this hyphenated directory's modules top-level, so the shared
+    # core test helper cannot be imported through the `tests` package here.
+    path = Path(__file__).resolve().parents[1] / "durabletask" / "_port_utils.py"
+    spec = importlib.util.spec_from_file_location("_durabletask_test_port_utils", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return cast(Callable[[], int], module.find_free_port)
+
+
+find_free_port = _load_find_free_port()
+taskhub_name = os.getenv("TASKHUB", "default")
+endpoint = os.getenv("ENDPOINT", "http://localhost:8080")
+
+
+def _get_credential():
+    """Returns DefaultAzureCredential if endpoint is https, otherwise None (for emulator)."""
+    if endpoint.startswith("https://"):
+        from azure.identity import DefaultAzureCredential
+        return DefaultAzureCredential()
+    return None
+
+
+@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 = endpoint
+        taskhub = taskhub_name
+        credential = _get_credential()
+    else:
+        port = find_free_port()
+        backend = create_test_backend(port=port)
+        # The in-memory backend has no TLS; the http:// scheme selects an insecure channel.
+        host = f"http://localhost:{port}"
+        taskhub = "test"
+        credential = None
+    if transport != "core":
+        w = DurableTaskSchedulerWorker(
+            host_address=host, secure_channel=True, taskhub=taskhub, token_credential=credential)
+        c = DurableTaskSchedulerClient(
+            host_address=host, secure_channel=True, taskhub=taskhub, token_credential=credential)
+    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()