From b4e4f97ce2a27e859bf0c44501e5d27ff3f9d646 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Thu, 15 Oct 2026 11:32:13 -0600 Subject: [PATCH 01/11] Prototype replay-driven async orchestrators across providers Make durable tasks awaitable, drive native coroutines through the existing replay engine, and preserve generator compatibility. Include provider smoke coverage, prototype documentation, and coordinated-release notes. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 7 + README.md | 38 +++ azure-functions-durable/CHANGELOG.md | 8 + .../testing/orchestrator_generator_wrapper.py | 20 +- durabletask-azuremanaged/CHANGELOG.md | 7 + durabletask/task.py | 19 +- durabletask/worker.py | 42 +-- .../e2e/apps/dtask_style/orchestrators.py | 9 + .../e2e/apps/v1_style/orchestrators.py | 10 + .../e2e/test_coroutine_e2e.py | 24 ++ .../test_functions_coroutine_orchestrators.py | 106 +++++++ .../test_dts_coroutine_orchestrators.py | 90 ++++++ .../test_coroutine_orchestrators.py | 279 ++++++++++++++++++ 13 files changed, 629 insertions(+), 30 deletions(-) create mode 100644 tests/azure-functions-durable/e2e/test_coroutine_e2e.py create mode 100644 tests/azure-functions-durable/test_functions_coroutine_orchestrators.py create mode 100644 tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py create mode 100644 tests/durabletask/test_coroutine_orchestrators.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 16194e95..cf3f4536 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 2bd8e7fb..efc47a06 100644 --- a/README.md +++ b/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 diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index b17a3654..8b9ba663 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/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 diff --git a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py index 43daf0c5..d516a697 100644 --- a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py +++ b/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 diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 338b183f..b441a795 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/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 diff --git a/durabletask/task.py b/durabletask/task.py index b2ca7b32..278527e6 100644 --- a/durabletask/task.py +++ b/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] diff --git a/durabletask/worker.py b/durabletask/worker.py index 25010a17..ef3cf3b3 100644 --- a/durabletask/worker.py +++ b/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) diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py index 34b44d98..347bce46 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py +++ b/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) diff --git a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py index a96467c7..d4766d87 100644 --- a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py +++ b/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 diff --git a/tests/azure-functions-durable/e2e/test_coroutine_e2e.py b/tests/azure-functions-durable/e2e/test_coroutine_e2e.py new file mode 100644 index 00000000..06215119 --- /dev/null +++ b/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!"], + } diff --git a/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py new file mode 100644 index 00000000..f264499c --- /dev/null +++ b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py @@ -0,0 +1,106 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Coroutine replay through native and compatibility Functions entry points.""" + +import base64 +import json +from collections.abc import Callable +from typing import Any + +import pytest + +import azure.durable_functions as df +from azure.durable_functions.orchestrator import Orchestrator +from azure.durable_functions.testing import orchestrator_generator_wrapper +from azure.durable_functions.worker import DurableFunctionsWorker +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb + + +@pytest.mark.parametrize("native", [False, True], ids=["compat", "native"]) +@pytest.mark.parametrize("entry", ["worker-sync", "worker-async", "decorator", "v1-handle"]) +async def test_coroutine_replay_through_functions_entry_points(native: bool, entry: str) -> None: + async def modern(ctx: task.OrchestrationContext, value: int) -> list[Any]: + return await task.when_all([ + ctx.call_activity("work", input=value, return_type=str), + ctx.call_activity("work", input=value + 1, return_type=str), + ]) + + async def compatible(ctx: df.DurableOrchestrationContext) -> list[Any]: + value = ctx.get_input() + return await ctx.task_all([ + ctx.call_activity("work", value), + ctx.call_activity("work", value + 1), + ]) + + fn: Callable[..., Any] = modern if native else compatible + worker = DurableFunctionsWorker() + handle = Orchestrator.create(fn) + app = df.DFApp() + builder = app.orchestration_trigger(context_name="ctx")(fn) + registered = builder.build().get_user_function() + instance_id = "async-functions" + + async def invoke(request: pb.OrchestratorRequest) -> pb.OrchestratorResponse: + encoded = base64.b64encode(request.SerializeToString()).decode() + if entry == "worker-sync": + output = worker.execute_orchestration_request(fn, encoded) + elif entry == "worker-async": + output = await worker.execute_orchestration_request_async(fn, encoded) + elif entry == "decorator": + output = await registered(ctx=encoded) + else: + output = handle(encoded) + response = pb.OrchestratorResponse() + response.ParseFromString(base64.b64decode(output)) + return response + + start = [ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event(fn.__name__, instance_id, "3"), + ] + first = await invoke(pb.OrchestratorRequest(instanceId=instance_id, newEvents=start)) + assert [a.scheduleTask.name for a in first.actions] == ["work", "work"] + assert [json.loads(a.scheduleTask.input.value) for a in first.actions] == [3, 4] + assert [a.id for a in first.actions] == [1, 2] + + response = await invoke(pb.OrchestratorRequest( + instanceId=instance_id, + pastEvents=start + [helpers.new_task_scheduled_event(i, "work") for i in (1, 2)], + newEvents=[ + helpers.new_task_completed_event(2, '"second"'), + helpers.new_task_completed_event(1, '"first"'), + ], + )) + assert len(response.actions) == 1 + complete = response.actions[0].completeOrchestration + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert json.loads(complete.result.value) == ["first", "second"] + + +@pytest.mark.parametrize("fail", [False, True]) +def test_testing_wrapper_drives_coroutine(fail: bool) -> None: + pending = task.CompletableTask[str]() + + async def orchestrator() -> str: + try: + return await pending + except task.TaskFailedError: + return "caught" + + values = orchestrator_generator_wrapper(orchestrator()) + assert next(values) is pending + if fail: + pending.fail("boom", ValueError("boom")) + else: + pending.complete("done") + assert list(values) == ["caught" if fail else "done"] + + +def test_testing_wrapper_drives_immediate_coroutine() -> None: + async def orchestrator() -> str: + return "done" + + assert list(orchestrator_generator_wrapper(orchestrator())) == ["done"] 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..42fd4e3a --- /dev/null +++ b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py @@ -0,0 +1,90 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Exercise both workers over gRPC and optionally against the DTS emulator.""" + +import json +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"]) +def test_mixed_orchestrators_over_grpc(transport: str, coroutine_parent: bool) -> None: + def increment(ctx: task.ActivityContext, value: int) -> int: + return value + 1 + + async def coroutine_child(ctx: task.OrchestrationContext, value: int) -> int: + await ctx.create_timer(timedelta(milliseconds=10)) + return await ctx.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.OrchestrationContext, value: int) -> list[Any]: + values = await task.when_all([ + ctx.call_sub_orchestrator(coroutine_child, input=value), + ctx.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() diff --git a/tests/durabletask/test_coroutine_orchestrators.py b/tests/durabletask/test_coroutine_orchestrators.py new file mode 100644 index 00000000..2eaef122 --- /dev/null +++ b/tests/durabletask/test_coroutine_orchestrators.py @@ -0,0 +1,279 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Replay-level coverage for the async/await orchestration prototype.""" + +import asyncio +import json +import logging +from collections.abc import AsyncGenerator, Generator +from datetime import datetime, timedelta +from typing import Any + +import pytest + +from durabletask import entities, task, worker +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb +from durabletask.serialization import JsonDataConverter + + +INSTANCE_ID = "coroutine-test" +NOW = datetime(2026, 1, 1) + + +def _start() -> list[pb.HistoryEvent]: + return [ + helpers.new_orchestrator_started_event(NOW), + helpers.new_execution_started_event("test", INSTANCE_ID, json.dumps(3)), + ] + + +def _execute( + fn: task.Orchestrator[Any, Any], + old: list[pb.HistoryEvent], + new: list[pb.HistoryEvent], +) -> worker.ExecutionResults: + registry = worker._Registry() # pyright: ignore[reportPrivateUsage] + registry.add_named_orchestrator("test", fn) + executor = worker._OrchestrationExecutor( # pyright: ignore[reportPrivateUsage] + registry, logging.getLogger(__name__), JsonDataConverter()) + return executor.execute(INSTANCE_ID, old, new) + + +def _completion(result: worker.ExecutionResults) -> pb.CompleteOrchestrationAction: + completions = [a.completeOrchestration for a in result.actions if a.HasField("completeOrchestration")] + assert len(completions) == 1 + return completions[0] + + +def test_coroutine_immediate_return() -> None: + async def orchestrator(ctx: task.OrchestrationContext, value: int) -> int: + return value + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert json.loads(complete.result.value) == 3 + + +def test_await_is_concrete_and_completed_tasks_do_not_suspend() -> None: + assert "__await__" not in task.Task.__abstractmethods__ + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[int]: + completed = task.CompletableTask[int]() + completed.complete(7) + return [await completed, await completed] + + assert json.loads(_completion(_execute(orchestrator, [], _start())).result.value) == [7, 7] + + +def test_precompleted_failure_and_cancellation_are_catchable() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[str]: + failed = task.CompletableTask[int]() + failed.fail("failed", ValueError("boom")) + cancelled = task.TimerTask() + cancelled.cancel() + results: list[str] = [] + try: + await failed + except task.TaskFailedError: + results.append("failed") + try: + await cancelled + except task.TaskCancelledError: + results.append("cancelled") + return results + + assert json.loads(_completion(_execute(orchestrator, [], _start())).result.value) == [ + "failed", "cancelled"] + + +def test_nested_async_helpers_replay_with_generator_action_parity() -> None: + async def helper(ctx: task.OrchestrationContext, value: int) -> int: + return await ctx.call_activity("increment", input=value, return_type=int) + + async def coroutine(ctx: task.OrchestrationContext, value: int) -> int: + first = await helper(ctx, value) + return await helper(ctx, first) + + def generator(ctx: task.OrchestrationContext, value: int) -> Generator[task.Task[Any], Any, int]: + first = yield ctx.call_activity("increment", input=value, return_type=int) + return (yield ctx.call_activity("increment", input=first, return_type=int)) + + batches = [ + _start(), + [helpers.new_task_scheduled_event(1, "increment"), + helpers.new_task_completed_event(1, json.dumps(4))], + [helpers.new_task_scheduled_event(2, "increment"), + helpers.new_task_completed_event(2, json.dumps(5))], + ] + old: list[pb.HistoryEvent] = [] + result: worker.ExecutionResults | None = None + for new in batches: + result = _execute(coroutine, old, new) + assert result.actions == _execute(generator, old, new).actions + old += new + assert result is not None + assert json.loads(_completion(result).result.value) == 5 + + +def test_fan_out_waits_for_all_and_preserves_input_order() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[str]: + pending: list[task.Task[str]] = [ + ctx.call_activity("work", input=i, return_type=str) for i in range(2)] + return await task.when_all(pending) + + old = _start() + assert len(_execute(orchestrator, [], old).actions) == 2 + old += [helpers.new_task_scheduled_event(i, "work") for i in (1, 2)] + second = [helpers.new_task_completed_event(2, '"second"')] + assert _execute(orchestrator, old, second).actions == [] + complete = _completion(_execute( + orchestrator, old + second, [helpers.new_task_completed_event(1, '"first"')])) + assert json.loads(complete.result.value) == ["first", "second"] + + +@pytest.mark.parametrize("caught", [False, True]) +def test_activity_failure_is_thrown_into_coroutine(caught: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + try: + await ctx.call_activity("boom") + except task.TaskFailedError: + if not caught: + raise + return "caught" + return "unexpected" + + old = _start() + [helpers.new_task_scheduled_event(1, "boom")] + complete = _completion(_execute( + orchestrator, old, [helpers.new_task_failed_event(1, ValueError("boom"))])) + assert complete.orchestrationStatus == ( + pb.ORCHESTRATION_STATUS_COMPLETED if caught else pb.ORCHESTRATION_STATUS_FAILED) + if caught: + assert json.loads(complete.result.value) == "caught" + else: + assert complete.failureDetails.innerFailure.errorMessage == "boom" + + +@pytest.mark.parametrize("event_wins", [False, True]) +def test_when_any_timer_and_external_event(event_wins: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + event = ctx.wait_for_external_event("approval") + timer = ctx.create_timer(timedelta(seconds=1)) + winner = await task.when_any([event, timer]) + if winner is event: + timer.cancel() + return await event + return "timeout" + + deadline = NOW + timedelta(seconds=1) + old = _start() + [helpers.new_timer_created_event(1, deadline)] + new = [helpers.new_event_raised_event("approval", '"approved"')] if event_wins else [ + helpers.new_timer_fired_event(1, deadline)] + complete = _completion(_execute(orchestrator, old, new)) + assert json.loads(complete.result.value) == ("approved" if event_wins else "timeout") + + +def test_sub_orchestration_and_replay_flags() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[Any]: + before = ctx.is_replaying + value = await ctx.call_sub_orchestrator("child", instance_id="child-test", return_type=str) + return [value, before, ctx.is_replaying] + + old = _start() + [helpers.new_sub_orchestration_created_event(1, "child", "child-test")] + complete = _completion(_execute( + orchestrator, old, [helpers.new_sub_orchestration_completed_event(1, '"done"')])) + assert json.loads(complete.result.value) == ["done", True, False] + + +def test_retry_timer_resumes_same_coroutine() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + return await ctx.call_activity( + "retry", retry_policy=task.RetryPolicy( + first_retry_interval=timedelta(seconds=1), max_number_of_attempts=2)) + + old = _start() + [helpers.new_task_scheduled_event(1, "retry")] + failure = [helpers.new_task_failed_event(1, ValueError("retry me"))] + retry = _execute(orchestrator, old, failure).actions + assert len(retry) == 1 + assert retry[0].id == 2 + deadline = retry[0].createTimer.fireAt.ToDatetime() + old += failure + [helpers.new_timer_created_event(2, deadline)] + fired = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(2, deadline)] + scheduled = _execute(orchestrator, old, fired).actions + assert len(scheduled) == 1 + assert scheduled[0].id == 1 + assert scheduled[0].scheduleTask.name == "retry" + old += fired + [helpers.new_task_scheduled_event(1, "retry")] + complete = _completion(_execute( + orchestrator, old, [helpers.new_task_completed_event(1, '"success"')])) + assert json.loads(complete.result.value) == "success" + + +def test_entity_lock_and_call_suspend_coroutine() -> None: + entity = entities.EntityInstanceId("counter", "one") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + with await ctx.lock_entities([entity]): + return await ctx.call_entity(entity, "get") + + old = _start() + lock = _execute(orchestrator, [], old).actions[0].sendEntityMessage.entityLockRequested + request = json.dumps({"id": lock.criticalSectionId}) + new = [ + helpers.new_event_sent_event(1, str(entity), request), + helpers.new_event_raised_event(lock.criticalSectionId, None), + ] + actions = _execute(orchestrator, old, new).actions + assert len(actions) == 1 + call = actions[0].sendEntityMessage.entityOperationCalled + assert call.targetInstanceId.value == str(entity) + old += new + [helpers.new_event_sent_event(2, str(entity), json.dumps({"id": call.requestId}))] + complete = _completion(_execute( + orchestrator, old, [helpers.new_event_raised_event(call.requestId, json.dumps({"result": "42"}))])) + assert json.loads(complete.result.value) == 42 + + +def test_continue_as_new_from_coroutine() -> None: + async def orchestrator(ctx: task.OrchestrationContext, value: int) -> None: + await ctx.wait_for_external_event("next") + ctx.continue_as_new(value + 1) + + complete = _completion(_execute( + orchestrator, _start(), [helpers.new_event_raised_event("next", None)])) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + assert json.loads(complete.result.value) == 4 + + +def test_nondeterminism_still_detected_for_coroutines() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + return await ctx.call_activity("changed") + + complete = _completion(_execute( + orchestrator, _start(), [helpers.new_task_scheduled_event(1, "original")])) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType.endswith("NonDeterminismError") + + +@pytest.mark.parametrize("after_activity", [False, True]) +def test_non_durable_suspension_is_rejected(after_activity: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + if after_activity: + await ctx.call_activity("work") + await asyncio.sleep(0) + + old = _start() + [helpers.new_task_scheduled_event(1, "work")] if after_activity else [] + new = [helpers.new_task_completed_event(1, "null")] if after_activity else _start() + complete = _completion(_execute(orchestrator, old, new)) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType == "builtins.TypeError" + assert "must only yield or await durabletask.Task" in complete.failureDetails.errorMessage + + +def test_async_generator_is_rejected() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> AsyncGenerator[task.Task[Any], None]: + yield ctx.call_activity("work", return_type=int) + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "Async generator orchestrators are not supported" in complete.failureDetails.errorMessage From ee356d7d734b284ad063241c3771af7485546fa0 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Fri, 30 Oct 2026 12:43:06 -0600 Subject: [PATCH 02/11] Add deterministic asyncio scheduling and durable timers to shared replay Preserve generator dispatch recovery and snapshots; resume coroutine branches per history completion, order cancellation wakeups, and define the bounded async contract. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 11 +- README.md | 12 +- docs/async-orchestrators.md | 172 +++++++++ durabletask/internal/asyncio_loop.py | 322 ++++++++++++++++ durabletask/task.py | 11 +- durabletask/worker.py | 17 +- .../durabletask/test_asyncio_orchestrators.py | 349 ++++++++++++++++++ .../test_coroutine_orchestrators.py | 6 +- 8 files changed, 884 insertions(+), 16 deletions(-) create mode 100644 docs/async-orchestrators.md create mode 100644 durabletask/internal/asyncio_loop.py create mode 100644 tests/durabletask/test_asyncio_orchestrators.py diff --git a/CHANGELOG.md b/CHANGELOG.md index cfed9a17..657bd191 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,10 +11,13 @@ ADDED - Added `OrchestrationQuery.task_hub_names` to filter orchestration instances by one or more task hubs. -- 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. +- Added native `async def` orchestrators and nested async helpers alongside +existing generator orchestrators. Await durable operations and compose them +with `asyncio.gather`, local tasks, and structured task groups. Async sleeps and +timeouts use durable timers and the orchestration clock; `sleep(0)` yields +locally without a timer. See the [supported contract](docs/async-orchestrators.md) +for cancellation, per-dispatch cleanup, replay/migration rules, and unsupported +APIs. External I/O still belongs in ordinary activities. FIXED diff --git a/README.md b/README.md index efc47a06..5d6f6a7b 100644 --- a/README.md +++ b/README.md @@ -27,9 +27,9 @@ and [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/ ## Optional Features -### Async Orchestrators (Prototype) +### Async Orchestrators -This branch prototypes `async def` orchestrators across all three packages. +All three packages support `async def` orchestrators and nested async helpers. Existing generator orchestrators continue to work. ```python @@ -52,9 +52,9 @@ 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`. +> 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. @@ -64,6 +64,8 @@ An async helper is not a durable sub-orchestration: use 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, framework adapters, and coordinated release rules. ### Large Payload Externalization diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md new file mode 100644 index 00000000..8344d203 --- /dev/null +++ b/docs/async-orchestrators.md @@ -0,0 +1,172 @@ +# Async orchestrators + +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. + +```python +import asyncio + +from durabletask import task + + +async def workflow(ctx: task.OrchestrationContext, values: list[int]) -> list[int]: + async def branch(value: int) -> int: + first = await ctx.call_activity("increment", input=value, return_type=int) + await asyncio.sleep(1) + return await ctx.call_activity("increment", input=first, return_type=int) + + return await asyncio.gather(*(branch(value) for value in values)) +``` + +Each branch resumes as its own durable operation completes, including when +multiple completions arrive in one dispatch. It does not wait for an entire +batch before scheduling its next operation. + +## Supported authoring contract + +Await activities, timers, external events, sub-orchestrations, durable +`task.when_all` / `task.when_any`, and entity calls/locks where supported by the +provider. Functions compatibility operations such as `context.task_all()` +remain awaitable. Generator and coroutine parents/children may call one another +through durable sub-orchestrations. + +The dispatch-local loop supports FIFO ready callbacks, local futures and tasks, +`asyncio.create_task`, `asyncio.gather`, and `asyncio.sleep`. On Python 3.11+, +it supports `asyncio.TaskGroup` and `asyncio.timeout` / `timeout_at`. +`asyncio.wait_for`, AnyIO's asyncio task groups, cancel scopes, shielded cleanup, +and `fail_after` use the same task/future scheduling and durable timers. +AnyIO support is bounded to these paths, not all backends or APIs. + +Create branches from ordered inputs. Ready work follows FIFO scheduling; +history completions are delivered in history order, draining ready work after +each completion. Task-group cancellation wakeups are ordered by local task +creation, not set/object-hash iteration. Local tasks are dispatch resources, +not new durable instances. + +`loop.time()` uses seconds from a fixed UTC epoch and the current orchestration +clock, never machine monotonic time. `loop.call_at` and `call_later` schedule +durable timers for future finite deadlines. Timer conversion rounds up when +necessary so a durable timer does not precede its floating-point deadline. +Core/Functions long timers reuse the existing chunking; Azure Managed retains +its native long timers. Past deadlines are ready callbacks. `sleep(0)` yields +to ready coroutines without a durable timer. Clock advancement and timers +require history events, not elapsed time in the worker. + +> [!WARNING] +> This is not a general-purpose asyncio event loop. External network/storage +> I/O, threads, executors, subprocesses, custom/eager task factories, and +> cross-thread callbacks are outside the orchestrator contract. Perform +> nondeterministic work in ordinary activities. Detected unsupported operations +> raise errors, but the SDK cannot detect every unsafe operation, especially +> synchronous side effects or already-completed foreign awaitables. + +`asyncio.wait` and `as_completed` are **not supported deterministic composition +APIs**. `wait` returns unordered sets; `as_completed` converts inputs to a set +and can create tasks in hash-dependent order. Pre-completed/tied tasks and +differences between Python implementations can change ordering even when a +particular example executes successfully. Do not iterate sets to schedule +durable actions. Use ordered `gather`, task groups, or durable `when_any` and +explicitly handle the selected durable task. Other asyncio APIs are not +implicitly supported because they happen to execute. + +## Failures, timeouts, and cancellation + +Durable failures raise the existing SDK exceptions at the await, preserving +failure details, typed results, and retry policies. `task.when_all` still waits +for every durable child before surfacing failure. In contrast, +`asyncio.gather` with default options reports the first failure without +cancelling siblings. Retain and await siblings if their results matter. +`return_exceptions=True` collects their outcomes in input order. + +Task groups cancel and drain local sibling tasks on failure and use their +library's normal exception grouping rules. AnyIO uses level cancellation: +awaited cleanup in a cancelled scope may require +`with anyio.CancelScope(shield=True)`. Python-version-specific exception +grouping and cancel-count behavior remain those of asyncio/AnyIO. + +Cancelling a local task or timing out its wait detaches that waiter; it does +**not** cancel the durable activity, sub-orchestration, entity operation, or +external-event subscription it was waiting on. Another waiter can still await +the same durable task. Keep the durable task if you intend to await it again. +Use explicit SDK cancellation for cancellable durable timers/event waits. +Cancelling a timer callback removes an uncommitted timer when possible, but +cannot undo a timer already committed to history. Neither local nor durable +cancellation implies external side effects were undone. + +An unhandled local root-coroutine cancellation fails the orchestration with +`TaskCancelledError`; it does not request scheduler termination or repeatedly +abandon the same deterministic replay. Use the client to terminate the durable +instance. Worker/dispatch cancellation and process-control exceptions continue +to follow the existing worker recovery policy. + +## Dispatch teardown is not workflow cancellation + +Every dispatch reconstructs the coroutine and its local tasks from history. +After the response and entity-release obligations are snapshotted, teardown +cancels remaining local tasks, drains cooperative cleanup, closes tracked async +generators, and disposes the loop. This also happens on suspension, termination, +continue-as-new, history failure, and version rejection. + +Nested `finally` blocks and async context-manager exits may run at **every +dispatch boundary**, before workflow completion. Local cleanup can await ready +work such as `sleep(0)`; it must not wait for new durable or external work. +Scheduling/awaiting durable work during async teardown raises an actionable +error. Teardown must cooperate with cancellation; a non-quiescent cleanup +queue is bounded and reported, not allowed to run indefinitely. + +Ordinary teardown failures (including `CancelledError`) are logged without +changing the decided response. Unexpected non-cancellation `BaseException` +exits retain the existing terminal-failure/recovery policy; an existing primary +failure is preserved. Teardown status/actions do not replace the response. +Held entity-release obligations are captured before teardown for terminal +payload/failure handling. + +Keep workflow cleanup that must be durable in **normal orchestration control +flow**, before return or continue-as-new, not solely in a dispatch finalizer. +No cleanup is guaranteed after process crash or forced termination. +Async-generator *helpers* are closed; async-generator *orchestrators* +(`async def` containing `yield`) are rejected. + +## Replay and framework adapters + +Ordinary async helpers rerun during replay. Their return values are not +automatically checkpointed. There are no local/virtual activities and no +automatic recording or interception of arbitrary await expressions. +`is_replaying` is not a substitute for recording results. Activities remain +at-least-once work: side effects require appropriate idempotency/deduplication. + +Pydantic AI's real `Agent.run` / parallel-tool AnyIO path is exercised with a +deterministic `TestModel` in +[`test_pydantic_ai_orchestrator.py`](../tests/durabletask/test_pydantic_ai_orchestrator.py). +The orchestrator directly awaits the agent; no adapter event loop, trampoline, +or per-batch durable join is needed. The proof adapter still registers ordinary +model/tool activities, encodes/decodes framework payloads, supplies replay-stable +IDs and a deadline clock, and explicitly canonicalizes volatile message +timestamps. These are **adapter responsibilities**, not SDK-wide interception. +This proof is not a complete production Pydantic adapter or compatibility +promise for arbitrary agents, model providers, tools, or framework versions. +No paid model calls are needed. + +Changing an in-flight generator to equivalent coroutine code is safe **only +if its durable history contract is preserved**: action IDs/order, names, +payloads, retries, timer deadlines/chunks, event consumption, and entity +release order. Introducing concurrency, a different join, or a new timer may +change that contract. Test recorded histories and version/drain affected +instances; basic sequential parity tests are not blanket migration safety. + +The Functions `orchestrator_generator_wrapper` helper supports direct durable +awaits for 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. + +## Coordinated delivery + +This feature is one draft PR/release unit across all packages. Do not ship a +limited intermediate contract. Release the core package first; then use a +dedicated coordinated release PR to update provider dependency minimums, +versions, and release notes together. Feature development does not change +package versions or dependency minimums. diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py new file mode 100644 index 00000000..0c3d3963 --- /dev/null +++ b/durabletask/internal/asyncio_loop.py @@ -0,0 +1,322 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Dispatch-local asyncio scheduling on top of the shared durable history engine.""" + +from __future__ import annotations + +import asyncio +import contextvars +import math +import logging +import sys +from collections import deque +from collections.abc import Callable, Coroutine, Generator +from datetime import datetime, timedelta +from typing import Any, TypeVar, TYPE_CHECKING, cast + +from durabletask import task + +if TYPE_CHECKING: + from durabletask.worker import _RuntimeOrchestrationContext # pyright: ignore[reportPrivateUsage] + +T = TypeVar("T") + + +class _DurableAsyncTask(asyncio.Task[T]): + def cancel(self, msg: Any = None) -> bool: + loop = self.get_loop() + assert isinstance(loop, DurableEventLoop) + previous = loop.cancelling_task + loop.cancelling_task = self + try: + return super().cancel(msg) + finally: + loop.cancelling_task = previous + + +class _DurableTimerHandle(asyncio.TimerHandle): + def __init__( + self, when: float, callback: Callable[..., Any], args: tuple[Any, ...], + loop: DurableEventLoop, context: contextvars.Context | None, + ) -> None: + super().__init__(when, callback, args, loop, context=context) + self.durable_timer: task.TimerTask | None = None + + def cancel(self) -> None: + super().cancel() + if self.durable_timer is not None: + self.durable_timer.cancel() + + +class DurableEventLoop(asyncio.AbstractEventLoop): + """A FIFO microtask loop; it never polls a selector or advances wall time. + + Only history events can complete durable waits/timers. asyncio owns coroutine + stepping, gather and structured cancellation; this loop owns their ordering. + """ + + def __init__(self, ctx: _RuntimeOrchestrationContext) -> None: + self.ctx = ctx + self._ready: deque[asyncio.Handle] = deque() + self._cancel_ready: list[tuple[int, int, asyncio.Handle]] = [] + self._tasks: dict[asyncio.Task[Any], int] = {} + self.cancelling_task: asyncio.Task[Any] | None = None + self._waits: dict[int, tuple[task.Task[Any], list[asyncio.Future[Any]]]] = {} + self._timers: list[_DurableTimerHandle] = [] + self._asyncgens: dict[int, Any] = {} + self._closed = False + self._closing = False + self._fatal: BaseException | None = None + self._root: asyncio.Task[Any] | None = None + + def time(self) -> float: + # A fixed UTC epoch avoids local-time and platform timestamp differences. + return (self.ctx.current_utc_datetime - datetime(1970, 1, 1)).total_seconds() + + def get_debug(self) -> bool: + return False + + def is_running(self) -> bool: + return asyncio._get_running_loop() is self + + def is_closed(self) -> bool: + return self._closed + + def create_future(self) -> asyncio.Future[Any]: + return asyncio.Future(loop=self) + + def get_task_factory(self) -> None: + return None + + def create_task( + self, coro: Coroutine[Any, Any, T] | Generator[Any, None, T], *, name: str | None = None, + context: contextvars.Context | None = None, **kwargs: Any, + ) -> asyncio.Task[T]: + if self._closed: + coro.close() + raise RuntimeError("The durable orchestration dispatch is closed.") + if kwargs: + coro.close() + raise TypeError("Custom/eager task execution is not supported in durable orchestrators.") + if sys.version_info >= (3, 11): + future = _DurableAsyncTask(coro, loop=self, name=name, context=context) + else: + if context is not None: + coro.close() + raise TypeError("Task contexts require Python 3.11 or later.") + future = _DurableAsyncTask(coro, loop=self, name=name) + self._tasks[future] = len(self._tasks) + return future + + def call_soon( + self, callback: Callable[..., Any], *args: Any, + context: contextvars.Context | None = None, + ) -> asyncio.Handle: + handle = asyncio.Handle(callback, args, self, context=context) + if self.cancelling_task is not None: + # TaskGroup and AnyIO cancel sets of tasks. Do not let object hashes + # determine the order in which their finally blocks execute. + owner = getattr(callback, "__self__", None) + index = self._tasks[self.cancelling_task] + if isinstance(owner, asyncio.Task): + index = self._tasks.get(cast(asyncio.Task[Any], owner), index) + self._cancel_ready.append((index, len(self._cancel_ready), handle)) + else: + self._ready.append(handle) + return handle + + def call_later( + self, delay: float, callback: Callable[..., Any], *args: Any, + context: contextvars.Context | None = None, + ) -> asyncio.TimerHandle: + if not math.isfinite(delay): + raise ValueError("Durable delays must be finite.") + return self.call_at(self.time() + max(0, delay), callback, *args, context=context) + + def call_at( + self, when: float, callback: Callable[..., Any], *args: Any, + context: contextvars.Context | None = None, + ) -> asyncio.TimerHandle: + if not math.isfinite(when): + raise ValueError("Durable deadlines must be finite.") + handle = _DurableTimerHandle(when, callback, args, self, context) + if when <= self.time(): + self._ready.append(handle) + else: + if self._closing: + raise RuntimeError("Dispatch cleanup cannot schedule durable timers.") + deadline = datetime(1970, 1, 1) + timedelta(seconds=when) + if (deadline - datetime(1970, 1, 1)).total_seconds() < when: + deadline += timedelta(microseconds=1) + handle.durable_timer = self.ctx.create_timer(deadline) + self._timers.append(handle) + return handle + + def _timer_handle_cancelled(self, handle: asyncio.TimerHandle) -> None: + # No selector heap to maintain; _DurableTimerHandle owns cancellation. + pass + + async def wait_durable(self, durable: task.Task[T]) -> T: + if self._closing: + raise RuntimeError("Dispatch cleanup cannot await durable work.") + if durable.is_complete: + return durable.get_result() + future = self.create_future() + entry = self._waits.setdefault(id(durable), (durable, [])) + entry[1].append(future) + try: + return await future + finally: + # Cancelling one waiter never cancels a shared durable operation. + entry[1].remove(future) + if not entry[1]: + self._waits.pop(id(durable), None) + + def _deliver_completions(self) -> None: + for durable, futures in tuple(self._waits.values()): + if durable.is_complete: + for future in tuple(futures): + if not future.done(): + try: + future.set_result(durable.get_result()) + except Exception as error: + future.set_exception(error) + for handle in tuple(self._timers): + if handle.cancelled(): + self._timers.remove(handle) + elif handle.durable_timer is not None and handle.durable_timer.is_complete: + self._timers.remove(handle) + self._ready.append(handle) + + def start(self, coro: Coroutine[Any, Any, Any]) -> None: + self._root = self.create_task(coro, name="durable-orchestrator") + self.resume() + + def resume(self) -> None: + if self.ctx._is_complete: # pyright: ignore[reportPrivateUsage] + return + self._deliver_completions() + self._drain() + assert self._root is not None + if self._root.done(): + if self._root.cancelled(): + raise task.TaskCancelledError( + "The orchestrator coroutine was cancelled. Catch local cancellation " + "inside the orchestrator; terminate the durable instance through the client.") + # result() deliberately propagates control-flow BaseExceptions to the + # existing executor recovery boundary, not a success-shaped fallback. + result = self._root.result() + raise StopIteration(result) + if not self._has_durable_work(): + raise RuntimeError( + "The async orchestrator is blocked without durable work. " + "Await durable activities, events or timers; external I/O is unsupported.") + + def _has_durable_work(self) -> bool: + return any(not durable.is_complete for durable, _ in self._waits.values()) or any( + not handle.cancelled() for handle in self._timers) + + def _drain(self) -> None: + previous_loop = asyncio._get_running_loop() + previous_task = asyncio.current_task(previous_loop) if previous_loop is not None else None + previous_hooks = sys.get_asyncgen_hooks() + if previous_task is not None and previous_loop is not None: + asyncio.tasks._leave_task(previous_loop, previous_task) + asyncio._set_running_loop(self) + sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter, finalizer=self._asyncgen_finalizer) + try: + steps = 0 + while self._ready or self._cancel_ready: + self._ready.extend(handle for _, _, handle in sorted(self._cancel_ready)) + self._cancel_ready.clear() + handle = self._ready.popleft() + if not handle.cancelled(): + handle._run() + if self._fatal is not None: + error, self._fatal = self._fatal, None + raise error + if not self._closing and self.ctx._is_complete: # pyright: ignore[reportPrivateUsage] + break + if not self._closing and self._root is not None and self._root.done(): + break + if not self._closing: + self._deliver_completions() + steps += 1 + if self._closing and steps >= 10000: + raise RuntimeError("Async cleanup did not quiesce; cleanup must cooperate with cancellation.") + finally: + sys.set_asyncgen_hooks(*previous_hooks) + asyncio._set_running_loop(previous_loop) + if previous_task is not None and previous_loop is not None: + asyncio.tasks._enter_task(previous_loop, previous_task) + + def _asyncgen_firstiter(self, generator: Any) -> None: + self._asyncgens[id(generator)] = generator + + def _asyncgen_finalizer(self, generator: Any) -> None: + if not self._closed: + self.create_task(generator.aclose()) + + def call_exception_handler(self, context: dict[str, Any]) -> None: + error = context.get("exception") + self._fatal = error if isinstance(error, BaseException) else RuntimeError(context["message"]) + + def close(self) -> None: + self._closing = True + self.ctx._async_cleanup = True # pyright: ignore[reportPrivateUsage] + try: + pending_at_close = {future for future in self._tasks if not future.done()} + for future in tuple(self._tasks): + if not future.done(): + future.cancel() + self._drain() + for generator in tuple(self._asyncgens.values()): + self.create_task(generator.aclose()) + self._drain() + for future in tuple(self._tasks): + if not future.done(): + future.cancel() + self._drain() + for future in tuple(self._tasks): + if not future.done(): + future.get_coro().close() + raise RuntimeError("Async cleanup must finish without waiting for durable or external work.") + if not future.cancelled(): + error = future.exception() + if error is not None and future in pending_at_close: + if not isinstance(error, Exception): + raise error + logger = self.ctx._logger or logging.getLogger(__name__) # pyright: ignore[reportPrivateUsage] + logger.warning( + "%s: Async task cleanup failed.", self.ctx.instance_id, exc_info=error) + finally: + self._closed = True + self._ready.clear() + self._cancel_ready.clear() + self._waits.clear() + self._timers.clear() + self._tasks.clear() + self._asyncgens.clear() + + def run_in_executor(self, executor: Any, func: Any, *args: Any) -> asyncio.Future[Any]: + raise RuntimeError("Threads/executors are unsupported in orchestrators; use an activity.") + + def call_soon_threadsafe(self, callback: Any, *args: Any, context: Any = None) -> asyncio.Handle: + raise RuntimeError("Cross-thread scheduling is unsupported in orchestrators; use an activity.") + + async def _unsupported_io(self, *args: Any, **kwargs: Any) -> Any: + raise RuntimeError("External I/O is unsupported in orchestrators; move it to an ordinary activity.") + + create_connection = _unsupported_io + create_server = _unsupported_io + create_datagram_endpoint = _unsupported_io + connect_read_pipe = _unsupported_io + connect_write_pipe = _unsupported_io + getaddrinfo = _unsupported_io + getnameinfo = _unsupported_io + sock_connect = _unsupported_io + sock_recv = _unsupported_io + sock_sendall = _unsupported_io + subprocess_exec = _unsupported_io + subprocess_shell = _unsupported_io diff --git a/durabletask/task.py b/durabletask/task.py index 278527e6..2c123b94 100644 --- a/durabletask/task.py +++ b/durabletask/task.py @@ -6,6 +6,7 @@ import logging import math +import asyncio from abc import ABC, abstractmethod from collections.abc import Callable, Coroutine, Generator, Sequence from dataclasses import dataclass @@ -634,9 +635,15 @@ def __init__(self) -> None: 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. + A local asyncio waiter does not own or cancel the durable operation. """ + from durabletask.internal.asyncio_loop import DurableEventLoop + + loop = asyncio._get_running_loop() + if isinstance(loop, DurableEventLoop): + return (yield from loop.wait_durable(self).__await__()) + if loop is not None: + raise RuntimeError("Durable tasks can only be awaited inside a durable orchestrator.") if not self.is_complete: yield self return self.get_result() diff --git a/durabletask/worker.py b/durabletask/worker.py index a15e93dd..613bea6b 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -51,6 +51,7 @@ from durabletask.payload import helpers as payload_helpers from durabletask import task from durabletask.internal.grpc_interceptor import DefaultClientInterceptorImpl +from durabletask.internal.asyncio_loop import DurableEventLoop from durabletask.payload.store import PayloadStore from durabletask.serialization import DataConverter, JsonDataConverter @@ -1627,8 +1628,14 @@ def __init__(self, self._data_converter = data_converter self._exception_properties_provider = exception_properties_provider self._logger = logger + self._async_loop: DurableEventLoop | None = None + self._async_cleanup = False def run(self, generator: Generator[task.Task[Any], Any, Any] | Coroutine[Any, Any, Any]) -> None: + if isinstance(generator, CoroutineType): + self._async_loop = DurableEventLoop(self) + self._async_loop.start(generator) + return self._generator = generator self._set_next_task(generator.send(None)) @@ -1641,6 +1648,9 @@ def _set_next_task(self, next_task: Any) -> None: self._previous_task = next_task def resume(self) -> None: + if self._async_loop is not None: + self._async_loop.resume() + return if self._generator is None: # This is never expected unless maybe there's an issue with the history raise TypeError( @@ -1791,6 +1801,8 @@ def get_actions(self) -> list[pb.OrchestratorAction]: return current_actions def next_sequence_number(self) -> int: + if self._async_cleanup: + raise RuntimeError("Dispatch cleanup cannot schedule durable work; use normal orchestrator control flow.") self._sequence_number += 1 return self._sequence_number @@ -2485,9 +2497,10 @@ def execute( finally: # Close after building the response, so the orchestrator's finally # blocks do not change the actions or custom status we return. - if ctx._generator is not None: # pyright: ignore[reportPrivateUsage] + closeable = ctx._async_loop or ctx._generator # pyright: ignore[reportPrivateUsage] + if closeable is not None: try: - ctx._generator.close() # pyright: ignore[reportPrivateUsage] + closeable.close() except (Exception, asyncio.CancelledError): cleanup_name = next( (e.executionStarted.name for e in new_events if e.HasField("executionStarted")), diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py new file mode 100644 index 00000000..6f72eb53 --- /dev/null +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -0,0 +1,349 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Deterministic asyncio and AnyIO behavior through the real history executor.""" + +import asyncio +import gc +import json +import sys +from contextlib import asynccontextmanager +from datetime import timedelta +from typing import Any +from weakref import ref + +import anyio +import pytest + +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb +from tests.durabletask.test_coroutine_orchestrators import _completion, _execute, _start, NOW + + +def test_gather_resumes_each_branch_without_waiting_for_batch() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[Any]: + async def branch(name: str) -> Any: + value = await ctx.call_activity(name) + await asyncio.sleep(0) + return await ctx.call_activity(name + "-next", input=value) + + return await asyncio.gather(branch("a"), branch("b")) + + start = _start() + actions = _execute(orchestrator, [], start).actions + assert [a.scheduleTask.name for a in actions] == ["a", "b"] + old = start + [helpers.new_task_scheduled_event(i, name) for i, name in [(1, "a"), (2, "b")]] + second = [helpers.new_task_completed_event(2, '"b"')] + actions = _execute(orchestrator, old, second).actions + assert [(a.id, a.scheduleTask.name) for a in actions] == [(3, "b-next")] + third = [helpers.new_task_scheduled_event(3, "b-next"), + helpers.new_task_completed_event(1, '"a"'), + helpers.new_task_scheduled_event(4, "a-next"), + helpers.new_task_completed_event(4, '"a-done"'), + helpers.new_task_completed_event(3, '"b-done"')] + assert json.loads(_completion(_execute(orchestrator, old + second, third)).result.value) == [ + "a-done", "b-done"] + + +@pytest.mark.parametrize("return_exceptions", [False, True]) +def test_gather_failure_does_not_cancel_sibling(return_exceptions: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + async def sibling() -> str: + await ctx.call_activity("slow") + return await ctx.call_activity("continued", return_type=str) + + pending = asyncio.create_task(sibling()) + try: + values = await asyncio.gather(ctx.call_activity("fail"), pending, return_exceptions=return_exceptions) + return [isinstance(values[0], task.TaskFailedError), values[1]] + except task.TaskFailedError: + assert not pending.cancelled() + return await pending + + old = _start() + [helpers.new_task_scheduled_event(1, "fail"), + helpers.new_task_scheduled_event(2, "slow")] + failed = [helpers.new_task_failed_event(1, ValueError("boom"))] + assert _execute(orchestrator, old, failed).actions == [] + completed = [helpers.new_task_completed_event(2, "null")] + assert _execute(orchestrator, old + failed, completed).actions[0].scheduleTask.name == "continued" + result = _completion(_execute(orchestrator, old + failed + completed, [ + helpers.new_task_scheduled_event(3, "continued"), helpers.new_task_completed_event(3, '"ok"')])) + assert json.loads(result.result.value) == ([True, "ok"] if return_exceptions else "ok") + + +@pytest.mark.parametrize("native", [False, True]) +def test_task_group_failure_cancels_waiters_not_durable_work(native: bool) -> None: + if native and sys.version_info < (3, 11): + pytest.skip("asyncio.TaskGroup requires Python 3.11") + cleaned: list[str] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def branch(name: str) -> None: + try: + await ctx.call_activity(name) + finally: + with anyio.CancelScope(shield=True): + await asyncio.sleep(0) + cleaned.append(name) + + if native: + async with asyncio.TaskGroup() as group: + group.create_task(branch("fail")) + group.create_task(branch("slow")) + else: + async with anyio.create_task_group() as group: + group.start_soon(branch, "fail") + group.start_soon(branch, "slow") + + old = _start() + [helpers.new_task_scheduled_event(1, "fail"), + helpers.new_task_scheduled_event(2, "slow")] + result = _completion(_execute(orchestrator, old, [helpers.new_task_failed_event(1, ValueError("boom"))])) + assert result.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert cleaned == ["fail", "slow"] + + +@pytest.mark.parametrize("kind", ["sleep", "wait_for", "anyio", "timeout"]) +def test_durable_timer_and_timeout(kind: str) -> None: + if kind == "timeout" and sys.version_info < (3, 11): + pytest.skip("asyncio.timeout requires Python 3.11") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + try: + if kind == "sleep": + await asyncio.sleep(2) + elif kind == "wait_for": + await asyncio.wait_for(ctx.wait_for_external_event("never"), timeout=2) + elif kind == "timeout": + async with asyncio.timeout(2): + await ctx.wait_for_external_event("never") + else: + with anyio.fail_after(2): + await ctx.wait_for_external_event("never") + except TimeoutError: + return "timed out" + return "slept" + + actions = _execute(orchestrator, [], _start()).actions + assert len(actions) == 1 + deadline = actions[0].createTimer.fireAt.ToDatetime() + assert deadline == NOW + timedelta(seconds=2) + old = _start() + [helpers.new_timer_created_event(1, deadline)] + new = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(1, deadline)] + assert json.loads(_completion(_execute(orchestrator, old, new)).result.value) == ( + "slept" if kind == "sleep" else "timed out") + + +def test_sleep_zero_fifo_without_timer() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[int]: + observed: list[int] = [] + + async def branch(index: int) -> None: + observed.append(index) + await asyncio.sleep(0) + observed.append(index + 2) + + await asyncio.gather(branch(0), branch(1)) + return observed + + result = _execute(orchestrator, [], _start()) + assert len(result.actions) == 1 + assert json.loads(_completion(result).result.value) == [0, 1, 2, 3] + + +@pytest.mark.parametrize("under_loop", [False, True]) +def test_dispatch_cleanup_unwinds_scopes_contexts_asyncgens(under_loop: bool) -> None: + cleaned: list[str] = [] + references: list[Any] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + references.extend([ref(ctx), ref(asyncio.get_running_loop()), ref(asyncio.current_task())]) + + async def resource(): + try: + yield 1 + finally: + await asyncio.sleep(0) + cleaned.append("asyncgen") + + opened = resource() + await anext(opened) + try: + async with anyio.create_task_group(): + await ctx.call_activity("wait") + finally: + await asyncio.sleep(0) + cleaned.append("root") + + def dispatch() -> None: + result = _execute(orchestrator, [], _start()) + assert [a.scheduleTask.name for a in result.actions] == ["wait"] + assert cleaned == ["root", "asyncgen"] + + async def outer() -> None: + before = asyncio.current_task() + dispatch() + assert asyncio.current_task() is before + await asyncio.sleep(0) + + if under_loop: + asyncio.run(outer()) + else: + dispatch() + gc.collect() + assert all(reference() is None for reference in references) + + +def test_shared_operation_survives_local_waiter_cancellation() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + durable = ctx.call_activity("shared") + + async def wait() -> Any: + return await durable + + first = asyncio.create_task(wait()) + second = asyncio.create_task(wait()) + await asyncio.sleep(0) + first.cancel() + await asyncio.gather(first, return_exceptions=True) + return await second + + old = _start() + [helpers.new_task_scheduled_event(1, "shared")] + assert json.loads(_completion(_execute(orchestrator, old, [ + helpers.new_task_completed_event(1, '"ok"')])).result.value) == "ok" + + +def test_local_durable_cancellation_wakes_waiter_without_history() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + timer = ctx.create_timer(timedelta(days=1)) + + async def waiter() -> None: + await timer + + pending = asyncio.create_task(waiter()) + await asyncio.sleep(0) + timer.cancel() + try: + await pending + except task.TaskCancelledError: + return "cancelled" + return "unexpected" + + result = _execute(orchestrator, [], _start()) + assert len(result.actions) == 1 + assert json.loads(_completion(result).result.value) == "cancelled" + + +@pytest.mark.parametrize("terminal", ["suspend", "terminate", "continue", "history-failure"]) +def test_async_teardown_preserves_response_and_cannot_schedule_work( + terminal: str, caplog: pytest.LogCaptureFixture) -> None: + cleaned: list[str] = [] + + @asynccontextmanager + async def resource(): + try: + yield + finally: + await asyncio.sleep(0) + cleaned.append("resource") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async with resource(): + try: + ctx.set_custom_status("waiting") + if terminal == "continue": + ctx.continue_as_new("next") + await ctx.call_activity("wait") + finally: + try: + await asyncio.sleep(0) + cleaned.append("inner") + ctx.call_activity("forbidden-cleanup") + finally: + await asyncio.sleep(0) + cleaned.append("outer") + + old = _start() + if terminal == "terminate": + new = [pb.HistoryEvent(executionTerminated=pb.ExecutionTerminatedEvent())] + elif terminal == "history-failure": + new = [helpers.new_task_scheduled_event(1, "wrong")] + else: + old, new = [], old + result = _execute(orchestrator, old, new) + assert cleaned == ["inner", "outer", "resource"] + assert result.encoded_custom_status == '"waiting"' + assert all(action.scheduleTask.name != "forbidden-cleanup" for action in result.actions) + if terminal == "terminate": + assert _completion(result).orchestrationStatus == pb.ORCHESTRATION_STATUS_TERMINATED + elif terminal == "history-failure": + assert _completion(result).failureDetails.errorType.endswith("NonDeterminismError") + elif terminal == "continue": + assert _completion(result).orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + else: + assert [action.scheduleTask.name for action in result.actions] == ["wait"] + assert "Dispatch cleanup cannot schedule durable work" in caplog.text + + +def test_unhandled_local_root_cancellation_fails_not_abandons() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + raise asyncio.CancelledError + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType.endswith("TaskCancelledError") + + +@pytest.mark.parametrize("delay", [0.0000003, 0.000001, 0.0000013, 2.25]) +def test_timer_rounding_never_precedes_loop_deadline(delay: float) -> None: + observed: list[float] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + loop = asyncio.get_running_loop() + observed.append(loop.time() + delay) + await asyncio.sleep(delay) + + actions = _execute(orchestrator, [], _start()).actions + assert len(actions) == 1 + deadline = actions[0].createTimer.fireAt.ToDatetime() + from datetime import datetime + assert (deadline - datetime(1970, 1, 1)).total_seconds() >= observed[0] + + +def test_async_sleep_uses_shared_long_timer_chunking() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + await asyncio.sleep(7 * 86400) + return "done" + + old = _start() + for timer_id, days in [(1, 3), (2, 6), (3, 7)]: + result = _execute(orchestrator, [], old) + assert len(result.actions) == 1 + action = result.actions[0] + assert action.id == timer_id + deadline = NOW + timedelta(days=days) + assert action.createTimer.fireAt.ToDatetime() == deadline + old += [helpers.new_timer_created_event(timer_id, deadline), + helpers.new_orchestrator_started_event(deadline), + helpers.new_timer_fired_event(timer_id, deadline)] + assert json.loads(_completion(_execute(orchestrator, old[:-1], old[-1:])).result.value) == "done" + + +@pytest.mark.parametrize("api", ["thread", "io", "foreign-loop"]) +def test_unsupported_work_fails_actionably(api: str) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + if api == "thread": + await asyncio.to_thread(lambda: None) + elif api == "io": + await asyncio.open_connection("localhost", 1) + else: + foreign = asyncio.new_event_loop() + try: + await foreign.create_future() + finally: + foreign.close() + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert any(text in complete.failureDetails.errorMessage for text in ( + "use an activity", "ordinary activity", "different loop")) diff --git a/tests/durabletask/test_coroutine_orchestrators.py b/tests/durabletask/test_coroutine_orchestrators.py index 2eaef122..be9af370 100644 --- a/tests/durabletask/test_coroutine_orchestrators.py +++ b/tests/durabletask/test_coroutine_orchestrators.py @@ -260,14 +260,14 @@ def test_non_durable_suspension_is_rejected(after_activity: bool) -> None: async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: if after_activity: await ctx.call_activity("work") - await asyncio.sleep(0) + await asyncio.Future() old = _start() + [helpers.new_task_scheduled_event(1, "work")] if after_activity else [] new = [helpers.new_task_completed_event(1, "null")] if after_activity else _start() complete = _completion(_execute(orchestrator, old, new)) assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED - assert complete.failureDetails.errorType == "builtins.TypeError" - assert "must only yield or await durabletask.Task" in complete.failureDetails.errorMessage + assert complete.failureDetails.errorType == "builtins.RuntimeError" + assert "blocked without durable work" in complete.failureDetails.errorMessage def test_async_generator_is_rejected() -> None: From 40b30ac13fffe05923172b6451712b7a7c3a6155 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Fri, 30 Oct 2026 12:47:05 -0600 Subject: [PATCH 03/11] Preserve exceptional-exit policy for async-generator teardown Handle closure tasks created during cleanup, retaining warnings for ordinary errors and terminal recovery for unexpected BaseException exits. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- durabletask/internal/asyncio_loop.py | 4 +-- .../durabletask/test_asyncio_orchestrators.py | 36 +++++++++++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index 0c3d3963..6f144953 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -266,7 +266,7 @@ def close(self) -> None: self._closing = True self.ctx._async_cleanup = True # pyright: ignore[reportPrivateUsage] try: - pending_at_close = {future for future in self._tasks if not future.done()} + completed_at_close = {future for future in self._tasks if future.done()} for future in tuple(self._tasks): if not future.done(): future.cancel() @@ -284,7 +284,7 @@ def close(self) -> None: raise RuntimeError("Async cleanup must finish without waiting for durable or external work.") if not future.cancelled(): error = future.exception() - if error is not None and future in pending_at_close: + if error is not None and future not in completed_at_close: if not isinstance(error, Exception): raise error logger = self.ctx._logger or logging.getLogger(__name__) # pyright: ignore[reportPrivateUsage] diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 6f72eb53..25984de1 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -347,3 +347,39 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED assert any(text in complete.failureDetails.errorMessage for text in ( "use an activity", "ordinary activity", "different loop")) + + +@pytest.mark.parametrize("async_generator", [False, True]) +@pytest.mark.parametrize("unexpected", [False, True]) +def test_cleanup_failure_policy_includes_async_generator_tasks( + async_generator: bool, unexpected: bool, caplog: pytest.LogCaptureFixture) -> None: + class CleanupAbort(BaseException): + pass + + error = CleanupAbort("cleanup failed") if unexpected else ValueError("cleanup failed") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def resource(): + try: + yield 1 + finally: + await asyncio.sleep(0) + raise error + + opened = resource() + await anext(opened) + try: + await ctx.call_activity("wait") + finally: + if not async_generator: + await opened.aclose() + + result = _execute(orchestrator, [], _start()) + if unexpected: + complete = _completion(result) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorMessage == "cleanup failed" + assert complete.failureDetails.errorType.endswith("CleanupAbort") + else: + assert [action.scheduleTask.name for action in result.actions] == ["wait"] + assert "cleanup failed" in caplog.text From 8c50027fe627b1ea257db71e7515894f4bfde819 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Fri, 30 Oct 2026 12:47:07 -0600 Subject: [PATCH 04/11] Prove native Pydantic AnyIO execution and cross-provider asyncio paths Exercise TestModel parallel tools and recorded failures without a trampoline, retain adapter serialization and replay-stable metadata, and cover Windows Python versions in CI. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/workflows/durabletask.yml | 23 +++ requirements.txt | 1 + .../e2e/apps/dtask_style/orchestrators.py | 6 +- .../e2e/apps/v1_style/orchestrators.py | 6 +- .../test_functions_coroutine_orchestrators.py | 9 +- .../test_dts_coroutine_orchestrators.py | 9 +- .../test_pydantic_ai_orchestrator.py | 131 ++++++++++++++++++ 7 files changed, 173 insertions(+), 12 deletions(-) create mode 100644 tests/durabletask/test_pydantic_ai_orchestrator.py diff --git a/.github/workflows/durabletask.yml b/.github/workflows/durabletask.yml index 5da31933..2720e25a 100644 --- a/.github/workflows/durabletask.yml +++ b/.github/workflows/durabletask.yml @@ -14,6 +14,29 @@ permissions: contents: read jobs: + async-runtime-windows: + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15"] + runs-on: windows-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: ${{ matrix.python-version }} + - name: Install runtime test dependencies + run: python -m pip install -r requirements.txt -e . + - name: Check asyncio, AnyIO, replay, and dispatch cleanup + run: >- + python -m pytest + tests\durabletask\test_asyncio_orchestrators.py + tests\durabletask\test_coroutine_orchestrators.py + tests\durabletask\test_pydantic_ai_orchestrator.py + tests\durabletask\test_orchestration_generator_cleanup.py + tests\durabletask\test_orchestration_dispatch_recovery.py + -q + lint-and-unit-tests: runs-on: ubuntu-latest steps: diff --git a/requirements.txt b/requirements.txt index 98fac019..2d865693 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,6 +4,7 @@ protobuf pytest pytest-asyncio anyio +pydantic-ai-slim==2.60.0; python_version >= "3.11" # Framework proof, no model providers pytest-cov azure-identity azure-functions diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py index 347bce46..ccb003c3 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py @@ -13,6 +13,7 @@ """ from datetime import timedelta +import asyncio import logging from typing import Any @@ -38,8 +39,9 @@ def activity_chain(ctx: task.OrchestrationContext, _: Any): @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)) + values = await asyncio.gather(*(ctx.call_activity("square", input=i, return_type=int) for i in (2, 3))) + await asyncio.sleep(0) + await asyncio.sleep(0.01) child: list[str] = await ctx.call_sub_orchestrator("activity_chain") return {"greeting": greeting, "total": sum(values), "child": child} diff --git a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py index d4766d87..28d70500 100644 --- a/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/v1_style/orchestrators.py @@ -13,6 +13,7 @@ """ from datetime import timedelta +import asyncio from typing import Any import azure.durable_functions as df @@ -35,8 +36,9 @@ def activity_chain(context: df.DurableOrchestrationContext): @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)) + values = await asyncio.gather(*(context.call_activity("square", i) for i in (2, 3))) + await asyncio.sleep(0) + await asyncio.sleep(0.01) child = await context.call_sub_orchestrator("activity_chain") return {"greeting": greeting, "total": sum(values), "child": child} diff --git a/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py index f264499c..cdb157e0 100644 --- a/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py +++ b/tests/azure-functions-durable/test_functions_coroutine_orchestrators.py @@ -4,6 +4,7 @@ """Coroutine replay through native and compatibility Functions entry points.""" import base64 +import asyncio import json from collections.abc import Callable from typing import Any @@ -23,17 +24,17 @@ @pytest.mark.parametrize("entry", ["worker-sync", "worker-async", "decorator", "v1-handle"]) async def test_coroutine_replay_through_functions_entry_points(native: bool, entry: str) -> None: async def modern(ctx: task.OrchestrationContext, value: int) -> list[Any]: - return await task.when_all([ + return await asyncio.gather( ctx.call_activity("work", input=value, return_type=str), ctx.call_activity("work", input=value + 1, return_type=str), - ]) + ) async def compatible(ctx: df.DurableOrchestrationContext) -> list[Any]: value = ctx.get_input() - return await ctx.task_all([ + return await asyncio.gather( ctx.call_activity("work", value), ctx.call_activity("work", value + 1), - ]) + ) fn: Callable[..., Any] = modern if native else compatible worker = DurableFunctionsWorker() diff --git a/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py index 42fd4e3a..2ac8ab01 100644 --- a/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py +++ b/tests/durabletask-azuremanaged/test_dts_coroutine_orchestrators.py @@ -4,10 +4,10 @@ """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 @@ -29,17 +29,18 @@ def increment(ctx: task.ActivityContext, value: int) -> int: return value + 1 async def coroutine_child(ctx: task.OrchestrationContext, value: int) -> int: - await ctx.create_timer(timedelta(milliseconds=10)) + await asyncio.sleep(0) + await asyncio.sleep(0.01) return await ctx.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.OrchestrationContext, value: int) -> list[Any]: - values = await task.when_all([ + values = await asyncio.gather( ctx.call_sub_orchestrator(coroutine_child, input=value), ctx.call_sub_orchestrator(generator_child, input=value + 1), - ]) + ) event = await ctx.wait_for_external_event("go") return values + [event] diff --git a/tests/durabletask/test_pydantic_ai_orchestrator.py b/tests/durabletask/test_pydantic_ai_orchestrator.py new file mode 100644 index 00000000..fa856410 --- /dev/null +++ b/tests/durabletask/test_pydantic_ai_orchestrator.py @@ -0,0 +1,131 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Real Pydantic AI parallel tools with ordinary activities and no trampoline. + +This small proof adapter retains model/tool registration and JSON codecs. It is +not a replacement for the framework adapter's full serialization/error policy. +""" + +import asyncio +import json +from contextvars import ContextVar +from datetime import timezone +from typing import Any + +import pytest + +from durabletask import task +from durabletask.internal import helpers +from durabletask.internal import orchestrator_service_pb2 as pb +from tests.durabletask.test_coroutine_orchestrators import _completion, _execute, _start + + +@pytest.mark.parametrize("fail_tool", [False, True]) +def test_pydantic_agent_parallel_tools_replay_without_bridge(fail_tool: bool) -> None: + pytest.importorskip("pydantic_ai", reason="The framework requires Python 3.11+") + from pydantic import TypeAdapter + from pydantic_ai import Agent + from pydantic_ai.capabilities import AbstractCapability + from pydantic_ai.messages import ModelMessage, ModelResponse + from pydantic_ai.models import ModelRequestParameters + from pydantic_ai.models.test import TestModel + from pydantic_ai.settings import ModelSettings + + context: ContextVar[task.OrchestrationContext] = ContextVar("context") + request_codec = TypeAdapter(tuple[list[ModelMessage], ModelSettings | None, ModelRequestParameters]) + response_codec = TypeAdapter(ModelResponse) + invoked: list[str] = [] + + class DurableModel(TestModel): + async def request(self, messages: Any, model_settings: Any, model_request_parameters: Any) -> ModelResponse: + payload = request_codec.dump_python((messages, model_settings, model_request_parameters), mode="json") + # This proof adapter explicitly canonicalizes framework message + # metadata; the SDK must not intercept wall-clock calls globally. + stamp = context.get().current_utc_datetime.replace(tzinfo=timezone.utc).isoformat() + for message in payload[0]: + message["timestamp"] = stamp + for part in message["parts"]: + if "timestamp" in part: + part["timestamp"] = stamp + result = await context.get().call_activity("model", input=payload) + return response_codec.validate_python(result) + + async def model(payload: Any) -> Any: + messages, settings, parameters = request_codec.validate_python(payload) + result = await TestModel(custom_output_text="done").request(messages, settings, parameters) + return response_codec.dump_python(result, mode="json") + + async def first_activity(value: int) -> str: + if fail_tool: + raise RuntimeError("recorded tool failure") + return f"first:{value}" + + async def second_activity(value: int) -> str: + return f"second:{value}" + + activities = {"model": model, "first": first_activity, "second": second_activity} + + class ReplayClock(AbstractCapability[None]): + def _run_clock(self) -> Any: + return lambda: context.get().current_utc_datetime.replace(tzinfo=timezone.utc) + + agent = Agent(DurableModel(), name="proof", capabilities=[ReplayClock()]) + + @agent.tool_plain + async def first(value: int) -> str: + return await context.get().call_activity("first", input=value, return_type=str) + + @agent.tool_plain + async def second(value: int) -> str: + return await context.get().call_activity("second", input=value, return_type=str) + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + token = context.set(ctx) + try: + result = await agent.run( + "Call both tools.", run_id=str(ctx.new_uuid()), conversation_id=ctx.instance_id) + return result.output + finally: + context.reset(token) + + history = _start() + result = _execute(orchestrator, [], history) + batches: list[list[str]] = [] + while not any(action.HasField("completeOrchestration") for action in result.actions): + assert result.actions + scheduled = list(result.actions) + batches.append([action.scheduleTask.name for action in scheduled]) + events = [] + for action in scheduled: + assert action.HasField("scheduleTask") + name = action.scheduleTask.name + history.append(helpers.new_task_scheduled_event(action.id, name)) + payload = json.loads(action.scheduleTask.input.value) + invoked.append(name) + try: + output = asyncio.run(activities[name](payload)) + except RuntimeError as error: + events.append(helpers.new_task_failed_event(action.id, error)) + else: + events.append(helpers.new_task_completed_event(action.id, json.dumps(output))) + # Replay twice with identical recorded results: no model/tool body reruns. + before = list(invoked) + first_result = _execute(orchestrator, history, events) + second_result = _execute(orchestrator, history, events) + assert first_result.actions == second_result.actions + assert invoked == before + history += events + result = first_result + assert len(batches) < 10 + complete = _completion(result) + if fail_tool: + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "recorded tool failure" in complete.failureDetails.errorMessage + assert batches == [["model"], ["first", "second"]] + assert invoked == ["model", "first", "second"] + return + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, complete.failureDetails + assert json.loads(complete.result.value) == "done" + assert batches == [["model"], ["first", "second"], ["model"]] + assert invoked == ["model", "first", "second", "model"] From 610a85197d7e04ee679ac2b3c4b96ce6550a0c39 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Fri, 30 Oct 2026 12:47:08 -0600 Subject: [PATCH 05/11] Document coordinated async contract across provider entry points Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- azure-functions-durable/CHANGELOG.md | 10 ++++++---- azure-functions-durable/README.md | 14 ++++++++++++++ .../testing/orchestrator_generator_wrapper.py | 4 ++++ docs/supported-patterns.md | 8 ++++++++ durabletask-azuremanaged/CHANGELOG.md | 8 +++++--- 5 files changed, 37 insertions(+), 7 deletions(-) diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 1d54e558..45871657 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -9,11 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 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. +- With the corresponding core SDK, native and v1-style orchestrators can use +`async def`, await durable tasks, and use deterministic asyncio composition and +durable sleeps/timeouts. See the [async contract](../docs/async-orchestrators.md). +The `orchestrator_generator_wrapper` helper accepts direct-await coroutines; +asyncio scheduling requires replay/backend tests. Requires a coordinated release after the core SDK is published; dependency -minimums are intentionally unchanged in this spike. +minimums are intentionally unchanged in this feature PR. - Added `DFApp.configure_serialization(ensure_ascii=False)` to opt in to compact Unicode JSON across durable clients, orchestration and entity workers, activity bindings, and registered blueprints. The default remains ASCII-escaped JSON. diff --git a/azure-functions-durable/README.md b/azure-functions-durable/README.md index 66caf2ce..125516b9 100644 --- a/azure-functions-durable/README.md +++ b/azure-functions-durable/README.md @@ -32,6 +32,20 @@ Key capabilities include durable orchestrations and sub-orchestrations, durable timers, external events, durable entities, retries, versioning, durable HTTP calls (`context.call_http(...)`), recurring scheduled tasks, and history export. +## 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. + +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. + ## Unicode JSON serialization By default, durable payloads preserve the shared `azure-functions` codec's diff --git a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py index d516a697..61408762 100644 --- a/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py +++ b/azure-functions-durable/azure/durable_functions/testing/orchestrator_generator_wrapper.py @@ -21,6 +21,10 @@ def orchestrator_generator_wrapper( back into the orchestrator (mirroring how failed activities, timers, or other tasks surface during replay). + This mock-based helper handles direct durable awaits only. It does not + simulate the runtime's asyncio scheduler; use replay/backend tests for + gather, task groups, sleeps and timeouts. + Parameters ---------- generator: Generator[Any, Any, Any] | Coroutine[Any, Any, Any] diff --git a/docs/supported-patterns.md b/docs/supported-patterns.md index b67df626..d8d0f3e2 100644 --- a/docs/supported-patterns.md +++ b/docs/supported-patterns.md @@ -2,6 +2,10 @@ The following orchestration patterns are currently supported. +Generator and native coroutine orchestrators share these durable patterns. +See the [async orchestrator contract](async-orchestrators.md) for supported +asyncio/AnyIO composition, durable clocks/timeouts, and dispatch-local cleanup. + ## Function chaining An orchestration can chain a sequence of function calls using the following syntax: @@ -81,6 +85,10 @@ to nested exception groups containing these signals on Python 3.11 and later. The existing warning-only handling of direct `asyncio.CancelledError` during forced teardown is unchanged. Other cancellation or process-control exits propagate after the worker attempts to abandon the accepted dispatch. +Native coroutine authoring distinguishes a cancelled local root task: an +unhandled local cancellation becomes `TaskCancelledError` and fails the +orchestration instead of repeatedly abandoning replay. It does not terminate +the durable instance; see the [async cancellation contract](async-orchestrators.md#failures-timeouts-and-cancellation). If a completion or abandonment RPC raises, its server-side outcome may be unknown. The worker logs the uncertainty and closes the dispatch's originating diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index df4de5ab..caabcd59 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -9,10 +9,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED -- With the corresponding core SDK prototype, `DurableTaskSchedulerWorker` +- With the corresponding core SDK, `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. +orchestrators, including deterministic asyncio composition and durable sleeps/ +timeouts. See the [async contract](../docs/async-orchestrators.md). +Requires a coordinated release after the core SDK is published; dependency +minimums are intentionally unchanged in this feature PR. FIXED From e314a6964ade70d849a084bee484c34bf87877bd Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Fri, 30 Oct 2026 12:50:30 -0600 Subject: [PATCH 06/11] Cover cancellation action order and Python 3.10 timeout behavior Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../durabletask/test_asyncio_orchestrators.py | 43 ++++++++++++++++++- 1 file changed, 42 insertions(+), 1 deletion(-) diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 25984de1..104d9baf 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -120,7 +120,7 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: else: with anyio.fail_after(2): await ctx.wait_for_external_event("never") - except TimeoutError: + except (TimeoutError, asyncio.TimeoutError): return "timed out" return "slept" @@ -383,3 +383,44 @@ async def resource(): else: assert [action.scheduleTask.name for action in result.actions] == ["wait"] assert "cleanup failed" in caplog.text + + +def test_task_group_cancellation_durable_cleanup_has_stable_action_order() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def branch(index: int) -> None: + try: + await ctx.call_activity("work", input=index) + finally: + # This is real task-group cancellation during replay, not + # dispatch teardown: durable compensation is still allowed. + with anyio.CancelScope(shield=True): + await ctx.call_activity("compensate", input=index) + + async with anyio.create_task_group() as group: + for index in range(4): + group.start_soon(branch, index) + + old = _start() + [helpers.new_task_scheduled_event(index, "work") for index in range(1, 5)] + failure = [helpers.new_task_failed_event(1, ValueError("boom"))] + # The failing branch first enters its durable finally, so deliver that + # completion before task-group sibling cancellation starts. + first = _execute(orchestrator, old, failure) + assert [(a.id, a.scheduleTask.name, json.loads(a.scheduleTask.input.value)) for a in first.actions] == [ + (5, "compensate", 0)] + old += failure + [helpers.new_task_scheduled_event(5, "compensate"), + helpers.new_task_completed_event(5, "null")] + expected = [helpers.new_schedule_task_action(index + 6, "compensate", json.dumps(index + 1), None) + for index in range(3)] + for _ in range(10): + assert _execute(orchestrator, [], old).actions == expected + + +def test_cancelling_last_loop_timer_cannot_hide_foreign_wait() -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + handle = asyncio.get_running_loop().call_later(1, lambda: None) + handle.cancel() + await asyncio.Future() + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "blocked without durable work" in complete.failureDetails.errorMessage From 8c050a7c1ca11b0372151592151f8776796e4485 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Fri, 30 Oct 2026 12:55:43 -0600 Subject: [PATCH 07/11] Report unobserved background failures and drain cleanup errors Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- docs/async-orchestrators.md | 2 + durabletask/internal/asyncio_loop.py | 22 ++++++++--- .../durabletask/test_asyncio_orchestrators.py | 39 +++++++++++++++++++ 3 files changed, 57 insertions(+), 6 deletions(-) diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md index 8344d203..db692fd8 100644 --- a/docs/async-orchestrators.md +++ b/docs/async-orchestrators.md @@ -81,6 +81,8 @@ for every durable child before surfacing failure. In contrast, `asyncio.gather` with default options reports the first failure without cancelling siblings. Retain and await siblings if their results matter. `return_exceptions=True` collects their outcomes in input order. +Unobserved local background-task failures are reported during dispatch cleanup; +ordinary exceptions are logged, not silently discarded or automatically joined. Task groups cancel and drain local sibling tasks on failure and use their library's normal exception grouping rules. AnyIO uses level cancellation: diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index 6f144953..ca9a0cf8 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -267,6 +267,8 @@ def close(self) -> None: self.ctx._async_cleanup = True # pyright: ignore[reportPrivateUsage] try: completed_at_close = {future for future in self._tasks if future.done()} + unobserved_at_close = { + future for future in completed_at_close if getattr(future, "_log_traceback", False)} for future in tuple(self._tasks): if not future.done(): future.cancel() @@ -278,18 +280,26 @@ def close(self) -> None: if not future.done(): future.cancel() self._drain() + failure: BaseException | None = None for future in tuple(self._tasks): if not future.done(): - future.get_coro().close() - raise RuntimeError("Async cleanup must finish without waiting for durable or external work.") - if not future.cancelled(): + try: + future.get_coro().close() + except BaseException as error: + failure = failure or error + failure = failure or RuntimeError( + "Async cleanup must finish without waiting for durable or external work.") + elif not future.cancelled(): error = future.exception() - if error is not None and future not in completed_at_close: + if error is not None and (future not in completed_at_close or future in unobserved_at_close): if not isinstance(error, Exception): - raise error + failure = failure or error logger = self.ctx._logger or logging.getLogger(__name__) # pyright: ignore[reportPrivateUsage] logger.warning( - "%s: Async task cleanup failed.", self.ctx.instance_id, exc_info=error) + "%s: Async %s task failed.", self.ctx.instance_id, + "background" if future in completed_at_close else "cleanup", exc_info=error) + if failure is not None: + raise failure finally: self._closed = True self._ready.clear() diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 104d9baf..552a618d 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -424,3 +424,42 @@ async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: complete = _completion(_execute(orchestrator, [], _start())) assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED assert "blocked without durable work" in complete.failureDetails.errorMessage + + +@pytest.mark.parametrize("observed", [False, True]) +def test_background_failure_is_reported_unless_observed( + observed: bool, caplog: pytest.LogCaptureFixture) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: + async def fail() -> None: + raise ValueError("background failure") + + pending = asyncio.create_task(fail()) + await asyncio.sleep(0) + if observed: + await asyncio.gather(pending, return_exceptions=True) + return "done" + + assert json.loads(_completion(_execute(orchestrator, [], _start())).result.value) == "done" + assert ("background failure" in caplog.text) is not observed + + +def test_unexpected_cleanup_failure_does_not_hide_other_task_errors( + caplog: pytest.LogCaptureFixture) -> None: + class CleanupAbort(BaseException): + pass + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def branch(error: BaseException) -> None: + try: + await ctx.wait_for_external_event("never") + finally: + await asyncio.sleep(0) + raise error + + asyncio.create_task(branch(CleanupAbort("first cleanup failure"))) + asyncio.create_task(branch(ValueError("second cleanup failure"))) + await ctx.call_activity("wait") + + complete = _completion(_execute(orchestrator, [], _start())) + assert complete.failureDetails.errorMessage == "first cleanup failure" + assert "second cleanup failure" in caplog.text From 635f5d3d20add350f04ebe11abe71b8b17d89627 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 2 Nov 2026 10:17:04 -0700 Subject: [PATCH 08/11] Harden async cleanup and scale completion delivery after review Continue bounded cleanup after callback and process-control failures. Deliver durable completions through ordered subscriptions instead of repeated pending-work scans, and scale the cleanup budget for large cooperative fan-out. Add focused regressions and keep documentation adoption guidance minimal. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 + README.md | 4 +- azure-functions-durable/CHANGELOG.md | 2 + docs/async-orchestrators.md | 7 +- durabletask-azuremanaged/CHANGELOG.md | 2 + durabletask/internal/asyncio_loop.py | 102 +++++++++++++----- .../durabletask/test_asyncio_orchestrators.py | 102 ++++++++++++++++++ 7 files changed, 194 insertions(+), 27 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 657bd191..e9956c38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,8 @@ timeouts use durable timers and the orchestration clock; `sleep(0)` yields locally without a timer. See the [supported contract](docs/async-orchestrators.md) for cancellation, per-dispatch cleanup, replay/migration rules, and unsupported APIs. External I/O still belongs in ordinary activities. +Async fan-out scales to large batches, and cleanup callback failures do not +skip other cooperative finalizers. FIXED diff --git a/README.md b/README.md index 5d6f6a7b..85c20be7 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,9 @@ and [Azure Durable Functions](https://learn.microsoft.com/azure/azure-functions/ ### Async Orchestrators All three packages support `async def` orchestrators and nested async helpers. -Existing generator orchestrators continue to work. +Existing generator orchestrators continue to work. Async authoring is an +additive option, not yet the preferred programming model; existing examples +and tests are not being broadly migrated in this feature. ```python from durabletask import task diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 45871657..b56f307b 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -12,6 +12,8 @@ ADDED - With the corresponding core SDK, native and v1-style orchestrators can use `async def`, await durable tasks, and use deterministic asyncio composition and durable sleeps/timeouts. See the [async contract](../docs/async-orchestrators.md). +Async fan-out scales to large batches, and cleanup callback failures do not +skip other cooperative finalizers. The `orchestrator_generator_wrapper` helper accepts direct-await coroutines; asyncio scheduling requires replay/backend tests. Requires a coordinated release after the core SDK is published; dependency diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md index db692fd8..5d456c74 100644 --- a/docs/async-orchestrators.md +++ b/docs/async-orchestrators.md @@ -7,6 +7,10 @@ retry, payload, and entity engine. Register the function normally with `orchestration_trigger` decorator. Functions supports both native two-argument and compatibility single-argument orchestrators. +Async authoring is an additive option, not yet the preferred programming model. +Broad conversion of existing examples, documentation, and tests is deferred +to a future change after the contract has settled. + ```python import asyncio @@ -121,7 +125,8 @@ error. Teardown must cooperate with cancellation; a non-quiescent cleanup queue is bounded and reported, not allowed to run indefinitely. Ordinary teardown failures (including `CancelledError`) are logged without -changing the decided response. Unexpected non-cancellation `BaseException` +changing the decided response or skipping other cooperative finalizers. +Unexpected non-cancellation `BaseException` exits retain the existing terminal-failure/recovery policy; an existing primary failure is preserved. Teardown status/actions do not replace the response. Held entity-release obligations are captured before teardown for terminal diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index caabcd59..3e08b4ac 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -13,6 +13,8 @@ ADDED accepts `async def` orchestrators that await durable tasks alongside generator orchestrators, including deterministic asyncio composition and durable sleeps/ timeouts. See the [async contract](../docs/async-orchestrators.md). +Async fan-out scales to large batches, and cleanup callback failures do not +skip other cooperative finalizers. Requires a coordinated release after the core SDK is published; dependency minimums are intentionally unchanged in this feature PR. diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index ca9a0cf8..fba15ca5 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -23,6 +23,21 @@ T = TypeVar("T") +class _CompletionObserver(task.CompositeTask[None]): + def __init__(self, durable: task.Task[Any], callback: Callable[[], None]) -> None: + self._callback = callback + super().__init__([durable]) + + def on_child_completed(self, task: task.Task[Any]) -> None: + if not self._is_complete: + self._is_complete = True + self._notify_parent() + self._callback() + + def detach(self) -> None: + self._notify_parent() + + class _DurableAsyncTask(asyncio.Task[T]): def cancel(self, msg: Any = None) -> bool: loop = self.get_loop() @@ -63,11 +78,14 @@ def __init__(self, ctx: _RuntimeOrchestrationContext) -> None: self._tasks: dict[asyncio.Task[Any], int] = {} self.cancelling_task: asyncio.Task[Any] | None = None self._waits: dict[int, tuple[task.Task[Any], list[asyncio.Future[Any]]]] = {} - self._timers: list[_DurableTimerHandle] = [] + self._observers: dict[int, _CompletionObserver] = {} + self._completions: deque[Callable[[], None]] = deque() + self._timers: dict[_DurableTimerHandle, _CompletionObserver] = {} self._asyncgens: dict[int, Any] = {} self._closed = False self._closing = False self._fatal: BaseException | None = None + self._cleanup_failure: BaseException | None = None self._root: asyncio.Task[Any] | None = None def time(self) -> float: @@ -150,12 +168,21 @@ def call_at( if (deadline - datetime(1970, 1, 1)).total_seconds() < when: deadline += timedelta(microseconds=1) handle.durable_timer = self.ctx.create_timer(deadline) - self._timers.append(handle) + self._timers[handle] = _CompletionObserver( + handle.durable_timer, + lambda: self._completions.append(lambda: self._complete_timer(handle))) return handle def _timer_handle_cancelled(self, handle: asyncio.TimerHandle) -> None: - # No selector heap to maintain; _DurableTimerHandle owns cancellation. - pass + if isinstance(handle, _DurableTimerHandle): + observer = self._timers.pop(handle, None) + if observer is not None: + observer.detach() + + def _complete_timer(self, handle: _DurableTimerHandle) -> None: + self._timers.pop(handle, None) + if not handle.cancelled(): + self._ready.append(handle) async def wait_durable(self, durable: task.Task[T]) -> T: if self._closing: @@ -165,6 +192,9 @@ async def wait_durable(self, durable: task.Task[T]) -> T: future = self.create_future() entry = self._waits.setdefault(id(durable), (durable, [])) entry[1].append(future) + if id(durable) not in self._observers: + self._observers[id(durable)] = _CompletionObserver( + durable, lambda: self._completions.append(lambda: self._complete_wait(durable))) try: return await future finally: @@ -172,22 +202,21 @@ async def wait_durable(self, durable: task.Task[T]) -> T: entry[1].remove(future) if not entry[1]: self._waits.pop(id(durable), None) + self._observers.pop(id(durable)).detach() + + def _complete_wait(self, durable: task.Task[Any]) -> None: + entry = self._waits.get(id(durable)) + if entry is not None: + for future in tuple(entry[1]): + if not future.done(): + try: + future.set_result(durable.get_result()) + except Exception as error: + future.set_exception(error) def _deliver_completions(self) -> None: - for durable, futures in tuple(self._waits.values()): - if durable.is_complete: - for future in tuple(futures): - if not future.done(): - try: - future.set_result(durable.get_result()) - except Exception as error: - future.set_exception(error) - for handle in tuple(self._timers): - if handle.cancelled(): - self._timers.remove(handle) - elif handle.durable_timer is not None and handle.durable_timer.is_complete: - self._timers.remove(handle) - self._ready.append(handle) + while self._completions: + self._completions.popleft()() def start(self, coro: Coroutine[Any, Any, Any]) -> None: self._root = self.create_task(coro, name="durable-orchestrator") @@ -214,8 +243,7 @@ def resume(self) -> None: "Await durable activities, events or timers; external I/O is unsupported.") def _has_durable_work(self) -> bool: - return any(not durable.is_complete for durable, _ in self._waits.values()) or any( - not handle.cancelled() for handle in self._timers) + return any(not durable.is_complete for durable, _ in self._waits.values()) or bool(self._timers) def _drain(self) -> None: previous_loop = asyncio._get_running_loop() @@ -227,15 +255,23 @@ def _drain(self) -> None: sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter, finalizer=self._asyncgen_finalizer) try: steps = 0 + cleanup_limit = max(10000, 4 * (len(self._tasks) + len(self._asyncgens))) while self._ready or self._cancel_ready: self._ready.extend(handle for _, _, handle in sorted(self._cancel_ready)) self._cancel_ready.clear() handle = self._ready.popleft() if not handle.cancelled(): - handle._run() + try: + handle._run() + except BaseException as error: + if not self._closing: + raise + self._record_cleanup_error(error) if self._fatal is not None: error, self._fatal = self._fatal, None - raise error + if not self._closing: + raise error + self._record_cleanup_error(error) if not self._closing and self.ctx._is_complete: # pyright: ignore[reportPrivateUsage] break if not self._closing and self._root is not None and self._root.done(): @@ -243,8 +279,12 @@ def _drain(self) -> None: if not self._closing: self._deliver_completions() steps += 1 - if self._closing and steps >= 10000: - raise RuntimeError("Async cleanup did not quiesce; cleanup must cooperate with cancellation.") + if self._closing and steps >= cleanup_limit: + self._record_cleanup_error(RuntimeError( + "Async cleanup did not quiesce; cleanup must cooperate with cancellation.")) + self._ready.clear() + self._cancel_ready.clear() + break finally: sys.set_asyncgen_hooks(*previous_hooks) asyncio._set_running_loop(previous_loop) @@ -262,6 +302,14 @@ def call_exception_handler(self, context: dict[str, Any]) -> None: error = context.get("exception") self._fatal = error if isinstance(error, BaseException) else RuntimeError(context["message"]) + def _record_cleanup_error(self, error: BaseException) -> None: + logger = self.ctx._logger or logging.getLogger(__name__) # pyright: ignore[reportPrivateUsage] + logger.warning("%s: Async cleanup callback failed.", self.ctx.instance_id, exc_info=error) + if not isinstance(error, (Exception, asyncio.CancelledError)): + from durabletask.worker import _is_control_flow_exception # pyright: ignore[reportPrivateUsage] + if self._cleanup_failure is None or _is_control_flow_exception(error): + self._cleanup_failure = error + def close(self) -> None: self._closing = True self.ctx._async_cleanup = True # pyright: ignore[reportPrivateUsage] @@ -280,7 +328,7 @@ def close(self) -> None: if not future.done(): future.cancel() self._drain() - failure: BaseException | None = None + failure: BaseException | None = self._cleanup_failure for future in tuple(self._tasks): if not future.done(): try: @@ -304,6 +352,10 @@ def close(self) -> None: self._closed = True self._ready.clear() self._cancel_ready.clear() + for observer in (*self._observers.values(), *self._timers.values()): + observer.detach() + self._completions.clear() + self._observers.clear() self._waits.clear() self._timers.clear() self._tasks.clear() diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 552a618d..9048e799 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -463,3 +463,105 @@ async def branch(error: BaseException) -> None: complete = _completion(_execute(orchestrator, [], _start())) assert complete.failureDetails.errorMessage == "first cleanup failure" assert "second cleanup failure" in caplog.text + + +@pytest.mark.parametrize("control_flow", [False, True]) +def test_cleanup_callback_failure_still_drains_tasks_and_asyncgens( + control_flow: bool, caplog: pytest.LogCaptureFixture) -> None: + cleaned: list[str] = [] + retained: list[asyncio.Task[Any]] = [] + + def fail() -> None: + raise SystemExit("cleanup exit") if control_flow else ValueError("cleanup callback") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def resource(): + try: + yield 1 + finally: + await asyncio.sleep(0) + cleaned.append("asyncgen") + + async def sibling() -> None: + try: + await ctx.wait_for_external_event("never") + finally: + await asyncio.sleep(0) + cleaned.append("sibling") + + opened = resource() + await anext(opened) + retained.append(asyncio.create_task(sibling())) + current = asyncio.current_task() + assert current is not None + retained.append(current) + try: + await ctx.call_activity("wait") + finally: + if control_flow: + fail() + asyncio.get_running_loop().call_soon(fail) + await asyncio.sleep(0) + cleaned.append("root") + + if control_flow: + with pytest.raises(SystemExit, match="cleanup exit"): + _execute(orchestrator, [], _start()) + else: + result = _execute(orchestrator, [], _start()) + assert [action.scheduleTask.name for action in result.actions] == ["wait"] + assert "cleanup callback" in caplog.text + assert "root" in cleaned + assert "sibling" in cleaned + assert "asyncgen" in cleaned + assert all(pending.done() for pending in retained) + + +@pytest.mark.parametrize("count", [128, 4096]) +def test_fanout_waiter_inspections_scale_linearly( + count: int, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None: + inspections = 0 + + def is_complete(durable: task.Task[Any]) -> bool: + nonlocal inspections + inspections += 1 + return durable._is_complete + + monkeypatch.setattr(task.Task, "is_complete", property(is_complete)) + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[Any]: + async def branch(index: int) -> Any: + return await ctx.call_activity("work", input=index) + + return await asyncio.gather(*(branch(index) for index in range(count))) + + result = _execute(orchestrator, [], _start()) + assert [action.id for action in result.actions] == list(range(1, count + 1)) + assert inspections < count * 20 + assert not caplog.records + + +def test_nonquiescent_cleanup_does_not_skip_asyncgen_closure( + caplog: pytest.LogCaptureFixture) -> None: + cleaned: list[str] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def resource(): + try: + yield 1 + finally: + await asyncio.sleep(0) + cleaned.append("asyncgen") + + opened = resource() + await anext(opened) + try: + await ctx.call_activity("wait") + finally: + while True: + await asyncio.sleep(0) + + result = _execute(orchestrator, [], _start()) + assert [action.scheduleTask.name for action in result.actions] == ["wait"] + assert cleaned == ["asyncgen"] + assert "cleanup did not quiesce" in caplog.text From 695bb12dc502b479bbb3c51727d215e0098b73fe Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 2 Nov 2026 10:37:31 -0700 Subject: [PATCH 09/11] Keep async documentation focused on feature behavior Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- README.md | 6 ++---- azure-functions-durable/CHANGELOG.md | 2 -- docs/async-orchestrators.md | 29 ++++++--------------------- durabletask-azuremanaged/CHANGELOG.md | 2 -- 4 files changed, 8 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index 41834b7d..faf545fd 100644 --- a/README.md +++ b/README.md @@ -35,9 +35,7 @@ The Azure Functions Durable provider requires Python 3.13 or later. ### Async Orchestrators All three packages support `async def` orchestrators and nested async helpers. -Existing generator orchestrators continue to work. Async authoring is an -additive option, not yet the preferred programming model; existing examples -and tests are not being broadly migrated in this feature. +Existing generator orchestrators continue to work. ```python from durabletask import task @@ -72,7 +70,7 @@ An `async def` function containing `yield` is an async generator, not a supporte 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, framework adapters, and coordinated release rules. +ordering-sensitive APIs and framework adapters. ### Large Payload Externalization diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index c048d83c..fac6c245 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -16,8 +16,6 @@ Async fan-out scales to large batches, and cleanup callback failures do not skip other cooperative finalizers. The `orchestrator_generator_wrapper` helper accepts direct-await coroutines; asyncio scheduling requires replay/backend tests. -Requires a coordinated release after the core SDK is published; dependency -minimums are intentionally unchanged in this feature PR. - Added `DFApp.configure_serialization(ensure_ascii=False)` to opt in to compact Unicode JSON across durable clients, orchestration and entity workers, activity bindings, and registered blueprints. The default remains ASCII-escaped JSON. diff --git a/docs/async-orchestrators.md b/docs/async-orchestrators.md index 5d456c74..08ee638a 100644 --- a/docs/async-orchestrators.md +++ b/docs/async-orchestrators.md @@ -7,10 +7,6 @@ retry, payload, and entity engine. Register the function normally with `orchestration_trigger` decorator. Functions supports both native two-argument and compatibility single-argument orchestrators. -Async authoring is an additive option, not yet the preferred programming model. -Broad conversion of existing examples, documentation, and tests is deferred -to a future change after the contract has settled. - ```python import asyncio @@ -146,17 +142,12 @@ automatic recording or interception of arbitrary await expressions. `is_replaying` is not a substitute for recording results. Activities remain at-least-once work: side effects require appropriate idempotency/deduplication. -Pydantic AI's real `Agent.run` / parallel-tool AnyIO path is exercised with a -deterministic `TestModel` in -[`test_pydantic_ai_orchestrator.py`](../tests/durabletask/test_pydantic_ai_orchestrator.py). -The orchestrator directly awaits the agent; no adapter event loop, trampoline, -or per-batch durable join is needed. The proof adapter still registers ordinary -model/tool activities, encodes/decodes framework payloads, supplies replay-stable -IDs and a deadline clock, and explicitly canonicalizes volatile message -timestamps. These are **adapter responsibilities**, not SDK-wide interception. -This proof is not a complete production Pydantic adapter or compatibility -promise for arbitrary agents, model providers, tools, or framework versions. -No paid model calls are needed. +Framework adapters can compose supported durable operations using async +helpers. Adapters remain responsible for registering ordinary model/tool +activities, encoding/decoding framework payloads, and supplying replay-stable +IDs, clocks, and message metadata. These responsibilities are not handled by +SDK-wide interception. Libraries that require unsupported asyncio APIs or +nondeterministic operations need an appropriate adapter or activity boundary. Changing an in-flight generator to equivalent coroutine code is safe **only if its durable history contract is preserved**: action IDs/order, names, @@ -169,11 +160,3 @@ The Functions `orchestrator_generator_wrapper` helper supports direct durable awaits for 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. - -## Coordinated delivery - -This feature is one draft PR/release unit across all packages. Do not ship a -limited intermediate contract. Release the core package first; then use a -dedicated coordinated release PR to update provider dependency minimums, -versions, and release notes together. Feature development does not change -package versions or dependency minimums. diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 4bfcf984..35faac27 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -15,8 +15,6 @@ orchestrators, including deterministic asyncio composition and durable sleeps/ timeouts. See the [async contract](../docs/async-orchestrators.md). Async fan-out scales to large batches, and cleanup callback failures do not skip other cooperative finalizers. -Requires a coordinated release after the core SDK is published; dependency -minimums are intentionally unchanged in this feature PR. CHANGED From eba443150c74e2b54b11b53c7eb97ece2a0549d7 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 2 Nov 2026 10:42:09 -0700 Subject: [PATCH 10/11] Preserve process-control exits across coroutine cleanup failures Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 3 +- azure-functions-durable/CHANGELOG.md | 3 +- durabletask-azuremanaged/CHANGELOG.md | 3 +- durabletask/internal/asyncio_loop.py | 17 +++++----- .../durabletask/test_asyncio_orchestrators.py | 34 +++++++++++++++++++ 5 files changed, 49 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 32feb551..2bd57759 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,7 +19,8 @@ locally without a timer. See the [supported contract](docs/async-orchestrators.m for cancellation, per-dispatch cleanup, replay/migration rules, and unsupported APIs. External I/O still belongs in ordinary activities. Async fan-out scales to large batches, and cleanup callback failures do not -skip other cooperative finalizers. +skip other cooperative finalizers. Process-control exits propagate even when +another finalizer fails. CHANGED diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index fac6c245..4d8e12b4 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -13,7 +13,8 @@ ADDED `async def`, await durable tasks, and use deterministic asyncio composition and durable sleeps/timeouts. See the [async contract](../docs/async-orchestrators.md). Async fan-out scales to large batches, and cleanup callback failures do not -skip other cooperative finalizers. +skip other cooperative finalizers. Process-control exits propagate even when +another finalizer fails. The `orchestrator_generator_wrapper` helper accepts direct-await coroutines; asyncio scheduling requires replay/backend tests. - Added `DFApp.configure_serialization(ensure_ascii=False)` to opt in to compact diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 35faac27..825f4a6f 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -14,7 +14,8 @@ accepts `async def` orchestrators that await durable tasks alongside generator orchestrators, including deterministic asyncio composition and durable sleeps/ timeouts. See the [async contract](../docs/async-orchestrators.md). Async fan-out scales to large batches, and cleanup callback failures do not -skip other cooperative finalizers. +skip other cooperative finalizers. Process-control exits propagate even when +another finalizer fails. CHANGED diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index fba15ca5..e8f44901 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -266,6 +266,7 @@ def _drain(self) -> None: except BaseException as error: if not self._closing: raise + # Defer all exits until other cleanup tasks have drained. self._record_cleanup_error(error) if self._fatal is not None: error, self._fatal = self._fatal, None @@ -302,9 +303,9 @@ def call_exception_handler(self, context: dict[str, Any]) -> None: error = context.get("exception") self._fatal = error if isinstance(error, BaseException) else RuntimeError(context["message"]) - def _record_cleanup_error(self, error: BaseException) -> None: + def _record_cleanup_error(self, error: BaseException, source: str = "cleanup callback") -> None: logger = self.ctx._logger or logging.getLogger(__name__) # pyright: ignore[reportPrivateUsage] - logger.warning("%s: Async cleanup callback failed.", self.ctx.instance_id, exc_info=error) + logger.warning("%s: Async %s failed.", self.ctx.instance_id, source, exc_info=error) if not isinstance(error, (Exception, asyncio.CancelledError)): from durabletask.worker import _is_control_flow_exception # pyright: ignore[reportPrivateUsage] if self._cleanup_failure is None or _is_control_flow_exception(error): @@ -334,18 +335,18 @@ def close(self) -> None: try: future.get_coro().close() except BaseException as error: + # Close siblings before propagating process-control exits. + self._record_cleanup_error(error, "coroutine cleanup") failure = failure or error failure = failure or RuntimeError( "Async cleanup must finish without waiting for durable or external work.") elif not future.cancelled(): error = future.exception() if error is not None and (future not in completed_at_close or future in unobserved_at_close): - if not isinstance(error, Exception): - failure = failure or error - logger = self.ctx._logger or logging.getLogger(__name__) # pyright: ignore[reportPrivateUsage] - logger.warning( - "%s: Async %s task failed.", self.ctx.instance_id, - "background" if future in completed_at_close else "cleanup", exc_info=error) + self._record_cleanup_error( + error, "background task" if future in completed_at_close else "task cleanup") + if self._cleanup_failure is not None: + raise self._cleanup_failure if failure is not None: raise failure finally: diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 9048e799..10ed3c84 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -565,3 +565,37 @@ async def resource(): assert [action.scheduleTask.name for action in result.actions] == ["wait"] assert cleaned == ["asyncgen"] assert "cleanup did not quiesce" in caplog.text + + +@pytest.mark.parametrize("grouped", [False, True]) +def test_forced_cleanup_preserves_process_control_after_other_errors(grouped: bool) -> None: + cleaned: list[str] = [] + exit_error: BaseException = SystemExit("forced cleanup exit") + if grouped: + exit_error = BaseExceptionGroup("forced cleanup group", [exit_error, ValueError("nested")]) + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def branch(name: str, error: BaseException | None) -> None: + try: + await ctx.wait_for_external_event(name) + finally: + try: + while True: + try: + await asyncio.Future() + except asyncio.CancelledError: + pass + finally: + cleaned.append(name) + if error is not None: + raise error + + asyncio.create_task(branch("ordinary", ValueError("ordinary cleanup error"))) + asyncio.create_task(branch("exit", exit_error)) + asyncio.create_task(branch("last", None)) + await ctx.call_activity("wait") + + with pytest.raises(type(exit_error)) as caught: + _execute(orchestrator, [], _start()) + assert caught.value is exit_error + assert cleaned == ["ordinary", "exit", "last"] From 688c8cb45dccedcee2196eb0d472f9fa6334bdbe Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Mon, 2 Nov 2026 11:35:14 -0700 Subject: [PATCH 11/11] Harden async callbacks and retain framework-independent coverage Keep identical timer registrations independent and reject callback StopIteration as a failure. Replace the custom Pydantic integration proof with AnyIO context/replay coverage and preserve cleanup control-flow precedence. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/workflows/durabletask.yml | 1 - CHANGELOG.md | 2 + azure-functions-durable/CHANGELOG.md | 2 + durabletask-azuremanaged/CHANGELOG.md | 2 + durabletask/internal/asyncio_loop.py | 9 + requirements.txt | 1 - .../durabletask/test_asyncio_orchestrators.py | 156 +++++++++++++++++- .../test_pydantic_ai_orchestrator.py | 131 --------------- 8 files changed, 169 insertions(+), 135 deletions(-) delete mode 100644 tests/durabletask/test_pydantic_ai_orchestrator.py diff --git a/.github/workflows/durabletask.yml b/.github/workflows/durabletask.yml index 93e23b45..1d28fecc 100644 --- a/.github/workflows/durabletask.yml +++ b/.github/workflows/durabletask.yml @@ -32,7 +32,6 @@ jobs: python -m pytest tests\durabletask\test_asyncio_orchestrators.py tests\durabletask\test_coroutine_orchestrators.py - tests\durabletask\test_pydantic_ai_orchestrator.py tests\durabletask\test_orchestration_generator_cleanup.py tests\durabletask\test_orchestration_dispatch_recovery.py -q diff --git a/CHANGELOG.md b/CHANGELOG.md index 2bd57759..5855f301 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,8 @@ APIs. External I/O still belongs in ordinary activities. Async fan-out scales to large batches, and cleanup callback failures do not skip other cooperative finalizers. Process-control exits propagate even when another finalizer fails. +Identical timer callback registrations remain independent, and callback +`StopIteration` errors fail the orchestration rather than completing it. CHANGED diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 4d8e12b4..bd4d5f74 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -15,6 +15,8 @@ durable sleeps/timeouts. See the [async contract](../docs/async-orchestrators.md Async fan-out scales to large batches, and cleanup callback failures do not skip other cooperative finalizers. Process-control exits propagate even when another finalizer fails. +Identical timer callback registrations remain independent, and callback +`StopIteration` errors fail the orchestration rather than completing it. The `orchestrator_generator_wrapper` helper accepts direct-await coroutines; asyncio scheduling requires replay/backend tests. - Added `DFApp.configure_serialization(ensure_ascii=False)` to opt in to compact diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index 825f4a6f..f15161a4 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -16,6 +16,8 @@ timeouts. See the [async contract](../docs/async-orchestrators.md). Async fan-out scales to large batches, and cleanup callback failures do not skip other cooperative finalizers. Process-control exits propagate even when another finalizer fails. +Identical timer callback registrations remain independent, and callback +`StopIteration` errors fail the orchestration rather than completing it. CHANGED diff --git a/durabletask/internal/asyncio_loop.py b/durabletask/internal/asyncio_loop.py index e8f44901..cd9383e8 100644 --- a/durabletask/internal/asyncio_loop.py +++ b/durabletask/internal/asyncio_loop.py @@ -51,6 +51,11 @@ def cancel(self, msg: Any = None) -> bool: class _DurableTimerHandle(asyncio.TimerHandle): + def __eq__(self, other: object) -> bool: + return self is other + + __hash__ = object.__hash__ + def __init__( self, when: float, callback: Callable[..., Any], args: tuple[Any, ...], loop: DurableEventLoop, context: contextvars.Context | None, @@ -301,6 +306,10 @@ def _asyncgen_finalizer(self, generator: Any) -> None: def call_exception_handler(self, context: dict[str, Any]) -> None: error = context.get("exception") + if isinstance(error, StopIteration): + failure = RuntimeError("An orchestration callback raised StopIteration.") + failure.__cause__ = error + error = failure self._fatal = error if isinstance(error, BaseException) else RuntimeError(context["message"]) def _record_cleanup_error(self, error: BaseException, source: str = "cleanup callback") -> None: diff --git a/requirements.txt b/requirements.txt index 0fcf7ce8..2e0a5d7e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,7 +4,6 @@ protobuf pytest pytest-asyncio anyio -pydantic-ai-slim==2.60.0; python_version >= "3.11" # Framework proof, no model providers pytest-cov azure-identity azure-functions diff --git a/tests/durabletask/test_asyncio_orchestrators.py b/tests/durabletask/test_asyncio_orchestrators.py index 10ed3c84..18a14d10 100644 --- a/tests/durabletask/test_asyncio_orchestrators.py +++ b/tests/durabletask/test_asyncio_orchestrators.py @@ -8,6 +8,7 @@ import json import sys from contextlib import asynccontextmanager +from contextvars import ContextVar from datetime import timedelta from typing import Any from weakref import ref @@ -21,6 +22,59 @@ from tests.durabletask.test_coroutine_orchestrators import _completion, _execute, _start, NOW +@pytest.mark.parametrize("fail_branch", [False, True]) +def test_anyio_branches_inherit_context_and_replay_stable_actions(fail_branch: bool) -> None: + context: ContextVar[task.OrchestrationContext] = ContextVar("orchestration") + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> Any: + token = context.set(ctx) + try: + results: dict[str, Any] = {} + + async def branch(name: str) -> None: + current = context.get() + assert current is ctx + results[name] = await current.call_activity(name, input=current.instance_id) + + async with anyio.create_task_group() as group: + group.start_soon(branch, "first") + group.start_soon(branch, "second") + return await context.get().call_activity("finish", input=[results["first"], results["second"]]) + finally: + context.reset(token) + + def replay(old: list[pb.HistoryEvent], new: list[pb.HistoryEvent]): + first = _execute(orchestrator, old, new) + assert _execute(orchestrator, old, new).actions == first.actions + with pytest.raises(LookupError): + context.get() + return first + + start = _start() + initial = replay([], start) + assert [action.scheduleTask.name for action in initial.actions] == ["first", "second"] + assert all(json.loads(action.scheduleTask.input.value) == "coroutine-test" for action in initial.actions) + old = start + [helpers.new_task_scheduled_event(1, "first"), + helpers.new_task_scheduled_event(2, "second")] + second_done = [helpers.new_task_completed_event(2, '"second result"')] + assert replay(old, second_done).actions == [] + first_done = [helpers.new_task_failed_event(1, ValueError("recorded branch failure"))] if fail_branch else [ + helpers.new_task_completed_event(1, '"first result"')] + result = replay(old + second_done, first_done) + if fail_branch: + complete = _completion(result) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert "recorded branch failure" in str(complete.failureDetails) + else: + assert len(result.actions) == 1 + assert result.actions[0].scheduleTask.name == "finish" + assert json.loads(result.actions[0].scheduleTask.input.value) == ["first result", "second result"] + complete = _completion(replay(old + second_done + first_done, [ + helpers.new_task_scheduled_event(3, "finish"), + helpers.new_task_completed_event(3, '"done"')])) + assert json.loads(complete.result.value) == "done" + + def test_gather_resumes_each_branch_without_waiting_for_batch() -> None: async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> list[Any]: async def branch(name: str) -> Any: @@ -568,7 +622,11 @@ async def resource(): @pytest.mark.parametrize("grouped", [False, True]) -def test_forced_cleanup_preserves_process_control_after_other_errors(grouped: bool) -> None: +@pytest.mark.parametrize("custom_error", [False, True]) +def test_forced_cleanup_preserves_process_control_after_other_errors(grouped: bool, custom_error: bool) -> None: + class CleanupAbort(BaseException): + pass + cleaned: list[str] = [] exit_error: BaseException = SystemExit("forced cleanup exit") if grouped: @@ -584,13 +642,15 @@ async def branch(name: str, error: BaseException | None) -> None: try: await asyncio.Future() except asyncio.CancelledError: + # Force teardown to close a cancellation-resistant coroutine. pass finally: cleaned.append(name) if error is not None: raise error - asyncio.create_task(branch("ordinary", ValueError("ordinary cleanup error"))) + first_error = CleanupAbort("custom cleanup error") if custom_error else ValueError("ordinary cleanup error") + asyncio.create_task(branch("ordinary", first_error)) asyncio.create_task(branch("exit", exit_error)) asyncio.create_task(branch("last", None)) await ctx.call_activity("wait") @@ -599,3 +659,95 @@ async def branch(name: str, error: BaseException | None) -> None: _execute(orchestrator, [], _start()) assert caught.value is exit_error assert cleaned == ["ordinary", "exit", "last"] + + +@pytest.mark.parametrize("timer", [False, True]) +def test_callback_stop_iteration_fails_instead_of_completing_orchestrator(timer: bool) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + def fail() -> None: + raise StopIteration("callback stopped") + + loop = asyncio.get_running_loop() + if timer: + loop.call_later(1, fail) + else: + loop.call_soon(fail) + await ctx.wait_for_external_event("never") + + start = _start() + if timer: + deadline = NOW + timedelta(seconds=1) + old = start + [helpers.new_timer_created_event(1, deadline)] + new = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(1, deadline)] + else: + old, new = [], start + complete = _completion(_execute(orchestrator, old, new)) + assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete.failureDetails.errorType == "builtins.RuntimeError" + assert complete.failureDetails.innerFailure.errorType == "builtins.StopIteration" + assert complete.failureDetails.innerFailure.errorMessage == "callback stopped" + + +@pytest.mark.parametrize("cancelled", [None, 0, 1]) +def test_identical_timer_callbacks_complete_and_cancel_independently(cancelled: int | None) -> None: + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> int: + loop = asyncio.get_running_loop() + result = loop.create_future() + completed: list[None] = [] + expected = 2 if cancelled is None else 1 + + def callback() -> None: + completed.append(None) + if len(completed) == expected: + result.set_result(len(completed)) + + handles = [loop.call_at(loop.time() + 1, callback) for _ in range(2)] + assert handles[0] != handles[1] + assert len(set(handles)) == 2 + if cancelled is not None: + handles[cancelled].cancel() + return await result + + start = _start() + initial = _execute(orchestrator, [], start) + remaining = [index + 1 for index in range(2) if index != cancelled] + assert [action.id for action in initial.actions] == remaining + deadline = NOW + timedelta(seconds=1) + old = start + [helpers.new_timer_created_event(index, deadline) for index in remaining] + for offset, timer_id in enumerate(remaining): + new = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(timer_id, deadline)] + result = _execute(orchestrator, old, new) + assert _execute(orchestrator, old, new).actions == result.actions + if offset < len(remaining) - 1: + assert result.actions == [] + else: + assert json.loads(_completion(result).result.value) == len(remaining) + old += new + + +def test_cooperative_cleanup_prioritizes_grouped_process_control() -> None: + class CleanupAbort(BaseException): + pass + + exit_error = BaseExceptionGroup("cleanup exit", [SystemExit("stop"), ValueError("nested")]) + cleaned: list[str] = [] + + async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None: + async def branch(name: str, error: BaseException | None) -> None: + try: + await ctx.wait_for_external_event(name) + finally: + await asyncio.sleep(0) + cleaned.append(name) + if error is not None: + raise error + + asyncio.create_task(branch("custom", CleanupAbort("first failure"))) + asyncio.create_task(branch("exit", exit_error)) + asyncio.create_task(branch("last", None)) + await ctx.call_activity("wait") + + with pytest.raises(BaseExceptionGroup) as caught: + _execute(orchestrator, [], _start()) + assert caught.value is exit_error + assert cleaned == ["custom", "exit", "last"] diff --git a/tests/durabletask/test_pydantic_ai_orchestrator.py b/tests/durabletask/test_pydantic_ai_orchestrator.py deleted file mode 100644 index fa856410..00000000 --- a/tests/durabletask/test_pydantic_ai_orchestrator.py +++ /dev/null @@ -1,131 +0,0 @@ -# Copyright (c) Microsoft Corporation. -# Licensed under the MIT License. - -"""Real Pydantic AI parallel tools with ordinary activities and no trampoline. - -This small proof adapter retains model/tool registration and JSON codecs. It is -not a replacement for the framework adapter's full serialization/error policy. -""" - -import asyncio -import json -from contextvars import ContextVar -from datetime import timezone -from typing import Any - -import pytest - -from durabletask import task -from durabletask.internal import helpers -from durabletask.internal import orchestrator_service_pb2 as pb -from tests.durabletask.test_coroutine_orchestrators import _completion, _execute, _start - - -@pytest.mark.parametrize("fail_tool", [False, True]) -def test_pydantic_agent_parallel_tools_replay_without_bridge(fail_tool: bool) -> None: - pytest.importorskip("pydantic_ai", reason="The framework requires Python 3.11+") - from pydantic import TypeAdapter - from pydantic_ai import Agent - from pydantic_ai.capabilities import AbstractCapability - from pydantic_ai.messages import ModelMessage, ModelResponse - from pydantic_ai.models import ModelRequestParameters - from pydantic_ai.models.test import TestModel - from pydantic_ai.settings import ModelSettings - - context: ContextVar[task.OrchestrationContext] = ContextVar("context") - request_codec = TypeAdapter(tuple[list[ModelMessage], ModelSettings | None, ModelRequestParameters]) - response_codec = TypeAdapter(ModelResponse) - invoked: list[str] = [] - - class DurableModel(TestModel): - async def request(self, messages: Any, model_settings: Any, model_request_parameters: Any) -> ModelResponse: - payload = request_codec.dump_python((messages, model_settings, model_request_parameters), mode="json") - # This proof adapter explicitly canonicalizes framework message - # metadata; the SDK must not intercept wall-clock calls globally. - stamp = context.get().current_utc_datetime.replace(tzinfo=timezone.utc).isoformat() - for message in payload[0]: - message["timestamp"] = stamp - for part in message["parts"]: - if "timestamp" in part: - part["timestamp"] = stamp - result = await context.get().call_activity("model", input=payload) - return response_codec.validate_python(result) - - async def model(payload: Any) -> Any: - messages, settings, parameters = request_codec.validate_python(payload) - result = await TestModel(custom_output_text="done").request(messages, settings, parameters) - return response_codec.dump_python(result, mode="json") - - async def first_activity(value: int) -> str: - if fail_tool: - raise RuntimeError("recorded tool failure") - return f"first:{value}" - - async def second_activity(value: int) -> str: - return f"second:{value}" - - activities = {"model": model, "first": first_activity, "second": second_activity} - - class ReplayClock(AbstractCapability[None]): - def _run_clock(self) -> Any: - return lambda: context.get().current_utc_datetime.replace(tzinfo=timezone.utc) - - agent = Agent(DurableModel(), name="proof", capabilities=[ReplayClock()]) - - @agent.tool_plain - async def first(value: int) -> str: - return await context.get().call_activity("first", input=value, return_type=str) - - @agent.tool_plain - async def second(value: int) -> str: - return await context.get().call_activity("second", input=value, return_type=str) - - async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> str: - token = context.set(ctx) - try: - result = await agent.run( - "Call both tools.", run_id=str(ctx.new_uuid()), conversation_id=ctx.instance_id) - return result.output - finally: - context.reset(token) - - history = _start() - result = _execute(orchestrator, [], history) - batches: list[list[str]] = [] - while not any(action.HasField("completeOrchestration") for action in result.actions): - assert result.actions - scheduled = list(result.actions) - batches.append([action.scheduleTask.name for action in scheduled]) - events = [] - for action in scheduled: - assert action.HasField("scheduleTask") - name = action.scheduleTask.name - history.append(helpers.new_task_scheduled_event(action.id, name)) - payload = json.loads(action.scheduleTask.input.value) - invoked.append(name) - try: - output = asyncio.run(activities[name](payload)) - except RuntimeError as error: - events.append(helpers.new_task_failed_event(action.id, error)) - else: - events.append(helpers.new_task_completed_event(action.id, json.dumps(output))) - # Replay twice with identical recorded results: no model/tool body reruns. - before = list(invoked) - first_result = _execute(orchestrator, history, events) - second_result = _execute(orchestrator, history, events) - assert first_result.actions == second_result.actions - assert invoked == before - history += events - result = first_result - assert len(batches) < 10 - complete = _completion(result) - if fail_tool: - assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED - assert "recorded tool failure" in complete.failureDetails.errorMessage - assert batches == [["model"], ["first", "second"]] - assert invoked == ["model", "first", "second"] - return - assert complete.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED, complete.failureDetails - assert json.loads(complete.result.value) == "done" - assert batches == [["model"], ["first", "second"], ["model"]] - assert invoked == ["model", "first", "second", "model"]