Validate unsupported workflow handlers before execution

CHANGELOG.md

@@ -14,6 +14,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `deprecate_patch` for the same change ID, while validating each call's range
   and preserving retained duplicate-marker sequences.
 
+## [2.6.2] - 2026-10-20
+
+### Fixed
+
+- Preserve workflow entry argument-binding refusals before application code
+  runs, with truthful Worker capability discovery for compatible deployment
+  routing. Workflow body exceptions remain ordinary workflow failures.
+- Run synchronous remote activities outside the asyncio event loop so other
+  workflow tasks, polls and heartbeats can progress.
+- Reject unsupported coroutine signal/update handlers and generator handlers
+  before registration or replay, including inherited and wrapped handlers.
+  Guard returned values without removing the Worker's existing asynchronous
+  query and update-validator support. Local synchronous query/validation helpers
+  refuse and close returned coroutines.
+
 ## [2.6.1] - 2026-10-19
 
 ### Fixed

docs/reference/workflow.md

@@ -1,5 +1,32 @@
 # Workflow
 
+## Signal, update and query handlers
+
+Handlers decorated with `@workflow.signal` or `@workflow.update` must be
+synchronous functions. Coroutine
+(`async def`), async-generator and generator handlers are rejected with
+`InvalidWorkflowDefinition` when the workflow is defined, registered or replayed,
+including inherited handlers. Returning an asynchronous or generator value from
+a synchronous wrapper is also refused.
+
+Queries and update validators may be synchronous or `async def`. The Worker
+awaits their results; queries retain their original deadline and cancellation
+authority. The local synchronous `query_state()` and `validate_update()` helpers
+refuse asynchronous results with `QueryFailed` and `UpdateRejected` respectively
+and close returned coroutines. Generator and async-generator query or validator
+handlers are rejected at definition or registration.
+
+Each replay constructs the workflow instance before executing `run()`. Signals
+already present at workflow start are applied after `run()` reaches its first
+yield, so constructor and pre-yield initialization happen before those handlers.
+Signal and update handlers mutate that replayed state; queries inspect it and
+update validators check a request before acceptance. Put durable activities,
+timers and child workflow commands in `run()`, not in a handler.
+
+An exception from a signal handler while replaying fails the workflow run.
+Update-handler exceptions fail the update, validator exceptions reject it,
+and query-handler exceptions fail the query. They do not silently succeed.
+
 Workflow retry and timeout settings are durable command budgets. Use
 `ActivityRetryPolicy` on `ctx.schedule_activity(...)` for activity attempts and
 `ChildWorkflowRetryPolicy` on `ctx.start_child_workflow(...)` for child workflow

src/durable_workflow/worker.py

@@ -109,12 +109,13 @@
     StartChildWorkflow,
     UpsertMemo,
     _prepare_query_state,
+    _prepare_update_validation,
+    _UpdateValidationReplayOptions,
     _validate_workflow_entrypoint,
     _workflow_handler_registries,
     apply_update,
     commands_to_server_commands,
     replay,
-    validate_update,
 )
 
 log = logging.getLogger("durable_workflow.worker")
@@ -3867,21 +3868,24 @@ async def _run_update_validation_task(self, task: dict[str, Any]) -> str:
                 codec=codec,
                 context="workflow update validation arguments",
             )
-            result = validate_update(
+            invoke = _prepare_update_validation(
                 workflow_cls,
                 history,
                 start_input,
                 update_name,
                 update_args,
-                workflow_id=task.get("workflow_id"),
-                run_id=task.get("run_id", ""),
-                payload_codec=codec,
-                external_storage=self.external_storage,
-                external_storage_cache=self.external_storage_cache,
-                prepare_local_activities=self._prepared_local_activities_supported,
-                prepare_local_activity_groups=self._prepared_local_activity_groups_supported,
-                local_activity_cancellation_policies=self._local_activity_cancellation_policies,
+                replay_options=_UpdateValidationReplayOptions(
+                    workflow_id=task.get("workflow_id"),
+                    run_id=task.get("run_id", ""),
+                    payload_codec=codec,
+                    external_storage=self.external_storage,
+                    external_storage_cache=self.external_storage_cache,
+                    prepare_local_activities=self._prepared_local_activities_supported,
+                    prepare_local_activity_groups=self._prepared_local_activity_groups_supported,
+                    local_activity_cancellation_policies=self._local_activity_cancellation_policies,
+                ),
             )
+            result = invoke()
             if inspect.isawaitable(result):
                 result = await result
             if result is False:

src/durable_workflow/workflow.py

@@ -138,6 +138,21 @@ def _workflow_handler_registries(
             wire_name = getattr(member, f"__{kind}_name__", None)
             if not isinstance(wire_name, str) or not wire_name:
                 continue
+            if not callable(member):
+                raise InvalidWorkflowDefinition(f"workflow {cls.__qualname__}.{attr} handler must be callable")
+            target = inspect.unwrap(member)
+            if not inspect.isroutine(target) and callable(target):
+                target = inspect.unwrap(target.__call__)
+            if (
+                (inspect.iscoroutinefunction(target) and kind not in {"query", "update_validator"})
+                or inspect.isasyncgenfunction(target)
+                or inspect.isgeneratorfunction(target)
+            ):
+                name = getattr(cls, "__workflow_name__", cls.__name__)
+                raise InvalidWorkflowDefinition(
+                    f"unsupported_workflow_handler: workflow {name!r} ({cls.__qualname__}.{attr}) "
+                    f"{kind} handler {wire_name!r} must be synchronous; yield durable commands from run()."
+                )
             if kind == "signal" and wire_name == MESSAGE_STREAM_SIGNAL:
                 raise ValueError(f"signal name {wire_name!r} is reserved by the workflow runtime")
             if wire_name in registry:
@@ -149,6 +164,28 @@ def _workflow_handler_registries(
     return handlers["signal"], handlers["query"], handlers["update"], handlers["update_validator"]
 
 
+def _invoke_workflow_handler(
+    handler: Callable[..., Any], args: list[Any], cls: type, method: str, *, allow_async: bool = False,
+) -> Any:
+    return _validate_workflow_handler_result(handler(*args), cls, method, allow_async=allow_async)
+
+
+def _validate_workflow_handler_result(result: Any, cls: type, method: str, *, allow_async: bool = False) -> Any:
+    if (
+        (inspect.isawaitable(result) and not allow_async)
+        or inspect.isasyncgen(result)
+        or inspect.isgenerator(result)
+    ):
+        if inspect.iscoroutine(result) or inspect.isgenerator(result):
+            result.close()
+        name = getattr(cls, "__workflow_name__", cls.__name__)
+        raise InvalidWorkflowDefinition(
+            f"unsupported_workflow_handler_result: workflow {name!r} ({cls.__qualname__}.{method}) "
+            "returned an async or generator value; handlers must return synchronous values."
+        )
+    return result
+
+
 def defn(*, name: str) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]:
     """Register a class as a workflow type under a language-neutral name.
 
@@ -3133,7 +3170,9 @@ def _prepare_query_state(
 
     def invoke() -> Any:
         try:
-            return handler(*(list(args) if args is not None else []))
+            return _invoke_workflow_handler(
+                handler, list(args) if args is not None else [], workflow_cls, method_name, allow_async=True,
+            )
         except QueryFailed:
             raise
         except Exception as exc:
@@ -3163,14 +3202,19 @@ def query_state(
     Server-routed Worker execution separately checks its original query claim
     between replay and handler invocation. Direct local use has no Server lease.
     """
-    return _prepare_query_state(
+    result = _prepare_query_state(
         workflow_cls, history_events, start_input, query_name, args,
         workflow_id=workflow_id, run_id=run_id, payload_codec=payload_codec,
         external_storage=external_storage, external_storage_cache=external_storage_cache,
         prepare_local_activities=prepare_local_activities,
         prepare_local_activity_groups=prepare_local_activity_groups,
         local_activity_cancellation_policies=local_activity_cancellation_policies,
     )()
+    try:
+        registry: dict[str, str] = getattr(workflow_cls, "__workflow_queries__", {}) or {}
+        return _validate_workflow_handler_result(result, workflow_cls, registry.get(query_name, query_name))
+    except InvalidWorkflowDefinition as exc:
+        raise QueryFailed(str(exc)) from exc
 
 
 def apply_update(
@@ -3276,11 +3320,24 @@ def apply_update(
         )
 
     try:
-        return CompleteUpdate(update_id=update_id, result=handler(*args))
+        result = _invoke_workflow_handler(handler, args, workflow_cls, method_name)
+        return CompleteUpdate(update_id=update_id, result=result)
     except Exception as exc:
         return _fail_update_from_exception(update_id, "update handler failed", exc)
 
 
+@dataclass(frozen=True)
+class _UpdateValidationReplayOptions:
+    workflow_id: str | None = None
+    run_id: str = ""
+    payload_codec: str | None = None
+    external_storage: ExternalStorageDriver | None = None
+    external_storage_cache: ExternalPayloadCache | None = None
+    prepare_local_activities: bool = False
+    prepare_local_activity_groups: bool = False
+    local_activity_cancellation_policies: tuple[str, ...] = ()
+
+
 def validate_update(
     workflow_cls: type,
     history_events: Iterable[dict[str, Any]],
@@ -3297,28 +3354,57 @@ def validate_update(
     prepare_local_activity_groups: bool = False,
     local_activity_cancellation_policies: tuple[str, ...] = (),
 ) -> Any:
+    """Replay and invoke a validator synchronously, refusing asynchronous results."""
+    from .errors import UpdateRejected
+
+    result = _prepare_update_validation(
+        workflow_cls, history_events, start_input, update_name, args,
+        replay_options=_UpdateValidationReplayOptions(
+            workflow_id=workflow_id, run_id=run_id, payload_codec=payload_codec,
+            external_storage=external_storage, external_storage_cache=external_storage_cache,
+            prepare_local_activities=prepare_local_activities,
+            prepare_local_activity_groups=prepare_local_activity_groups,
+            local_activity_cancellation_policies=local_activity_cancellation_policies,
+        ),
+    )()
+    try:
+        registry: dict[str, str] = getattr(workflow_cls, "__workflow_update_validators__", {}) or {}
+        return _validate_workflow_handler_result(result, workflow_cls, registry.get(update_name, update_name))
+    except InvalidWorkflowDefinition as exc:
+        raise UpdateRejected(str(exc), exception_type=type(exc).__name__) from exc
+
+
+def _prepare_update_validation(
+    workflow_cls: type,
+    history_events: Iterable[dict[str, Any]],
+    start_input: list[Any],
+    update_name: str,
+    args: list[Any],
+    *,
+    replay_options: _UpdateValidationReplayOptions,
+) -> Callable[[], Any]:
     """Replay state and invoke only the declared pre-accept update validator.
 
     The replayed instance is discarded after validation. This helper does not
     run the update handler and does not emit workflow commands, so the server
     remains the sole authority that can cross the accepted-state boundary.
     """
-    from .errors import InvalidArgument, UpdateRejected, UpdateValidationFailed
+    from .errors import UpdateValidationFailed
 
     events = list(history_events)
     try:
         state = _replay_state(
             workflow_cls,
             events,
             start_input,
-            workflow_id=workflow_id,
-            run_id=run_id,
-            payload_codec=payload_codec,
-            external_storage=external_storage,
-            external_storage_cache=external_storage_cache,
-            prepare_local_activities=prepare_local_activities,
-            prepare_local_activity_groups=prepare_local_activity_groups,
-            local_activity_cancellation_policies=local_activity_cancellation_policies,
+            workflow_id=replay_options.workflow_id,
+            run_id=replay_options.run_id,
+            payload_codec=replay_options.payload_codec,
+            external_storage=replay_options.external_storage,
+            external_storage_cache=replay_options.external_storage_cache,
+            prepare_local_activities=replay_options.prepare_local_activities,
+            prepare_local_activity_groups=replay_options.prepare_local_activity_groups,
+            local_activity_cancellation_policies=replay_options.local_activity_cancellation_policies,
         )
     except Exception as exc:
         raise UpdateValidationFailed(
@@ -3349,8 +3435,20 @@ def validate_update(
     if validator is None:
         raise UpdateValidationFailed(f"update validator {update_name!r} is not available")
 
+    return lambda: _invoke_update_validator(validator, args, workflow_cls, validator_name, update_name)
+
+
+def _invoke_update_validator(
+    validator: Callable[..., Any],
+    args: list[Any],
+    workflow_cls: type,
+    validator_name: str,
+    update_name: str,
+) -> Any:
+    from .errors import InvalidArgument, UpdateRejected
+
     try:
-        result = validator(*list(args))
+        result = _invoke_workflow_handler(validator, list(args), workflow_cls, validator_name, allow_async=True)
         if result is False:
             raise ValueError(f"update validator {update_name!r} returned false")
         return result
@@ -4240,6 +4338,7 @@ def _replay_state(
     allow_cancellation_scope_delivery: bool = False,
 ) -> _ReplayState:
     _validate_workflow_entrypoint(workflow_cls)
+    _workflow_handler_registries(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:
@@ -5285,7 +5384,7 @@ def _apply_receiver(receiver: _PendingReceiver) -> None:
                 return
             raise TypeError(f"update handler {receiver.name!r} is not available")
         ctx.logger._set_replaying(True)
-        handler(*receiver.args)
+        _invoke_workflow_handler(handler, receiver.args, workflow_cls, method_name)
 
     def _receiver_due(receiver: _PendingReceiver, *, before_consuming_result: bool) -> bool:
         if receiver.after_cancellation_delivery and not cancellation_consumed:

tests/fixtures/replay_regressions/unsupported-async-workflow-signal.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-signal",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.replay.unsupported-async-signal", "input": [], "payload_codec": "avro"},
+  "history": [{"event_type": "WorkflowStarted", "payload": {}}],
+  "expected": {"error": "unsupported_workflow_handler"}
+}

tests/test_replay_regression_corpus.py

@@ -287,6 +287,17 @@ async def run(self, ctx: WorkflowContext) -> str:
         return "body was never executed"
 
 
+class UnsupportedAsyncSignalWorkflow:
+    __workflow_name__ = "tests.replay.unsupported-async-signal"
+
+    @workflow.signal("change")
+    async def receive(self) -> None:
+        raise AssertionError("unsupported signal body must never run")
+
+    def run(self, ctx: WorkflowContext) -> str:
+        return "done"
+
+
 @workflow.defn(name="tests.replay.entry-arguments-required")
 class RequiredEntryArgumentWorkflow:
     def run(self, ctx: WorkflowContext, amount: int, required: int) -> int:
@@ -334,6 +345,7 @@ def run(self, ctx: WorkflowContext, cleanup_seconds: int = 0):  # type: ignore[n
     NexusTerminalDeliveryWorkflow,
     NexusTargetWorkflow,
     UnsupportedAsyncWorkflow,
+    UnsupportedAsyncSignalWorkflow,
     ChildPolicyAuthorWorkflow,
     ColdReplacementSatisfiedConditionWorkflow,
     CooperativeReopenedConditionCleanupWorkflow,

tests/test_worker.py

@@ -3805,7 +3805,7 @@ async def test_task_root_codec_rejection_precedes_decode_and_user_work(
         monkeypatch.setattr(worker_module, "replay", replay_spy)
         monkeypatch.setattr(worker_module, "_prepare_query_state", query_spy)
         monkeypatch.setattr(worker_module, "apply_update", update_spy)
-        monkeypatch.setattr(worker_module, "validate_update", validator_spy)
+        monkeypatch.setattr(worker_module, "_prepare_update_validation", validator_spy)
         monkeypatch.setattr(worker, "_execute_activity_callable", activity_spy)
 
         if task_kind == "activity":
@@ -3934,13 +3934,13 @@ async def test_exact_avro_task_completes_each_path_with_handler_work(
         replay_spy = Mock(wraps=worker_module.replay)
         query_spy = Mock(wraps=worker_module._prepare_query_state)
         update_spy = Mock(wraps=worker_module.apply_update)
-        validator_spy = Mock(wraps=worker_module.validate_update)
+        validator_spy = Mock(wraps=worker_module._prepare_update_validation)
         activity_spy = AsyncMock(wraps=worker._execute_activity_callable)
         monkeypatch.setattr(serializer, "decode_envelope", decode_spy)
         monkeypatch.setattr(worker_module, "replay", replay_spy)
         monkeypatch.setattr(worker_module, "_prepare_query_state", query_spy)
         monkeypatch.setattr(worker_module, "apply_update", update_spy)
-        monkeypatch.setattr(worker_module, "validate_update", validator_spy)
+        monkeypatch.setattr(worker_module, "_prepare_update_validation", validator_spy)
         monkeypatch.setattr(worker, "_execute_activity_callable", activity_spy)
 
         if task_kind == "activity":