Add opt-in typed workflow, child and client values (#184 slice 2)

CHANGELOG.md

@@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Opt-in typed workflow values (#184). `@workflow.defn(..., typed=True)` decodes the arguments of `run` (after `ctx`) and of every signal, query, update and update-validator handler to their annotations. `ctx.start_child_workflow(..., result_type=T)` decodes a child's result, and `WorkflowHandle.result(result_type=T)` / `Client.get_result(..., result_type=T)` decode a workflow's result on the client. Dataclass and `Enum` values in client workflow, signal, update, query and schedule arguments, child and continue-as-new arguments, and workflow and update results are sent as their neutral values. A typed `TypedValueError` names the path but never the value. Untyped workflows receive plain values as before, so in-flight histories replay unchanged.
+
 ## [2.14.0] - 2026-11-04
 
 **Upgrade from 2.13.2 and earlier.** Every earlier release fails replay when a condition wait has an orphaned duplicate open whose timer fires later, which Server 3.0.x can record (#253 below).

docs/sdk-reference.md

@@ -1020,6 +1020,40 @@ class Fulfil:
   for a single call, inside a parallel group (`yield [...]`) and inside
   `ctx.select(...)`. `result_type` is local: it is never sent, compared or
   written to history.
+- `@workflow.defn(..., typed=True)` decodes the arguments of `run` (after
+  `ctx`) and of every signal, query, update and update-validator handler.
+- `ctx.start_child_workflow(..., result_type=T)` decodes a child's result,
+  and `await handle.result(result_type=T)` decodes a workflow's result on the
+  client.
+
+Dataclass and `Enum` values are also sent as neutral values when they are
+client workflow, signal, update, query or schedule arguments, child or
+continue-as-new arguments, or workflow and update results.
+
+Where a mismatch surfaces, by surface:
+
+| Surface | What a `TypedValueError` does |
+| --- | --- |
+| `typed=True` activity argument | Fails the activity without retries. |
+| `result_type` on an activity or child | Raised at the `yield`; workflow code can catch it. In `ctx.select`, it is the winner's `failure`. |
+| `run` argument of a `typed=True` workflow | Fails the workflow run. |
+| Signal argument of a `typed=True` workflow | Raised in the signal handler during replay, so it fails the workflow run. |
+| Update or update-validator argument | Fails or rejects that update; the run continues. |
+| Query argument | Fails that query. |
+| Client `result_type` | Raised by `handle.result(...)`; the workflow itself is unaffected. |
+
+Turning on `typed=True` does not change a recorded `wait_condition` predicate:
+a handler with nothing to decode is left unwrapped, and a wrapped handler is
+fingerprinted as the method it wraps. Handlers must be instance methods;
+`staticmethod` and `classmethod` handlers raise `TypeError` under
+`typed=True`.
+
+**Changing a dataclass while runs are in flight.** Replay decodes recorded
+payloads with the current code. Adding a field is safe only with a default (or
+`default_factory`); a new required field makes replay of older payloads fail
+with `TypedValueError`, which fails the run or the activity as described above.
+Removing or renaming a field drops that recorded key silently, because
+undeclared keys are ignored.
 
 Supported annotations: dataclasses, `Enum` (including `IntEnum` and `StrEnum`,
 which come back as their class), `Optional[T]` / `T | None`, `list[T]`,

src/durable_workflow/_typed_values.py

@@ -255,22 +255,22 @@ def to_payload_values(values: list[Any]) -> list[Any]:
     return [to_payload_value(value) for value in values]
 
 
-def typed_callable(fn: Callable[..., Any]) -> Callable[..., Any]:
-    """Wrap *fn* so positional arguments decode to its annotations.
+def _positional_decoder(
+    fn: Callable[..., Any], skip: int, where: str,
+) -> Callable[[tuple[Any, ...]], list[Any]] | None:
+    """Build a decoder for *fn*'s positional arguments after the first *skip*.
 
-    Annotations are validated here, so an unsupported type fails when the
-    activity is declared. The wrapper keeps *fn*'s module and qualified name:
-    the decorator binds that name to the wrapper, so supervised callbacks that
-    pickle by reference resolve to it.
+    Returns None when no argument needs decoding (none, or only ``Any`` /
+    unannotated ones), so the caller can leave *fn* unwrapped.
     """
     signature = inspect.signature(fn)
     try:
         hints = get_type_hints(fn)
     except (NameError, TypeError) as error:
-        raise TypeError(f"typed activity {fn.__qualname__} has unresolvable annotations: {error}") from None
+        raise TypeError(f"{where} {fn.__qualname__} has unresolvable annotations: {error}") from None
     positional: list[tuple[str, Any]] = []
     variadic: tuple[str, Any] | None = None
-    for parameter in signature.parameters.values():
+    for parameter in list(signature.parameters.values())[skip:]:
         annotation = hints.get(parameter.name, Any)
         if parameter.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD):
             validate_annotation(annotation, parameter.name)
@@ -279,6 +279,10 @@ def typed_callable(fn: Callable[..., Any]) -> Callable[..., Any]:
             validate_annotation(annotation, parameter.name)
             variadic = (parameter.name, annotation)
 
+    annotations = [annotation for _, annotation in positional] + ([variadic[1]] if variadic is not None else [])
+    if all(annotation in _PASS_THROUGH for annotation in annotations):
+        return None
+
     def convert(args: tuple[Any, ...]) -> list[Any]:
         converted: list[Any] = []
         for index, value in enumerate(args):
@@ -292,6 +296,22 @@ def convert(args: tuple[Any, ...]) -> list[Any]:
                 converted.append(value)
         return converted
 
+    return convert
+
+
+def typed_callable(fn: Callable[..., Any]) -> Callable[..., Any]:
+    """Wrap an activity so positional arguments decode to its annotations.
+
+    Annotations are validated here, so an unsupported type fails when the
+    activity is declared. The wrapper keeps *fn*'s module and qualified name:
+    the decorator binds that name to the wrapper, so supervised callbacks that
+    pickle by reference resolve to it.
+    """
+    decoder = _positional_decoder(fn, 0, "typed activity")
+
+    def convert(args: tuple[Any, ...]) -> list[Any]:
+        return list(args) if decoder is None else decoder(args)
+
     wrapper: Any
     if inspect.iscoroutinefunction(fn):
 
@@ -309,3 +329,42 @@ def typed_sync(*args: Any, **kwargs: Any) -> Any:
         wrapper = typed_sync
     wrapper.__typed_activity__ = fn
     return cast(Callable[..., Any], wrapper)
+
+
+def typed_method(fn: Callable[..., Any], skip: int) -> Callable[..., Any]:
+    """Wrap a workflow method so arguments after the first *skip* decode.
+
+    ``skip`` is 1 for handlers (``self``) and 2 for ``run`` (``self, ctx``). A
+    generator ``run`` stays a generator. Results are returned unchanged; the
+    commands that carry them send dataclass and Enum values as neutral values.
+    A method with nothing to decode is returned unwrapped.
+    """
+    convert = _positional_decoder(fn, skip, "typed workflow method")
+    if convert is None:
+        # Nothing to decode: keep the original function, so a handler used as a
+        # wait_condition predicate keeps its recorded fingerprint.
+        return fn
+    wrapper: Any
+    if inspect.isgeneratorfunction(fn):
+
+        @functools.wraps(fn)
+        def typed_generator(*args: Any, **kwargs: Any) -> Any:
+            return (yield from fn(*args[:skip], *convert(args[skip:]), **kwargs))
+
+        wrapper = typed_generator
+    elif inspect.iscoroutinefunction(fn):
+
+        @functools.wraps(fn)
+        async def typed_async(*args: Any, **kwargs: Any) -> Any:
+            return await fn(*args[:skip], *convert(args[skip:]), **kwargs)
+
+        wrapper = typed_async
+    else:
+
+        @functools.wraps(fn)
+        def typed_sync(*args: Any, **kwargs: Any) -> Any:
+            return fn(*args[:skip], *convert(args[skip:]), **kwargs)
+
+        wrapper = typed_sync
+    wrapper.__typed_workflow_method__ = fn
+    return cast(Callable[..., Any], wrapper)

src/durable_workflow/client.py

@@ -62,6 +62,7 @@
 from ._poll_transport import unwrap_response as _unwrap_poll_response
 from ._poll_transport import warn_early_cutoff as _warn_poll_cutoff
 from ._query_task_authority import QueryTaskAuthority, current_query_authority
+from ._typed_values import decode_payload_value, to_payload_values, validate_annotation
 from .errors import (
     REQUEST_ID_CONFLICT,
     ExternalPayloadIntegrityMismatch,
@@ -1328,6 +1329,12 @@
         )
 
 
+# Client payloads that carry caller arguments.
+_ARGUMENT_PAYLOAD_KINDS = frozenset({
+    "workflow_input", "signal", "update", "query", "activity_input", "schedule_input",
+})
+
+
 class WorkflowHandle:
     """Convenience wrapper for operating on one workflow ID."""
 
@@ -1337,9 +1344,11 @@
         self.run_id = run_id
         self.workflow_type = workflow_type
 
-    async def result(self, *, poll_interval: float = 0.5, timeout: float = 30.0) -> Any:
+    async def result(self, *, poll_interval: float = 0.5, timeout: float = 30.0, result_type: Any = None) -> Any:
         """Block until this workflow terminates and return its result. See :meth:`Client.get_result`."""
-        return await self._client.get_result(self, poll_interval=poll_interval, timeout=timeout)
+        return await self._client.get_result(
+            self, poll_interval=poll_interval, timeout=timeout, result_type=result_type,
+        )
 
     async def describe(self) -> WorkflowExecution:
         """Return the server's current view of this workflow. See :meth:`Client.describe_workflow`."""
@@ -1807,6 +1816,11 @@
         external_storage: ExternalStorageDriver | None = None,
         external_storage_threshold_bytes: int | None = None,
     ) -> dict[str, Any]:
+        if kind in _ARGUMENT_PAYLOAD_KINDS and isinstance(value, list):
+            # Dataclass and Enum arguments travel as their neutral values, the
+            # bytes dataclasses.asdict() and .value give. The codec refuses
+            # those values, so no existing caller depends on this conversion.
+            value = to_payload_values(value)
         warning_context = self._payload_context(
             kind=kind,
             workflow_id=workflow_id,
@@ -4760,16 +4774,25 @@
         *,
         poll_interval: float = 0.5,
         timeout: float = 30.0,
+        result_type: Any = None,
     ) -> Any:
         """Poll a workflow until it reaches a terminal state and return its result.
 
+        ``result_type`` opts in to decoding the result to that annotation
+        (the types :func:`durable_workflow.activity.defn` lists for
+        ``typed=True``); a mismatch raises
+        :class:`~durable_workflow.errors.TypedValueError`, which names the
+        field path but never the value. Without it the plain value is returned.
+
         Raises :class:`~durable_workflow.errors.WorkflowFailed`,
         :class:`~durable_workflow.errors.WorkflowCancelled`, or
         :class:`~durable_workflow.errors.WorkflowTerminated` if the workflow
         ended in a non-success state. A persisted execution or run deadline
         raises :class:`~durable_workflow.errors.WorkflowTimedOut`; a caller's
         :class:`TimeoutError` means ``timeout`` seconds elapsed while waiting.
         """
+        if result_type is not None:
+            validate_annotation(result_type, "result_type")
         deadline = asyncio.get_running_loop().time() + timeout
         while True:
             desc = await self.describe_workflow(handle.workflow_id)
@@ -4789,12 +4812,15 @@
                         etype = ev.get("event_type")
                         payload = ev.get("payload") or {}
                         if etype == "WorkflowCompleted":
-                            return serializer.decode_envelope(
+                            output = serializer.decode_envelope(
                                 payload.get("output"),
                                 codec=payload.get("payload_codec") or desc.payload_codec,
                                 external_storage=self.external_storage,
                                 external_storage_cache=self.external_storage_cache,
                             )
+                            if result_type is None:
+                                return output
+                            return decode_payload_value(output, result_type, "workflow result")
                         if etype == "WorkflowFailed":
                             raise WorkflowFailed(
                                 payload.get("message", "workflow failed"),