initial asyncio implementation after durable task asyncio work

dev-requirements.txt

@@ -4,6 +4,7 @@ mypy-protobuf>=2.9
 flake8>=3.7.9
 tox>=4.3.0
 coverage>=5.3
+pytest
 wheel
 # used in unit test only
 opentelemetry-sdk

examples/workflow-async/README.md

@@ -0,0 +1,15 @@
+# Dapr Workflow Async Examples (Python)
+
+These examples mirror `examples/workflow/` but author orchestrators with `async def` using the
+async workflow APIs. Activities remain regular functions unless noted.
+
+How to run:
+- Ensure a Dapr sidecar is running locally. If needed, set `DURABLETASK_GRPC_ENDPOINT`, or
+  `DURABLETASK_GRPC_HOST/PORT`.
+- Install requirements: `pip install -r requirements.txt`
+- Run any example: `python simple.py`
+
+Notes:
+- Orchestrators use `await ctx.activity(...)`, `await ctx.sleep(...)`, `await ctx.when_all/when_any(...)`, etc.
+- No event loop is started manually; the Durable Task worker drives the async orchestrators.
+- You can also launch instances using `DaprWorkflowClient` as in the non-async examples.

examples/workflow-async/child_workflow.py

@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowRuntime,
+)
+
+wfr = WorkflowRuntime()
+
+
+@wfr.async_workflow(name='child_async')
+async def child(ctx: AsyncWorkflowContext, n: int) -> int:
+    return n * 2
+
+
+@wfr.async_workflow(name='parent_async')
+async def parent(ctx: AsyncWorkflowContext, n: int) -> int:
+    r = await ctx.call_child_workflow(child, input=n)
+    print(f'Child workflow returned {r}')
+    return r + 1
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'parent_async_instance'
+    client.schedule_new_workflow(workflow=parent, input=5, instance_id=instance_id)
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/fan_out_fan_in.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowActivityContext,
+    WorkflowRuntime,
+)
+
+wfr = WorkflowRuntime()
+
+
+@wfr.activity(name='square')
+def square(ctx: WorkflowActivityContext, x: int) -> int:
+    return x * x
+
+
+@wfr.async_workflow(name='fan_out_fan_in_async')
+async def orchestrator(ctx: AsyncWorkflowContext):
+    tasks = [ctx.call_activity(square, input=i) for i in range(1, 6)]
+    results = await ctx.when_all(tasks)
+    total = sum(results)
+    return total
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'fofi_async'
+    client.schedule_new_workflow(workflow=orchestrator, instance_id=instance_id)
+    wf_state = client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    print(f'Workflow state: {wf_state}')
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/human_approval.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+from dapr.ext.workflow import AsyncWorkflowContext, DaprWorkflowClient, WorkflowRuntime
+
+wfr = WorkflowRuntime()
+
+
+@wfr.async_workflow(name='human_approval_async')
+async def orchestrator(ctx: AsyncWorkflowContext, request_id: str):
+    decision = await ctx.when_any([
+        ctx.wait_for_external_event(f'approve:{request_id}'),
+        ctx.wait_for_external_event(f'reject:{request_id}'),
+        ctx.create_timer(300.0),
+    ])
+    if isinstance(decision, dict) and decision.get('approved'):
+        return 'APPROVED'
+    if isinstance(decision, dict) and decision.get('rejected'):
+        return 'REJECTED'
+    return 'TIMEOUT'
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'human_approval_async_1'
+    client.schedule_new_workflow(workflow=orchestrator, input='REQ-1', instance_id=instance_id)
+    # In a real scenario, raise approve/reject event from another service.
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/requirements.txt

@@ -0,0 +1,2 @@
+dapr-ext-workflow-dev>=1.15.0.dev
+dapr-dev>=1.15.0.dev

examples/workflow-async/simple.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+from datetime import timedelta
+from time import sleep
+
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    RetryPolicy,
+    WorkflowActivityContext,
+    WorkflowRuntime,
+)
+
+counter = 0
+retry_count = 0
+child_orchestrator_string = ''
+instance_id = 'asyncExampleInstanceID'
+child_instance_id = 'asyncChildInstanceID'
+workflow_name = 'async_hello_world_wf'
+child_workflow_name = 'async_child_wf'
+input_data = 'Hi Async Counter!'
+event_name = 'event1'
+event_data = 'eventData'
+
+retry_policy = RetryPolicy(
+    first_retry_interval=timedelta(seconds=1),
+    max_number_of_attempts=3,
+    backoff_coefficient=2,
+    max_retry_interval=timedelta(seconds=10),
+    retry_timeout=timedelta(seconds=100),
+)
+
+wfr = WorkflowRuntime()
+
+
+@wfr.async_workflow(name=workflow_name)
+async def hello_world_wf(ctx: AsyncWorkflowContext, wf_input):
+    global counter
+    # activities
+    result_1 = await ctx.call_activity(hello_act, input=1)
+    print(f'Activity 1 returned {result_1}')
+    result_2 = await ctx.call_activity(hello_act, input=10)
+    print(f'Activity 2 returned {result_2}')
+    result_3 = await ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)
+    print(f'Activity 3 returned {result_3}')
+    result_4 = await ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)
+    print(f'Child workflow returned {result_4}')
+
+    # Event vs timeout using when_any
+    first = await ctx.when_any([
+        ctx.wait_for_external_event(event_name),
+        ctx.create_timer(timedelta(seconds=30)),
+    ])
+
+    # Proceed only if event won
+    if isinstance(first, dict) and 'event' in first:
+        await ctx.call_activity(hello_act, input=100)
+        await ctx.call_activity(hello_act, input=1000)
+        return 'Completed'
+    return 'Timeout'
+
+
+@wfr.activity(name='async_hello_act')
+def hello_act(ctx: WorkflowActivityContext, wf_input):
+    global counter
+    counter += wf_input
+    return f'Activity returned {wf_input}'
+
+
+@wfr.activity(name='async_hello_retryable_act')
+def hello_retryable_act(ctx: WorkflowActivityContext):
+    global retry_count
+    if (retry_count % 2) == 0:
+        retry_count += 1
+        raise ValueError('Retryable Error')
+    retry_count += 1
+    return f'Activity returned {retry_count}'
+
+
+@wfr.async_workflow(name=child_workflow_name)
+async def child_retryable_wf(ctx: AsyncWorkflowContext):
+    global child_orchestrator_string
+    # Call activity with retry and simulate retryable workflow failure until certain state
+    child_activity_result = await ctx.call_activity(act_for_child_wf, input='x', retry_policy=retry_policy)
+    print(f'Child activity returned {child_activity_result}')
+    # In a real sample, you might check state and raise to trigger retry
+    return 'ok'
+
+
+@wfr.activity(name='async_act_for_child_wf')
+def act_for_child_wf(ctx: WorkflowActivityContext, inp):
+    global child_orchestrator_string
+    child_orchestrator_string += inp
+
+
+def main():
+    wfr.start()
+    wf_client = DaprWorkflowClient()
+
+    wf_client.schedule_new_workflow(
+        workflow=hello_world_wf, input=input_data, instance_id=instance_id
+    )
+
+    wf_client.wait_for_workflow_start(instance_id)
+
+    # Let initial activities run
+    sleep(5)
+
+    # Raise event to continue
+    wf_client.raise_workflow_event(instance_id=instance_id, event_name=event_name, data={'ok': True})
+
+    # Wait for completion
+    state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    print(f'Workflow status: {state.runtime_status.name}')
+
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

examples/workflow-async/task_chaining.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the specific language governing permissions and
+limitations under the License.
+"""
+from dapr.ext.workflow import (
+    AsyncWorkflowContext,
+    DaprWorkflowClient,
+    WorkflowActivityContext,
+    WorkflowRuntime,
+)
+
+wfr = WorkflowRuntime()
+
+
+@wfr.activity(name='sum')
+def sum_act(ctx: WorkflowActivityContext, nums):
+    return sum(nums)
+
+
+@wfr.async_workflow(name='task_chaining_async')
+async def orchestrator(ctx: AsyncWorkflowContext):
+    a = await ctx.call_activity(sum_act, input=[1, 2])
+    b = await ctx.call_activity(sum_act, input=[a, 3])
+    c = await ctx.call_activity(sum_act, input=[b, 4])
+    return c
+
+
+def main():
+    wfr.start()
+    client = DaprWorkflowClient()
+    instance_id = 'task_chain_async'
+    client.schedule_new_workflow(workflow=orchestrator, instance_id=instance_id)
+    client.wait_for_workflow_completion(instance_id, timeout_in_seconds=60)
+    wfr.shutdown()
+
+
+if __name__ == '__main__':
+    main()

ext/dapr-ext-workflow/README.rst

@@ -16,6 +16,41 @@ Installation
 
     pip install dapr-ext-workflow
 
+Async authoring (experimental)
+------------------------------
+
+This package supports authoring workflows with ``async def`` in addition to the existing generator-based orchestrators.
+
+- Register async workflows using ``WorkflowRuntime.async_workflow`` or ``register_async_workflow``.
+- Use ``AsyncWorkflowContext`` for deterministic operations:
+
+  - Activities: ``await ctx.activity(activity_fn, input=...)``
+  - Sub-orchestrators: ``await ctx.sub_orchestrator(workflow_fn, input=...)``
+  - Timers: ``await ctx.sleep(seconds|timedelta)``
+  - External events: ``await ctx.wait_for_external_event(name)``
+  - Concurrency: ``await ctx.when_all([...])``, ``await ctx.when_any([...])``
+  - Deterministic utils: ``ctx.now()``, ``ctx.random()``, ``ctx.uuid4()``
+
+Best-effort sandbox
+~~~~~~~~~~~~~~~~~~~
+
+Opt-in scoped compatibility mode maps ``asyncio.sleep``, ``random``, ``uuid.uuid4``, and ``time.time`` to deterministic equivalents during workflow execution. Use ``sandbox_mode="best_effort"`` or ``"strict"`` when registering async workflows. Strict mode blocks ``asyncio.create_task`` in orchestrators.
+
+Examples
+~~~~~~~~
+
+See ``ext/dapr-ext-workflow/examples`` for:
+
+- ``async_activity_sequence.py``
+- ``async_external_event.py``
+- ``async_sub_orchestrator.py``
+
+Determinism and semantics
+~~~~~~~~~~~~~~~~~~~~~~~~~
+
+- ``when_any`` losers: the first-completer result is returned; non-winning awaitables are ignored deterministically (no additional commands are emitted by the orchestrator for cancellation). This ensures replay stability. Integration behavior with the sidecar is subject to the Durable Task scheduler; the orchestrator does not actively cancel losers.
+- Suspension and termination: when an instance is suspended, only new external events are buffered while replay continues to reconstruct state; async orchestrators can inspect ``ctx.is_suspended`` if exposed by the runtime. Termination completes the orchestrator with TERMINATED status and does not raise into the coroutine. End-to-end confirmation requires running against a sidecar; unit tests in this repo do not start a sidecar.
+
 References
 ----------