Reject unsupported asynchronous workflow definitions and results

docs/sdk-reference.md

@@ -405,6 +405,12 @@ language, target service language, operation name, request payload,
 service-call id, response or failure surface, and optional artifact metadata
 to the recorded result.
 
+Workflow entry points use synchronous `def run(...)` methods, either returning
+a result directly or yielding durable commands. `async def` and async-generator
+entry points are rejected with `InvalidWorkflowDefinition` during definition or
+worker registration. Async activity handlers remain supported. Direct replay
+also rejects async entry points and awaitable results.
+
 ## Workflow signals, queries, and updates
 
 Signals mutate workflow state during replay:

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.5.1"
+version = "2.5.2"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.5.1"
-registry-version = "2.5.1"
+product-train = "2.5.2"
+registry-version = "2.5.2"
 supported-server-versions = "2.5.16"
 worker-protocol-version = "1.19"
 control-plane-version = "2"

src/durable_workflow/__init__.py

@@ -81,6 +81,7 @@
     ExternalPayloadUnavailable,
     ExternalPayloadUnsupported,
     InvalidArgument,
+    InvalidWorkflowDefinition,
     NamespaceNotFound,
     NexusOperationFailed,
     NonDeterministicReplayError,
@@ -369,6 +370,7 @@
     "NexusOperationResult",
     "NexusServiceCall",
     "NonDeterministicReplayError",
+    "InvalidWorkflowDefinition",
     "NoopMetrics",
     "QueryFailed",
     "RuntimeCapabilityUnsupported",

src/durable_workflow/errors.py

@@ -27,6 +27,10 @@ class DurableWorkflowError(Exception):
     """Base class for every exception raised by the SDK."""
 
 
+class InvalidWorkflowDefinition(DurableWorkflowError, ValueError):
+    """Workflow authoring uses an unsupported entry point or result."""
+
+
 class DurableOperationCancelled(DurableWorkflowError):
     """A durable selection member was explicitly cancelled."""
 

src/durable_workflow/testing.py

@@ -51,6 +51,7 @@ def test_my_workflow():
     StartChildWorkflow,
     StartTimer,
     UpsertSearchAttributes,
+    _validate_workflow_entrypoint,
     registry,
     replay,
 )
@@ -134,6 +135,7 @@ def register_workflow(self, workflow_cls: type) -> None:
                 f"{workflow_cls!r} is not a workflow class; "
                 "decorate it with @workflow.defn(name=...) first."
             )
+        _validate_workflow_entrypoint(workflow_cls)
         self._registered_workflows[name] = workflow_cls
 
     def signal(

src/durable_workflow/worker.py

@@ -104,6 +104,7 @@
     ScheduleActivity,
     StartChildWorkflow,
     UpsertMemo,
+    _validate_workflow_entrypoint,
     apply_update,
     commands_to_server_commands,
     query_state,
@@ -1023,7 +1024,10 @@ def __init__(
     ) -> None:
         self.client = client
         self.task_queue = task_queue
-        self.workflows = {_workflow_name(w): w for w in workflows}
+        self.workflows: dict[str, type] = {}
+        for definition in workflows:
+            _validate_workflow_entrypoint(definition)
+            self.workflows[_workflow_name(definition)] = definition
         self.workflow_definition_fingerprints = {
             workflow_type: _workflow_definition_fingerprint(workflow_cls)
             for workflow_type, workflow_cls in self.workflows.items()

src/durable_workflow/workflow.py

@@ -21,6 +21,7 @@
 import base64
 import contextlib
 import hashlib
+import inspect
 import logging
 import math
 import random
@@ -58,6 +59,7 @@
     ChildWorkflowFailed,
     ChildWorkflowTerminated,
     DurableOperationCancelled,
+    InvalidWorkflowDefinition,
     NexusOperationFailed,
     NonDeterministicReplayError,
     QueryFailed,
@@ -97,6 +99,31 @@ def _authored_scope_cleanup() -> dict[str, str] | None:
 MESSAGE_STREAM_MAX_BATCH = 100
 
 
+def _validate_workflow_entrypoint(cls: type) -> None:
+    entrypoint = getattr(cls, "run", None)
+    if not callable(entrypoint):
+        return
+    targets = [inspect.unwrap(entrypoint)]
+    if not inspect.isroutine(targets[0]):
+        targets.append(inspect.unwrap(targets[0].__call__))
+    if any(inspect.iscoroutinefunction(target) or inspect.isasyncgenfunction(target) for target in targets):
+        name = getattr(cls, "__workflow_name__", cls.__name__)
+        raise InvalidWorkflowDefinition(
+            f"unsupported_workflow_entrypoint: workflow {name!r} ({cls.__qualname__}.run) "
+            "must use a synchronous function or generator; yield durable commands instead of async def."
+        )
+
+
+def _validate_workflow_result(value: Any, cls: type) -> None:
+    if inspect.isawaitable(value) or inspect.isasyncgen(value):
+        if inspect.iscoroutine(value):
+            value.close()
+        raise InvalidWorkflowDefinition(
+            f"unsupported_workflow_result: {cls.__qualname__}.run returned an async value; "
+            "use a synchronous result or yield durable commands."
+        )
+
+
 def defn(*, name: str) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]:
     """Register a class as a workflow type under a language-neutral name.
 
@@ -108,6 +135,7 @@ def defn(*, name: str) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]:
 
     def wrap(cls: type[_WorkflowT]) -> type[_WorkflowT]:
         cls.__workflow_name__ = name  # type: ignore[attr-defined]
+        _validate_workflow_entrypoint(cls)
         signals: dict[str, str] = {}
         queries: dict[str, str] = {}
         updates: dict[str, str] = {}
@@ -4037,6 +4065,7 @@ def _replay_state(
     allow_cancellation_scope_authoring: bool = False,
     allow_cancellation_scope_delivery: bool = False,
 ) -> _ReplayState:
+    _validate_workflow_entrypoint(workflow_cls)
     if allow_cancellation_scope_delivery and not allow_cancellation_scope_authoring:
         raise LocalActivityExecutionAborted("scope delivery requires canonical scope authoring")
     if prepare_local_activity_groups and not prepare_local_activities:
@@ -5172,6 +5201,7 @@ def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition)
     except Exception as exc:
         return _state([_fail_workflow_from_exception(exc)])
     if not hasattr(gen, "__next__"):
+        _validate_workflow_result(gen, workflow_cls)
         if isinstance(gen, ContinueAsNew):
             _assert_no_unconsumed_history("continue as new")
             return _state([gen])
@@ -6032,6 +6062,7 @@ def _scope_call_membership(command: Any) -> str:
         return next(iter(memberships))
 
     def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState:
+        _validate_workflow_result(value, workflow_cls)
         _apply_due_receivers()
         _consume_terminal_condition_reopens()
         commands = list(pending) if include_pending else []
@@ -6784,10 +6815,10 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
         try:
             return _terminal_state(stop.value, include_pending=True)
         except Exception as exc:
-            if isinstance(exc, NonDeterministicReplayError):
+            if isinstance(exc, (NonDeterministicReplayError, InvalidWorkflowDefinition)):
                 raise
             return _state(pending + [_fail_workflow_from_exception(exc)])
-    except NonDeterministicReplayError:
+    except (NonDeterministicReplayError, InvalidWorkflowDefinition):
         raise
     except LocalActivityExecutionAborted:
         raise

tests/fixtures/replay_regressions/unsupported-async-workflow-entrypoint.json

@@ -0,0 +1,10 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "unsupported-async-workflow-entrypoint",
+  "protocol_version": "1.19",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.replay.unsupported-async", "input": [], "payload_codec": "avro"},
+  "history": [{"event_type": "WorkflowStarted", "payload": {}}],
+  "expected": {"error": "unsupported_workflow_entrypoint"}
+}

tests/test_replay_regression_corpus.py

@@ -251,7 +251,15 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         return (yield ctx.start_child_workflow("child", []))
 
 
+class UnsupportedAsyncWorkflow:
+    __workflow_name__ = "tests.replay.unsupported-async"
+
+    async def run(self, ctx: WorkflowContext) -> str:
+        return "body was never executed"
+
+
 WORKFLOWS = [
+    UnsupportedAsyncWorkflow,
     ChildPolicyAuthorWorkflow,
     ColdReplacementSatisfiedConditionWorkflow,
     CooperativeReopenedConditionCleanupWorkflow,

tests/test_workflow_entrypoint_validation.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Generator
+from functools import partial, wraps
+from typing import Any
+
+import pytest
+
+from durable_workflow import Client, Worker, workflow
+from durable_workflow.testing import WorkflowEnvironment
+
+
+class CoroutineWorkflow:
+    async def run(self, ctx: Any) -> str:
+        return "never executed"
+
+
+class AsyncGeneratorWorkflow:
+    async def run(self, ctx: Any) -> AsyncIterator[str]:
+        yield "unsupported"
+
+
+class AsyncCallable:
+    async def __call__(self, ctx: Any) -> str:
+        return "unsupported"
+
+
+class CallableWorkflow:
+    run = AsyncCallable()
+
+
+class PartialWorkflow:
+    run = staticmethod(partial(CoroutineWorkflow.run, object()))
+
+
+ASYNC_DEFINITIONS = [CoroutineWorkflow, AsyncGeneratorWorkflow, CallableWorkflow, PartialWorkflow]
+
+
+@pytest.mark.parametrize("definition", ASYNC_DEFINITIONS)
+def test_definition_rejects_async_entrypoints(definition: type) -> None:
+    with pytest.raises(ValueError, match="unsupported_workflow_entrypoint.*synchronous"):
+        workflow.defn(name="invalid-async")(definition)
+
+
+@pytest.mark.parametrize("definition", ASYNC_DEFINITIONS)
+def test_worker_rejects_undecorated_async_entrypoints_before_network(definition: type) -> None:
+    with pytest.raises(ValueError, match=definition.__name__):
+        Worker(Client("http://127.0.0.1:1"), task_queue="unused", workflows=[definition])
+
+
+@pytest.mark.parametrize("definition", ASYNC_DEFINITIONS)
+def test_direct_replay_rejects_async_without_creating_a_coroutine(definition: type) -> None:
+    with pytest.raises(ValueError, match="unsupported_workflow_entrypoint"):
+        workflow.replay(definition, [], [])
+
+
+def test_test_environment_revalidates_entrypoint_after_decoration() -> None:
+    @workflow.defn(name="changed-entrypoint")
+    class ChangedWorkflow:
+        def run(self, ctx: Any) -> str:
+            return "valid"
+
+    ChangedWorkflow.run = CoroutineWorkflow.run  # type: ignore[assignment]
+    with pytest.raises(ValueError, match="changed-entrypoint"):
+        WorkflowEnvironment().register_workflow(ChangedWorkflow)
+
+
+def test_wrapped_and_inherited_async_entrypoints_are_rejected() -> None:
+    class InheritedWorkflow(CoroutineWorkflow):
+        pass
+
+    class WrappedWorkflow:
+        @wraps(CoroutineWorkflow.run)
+        def run(self, ctx: Any) -> Any:
+            return CoroutineWorkflow.run(self, ctx)
+
+    for definition in [InheritedWorkflow, WrappedWorkflow]:
+        with pytest.raises(ValueError, match="unsupported_workflow_entrypoint"):
+            workflow.replay(definition, [], [])
+
+
+@pytest.mark.parametrize("generator", [False, True])
+def test_runtime_awaitable_result_is_not_completed(generator: bool) -> None:
+    class AwaitableResultWorkflow:
+        def run(self, ctx: Any) -> Any:
+            return CoroutineWorkflow.run(self, ctx)
+
+    class GeneratorAwaitableResultWorkflow:
+        def run(self, ctx: Any) -> Generator[Any, Any, Any]:
+            if False:
+                yield None
+            return CoroutineWorkflow.run(self, ctx)
+
+    definition = GeneratorAwaitableResultWorkflow if generator else AwaitableResultWorkflow
+    with pytest.raises(ValueError, match="unsupported_workflow_result"):
+        workflow.replay(definition, [], [])
+
+
+def test_synchronous_and_generator_workflows_remain_supported() -> None:
+    @workflow.defn(name="valid-sync")
+    class SynchronousWorkflow:
+        def run(self, ctx: Any) -> str:
+            return "done"
+
+    @workflow.defn(name="valid-generator")
+    class GeneratorWorkflow:
+        def run(self, ctx: Any) -> Generator[Any, Any, str]:
+            if False:
+                yield None
+            return "done"
+
+    Worker(Client("http://127.0.0.1:1"), task_queue="unused", workflows=[SynchronousWorkflow, GeneratorWorkflow])
+    environment = WorkflowEnvironment()
+    for definition in [SynchronousWorkflow, GeneratorWorkflow]:
+        assert environment.execute_workflow(definition) == "done"
+        assert workflow.replay(definition, [], []).commands[0].result == "done"  # type: ignore[union-attr]