diff --git a/CHANGELOG.md b/CHANGELOG.md index a653cdd2..9aed71b6 100644 --- a/CHANGELOG.md +++ b/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 activity values (#184). `@activity.defn(..., typed=True)` decodes an activity's positional arguments to its parameter annotations, and `ctx.schedule_activity(..., result_type=T)` decodes the completed result in single calls, parallel groups and selections. Supported types are dataclasses, `Enum` (including `IntEnum` and `StrEnum`), `Optional[T]`, `list[T]`, `dict[str, T]` and scalars; other annotations raise `TypeError` up front. Undeclared keys are ignored and missing fields take their defaults. A mismatch raises the new `TypedValueError`, which names the field path but never the value. It is non-retryable in a typed activity and is raised at the `yield` for `result_type`. Dataclass and `Enum` activity arguments and results are now sent as their neutral values, the same bytes as `dataclasses.asdict()` with each enum's `.value`; these values were refused before. Untyped activities and calls receive plain values as before, so in-flight histories replay unchanged. + ## [2.13.2] - 2026-11-04 ### Fixed diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 7ea6a91e..26ebb71e 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -957,6 +957,85 @@ activities, for example `OrderInput(order_id=UUID(data["order_id"]), ...)`. Adapter output is part of the durable history contract, so changing that shape is a workflow compatibility change. +## Typed activity values + +Activity boundaries can carry dataclasses and enums without hand-written +conversion. The wire format stays the neutral map, so PHP and Rust workers read +and write the same values. + +**Sending.** A dataclass or `Enum` passed as an activity argument, or returned +by an activity, is sent as its neutral value: a dataclass as the string-keyed +map of its fields and an `Enum` as its `.value`. The bytes are the same as +passing `dataclasses.asdict(value)` with each enum replaced by `.value`. Other +rich values (datetimes, `UUID`, `Decimal`, pydantic and attrs models) still +need `to_avro_payload_value(...)` or explicit conversion. + +**Receiving (opt in).** Decoding to a Python type happens only where you ask +for it, so existing workflows keep receiving plain values on replay: + +```python +from dataclasses import dataclass, field +from enum import Enum + +from durable_workflow import activity, workflow +from durable_workflow.workflow import WorkflowContext + + +class Status(Enum): + OPEN = "open" + SHIPPED = "shipped" + + +@dataclass +class Item: + sku: str + qty: int = 1 + + +@dataclass +class Shipment: + order_id: str + status: Status + items: list[Item] = field(default_factory=list) + note: str | None = None + + +@activity.defn(name="ship", typed=True) +def ship(shipment: Shipment) -> Shipment: + # shipment is a Shipment; the returned Shipment is sent as a neutral map. + return Shipment(shipment.order_id, Status.SHIPPED, shipment.items) + + +@workflow.defn(name="fulfil") +class Fulfil: + def run(self, ctx: WorkflowContext, order_id: str): + shipment = Shipment(order_id, Status.OPEN, [Item("a", 2)]) + shipped = yield ctx.schedule_activity("ship", [shipment], result_type=Shipment) + return shipped.status.value +``` + +- `@activity.defn(..., typed=True)` decodes positional arguments to the + function's parameter annotations. +- `ctx.schedule_activity(..., result_type=T)` decodes the completed result + 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. + +Supported annotations: dataclasses, `Enum` (including `IntEnum` and `StrEnum`, +which come back as their class), `Optional[T]` / `T | None`, `list[T]`, +`dict[str, T]`, `str`, `int`, `float` (an integer is accepted), `bool`, `bytes` +and `Any`. Any other annotation, including a union of two non-`None` types, +raises `TypeError` when the activity is declared or the call is made. + +Decoding follows rolling-deploy rules. Keys a dataclass does not declare are +ignored, and a missing field takes its default. A missing field with no +default, a value of the wrong kind, or an unknown enum value raises +`TypedValueError`, whose `path` names the field (for example +`shipment.items[0].sku`). The message never includes the value. In a typed +activity the error fails the activity without retries. With `result_type` it is +raised at the `yield`, where workflow code can catch it; a selection winner +reports it as `SelectionResult.failure`. + ## Authentication For local servers that use one shared bearer token, pass `token=`: diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 0dfd811c..3515e777 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -98,6 +98,7 @@ ScheduleNotFound, ServerError, SignalFailed, + TypedValueError, Unauthorized, UpdateFailed, UpdateRejected, @@ -280,6 +281,7 @@ "AuthCompositionContract", "AuthCompositionContractError", "NonRetryableError", + "TypedValueError", "ReplayOutcome", "Replayer", "Saga", diff --git a/src/durable_workflow/_typed_values.py b/src/durable_workflow/_typed_values.py new file mode 100644 index 00000000..2e4f5095 --- /dev/null +++ b/src/durable_workflow/_typed_values.py @@ -0,0 +1,293 @@ +"""Opt-in conversion between neutral payload values and annotated Python types. + +The wire format stays the neutral Avro Value: a dataclass travels as the +string-keyed map of its fields and an ``Enum`` as its value, exactly as +``dataclasses.asdict()`` and ``.value`` would produce by hand. Decoding is +driven only by the receiver's own annotations, so no class name, tag or +pickled object ever enters durable history. + +Decoding follows payload-evolution rules for rolling deploys: keys a target +dataclass does not declare are ignored, and a missing field takes its +default. A value that cannot be converted raises +:class:`~durable_workflow.errors.TypedValueError`, which names the field path +and the expected type but never the value. +""" + +from __future__ import annotations + +import dataclasses +import functools +import inspect +import types +from collections.abc import Callable +from enum import Enum +from typing import Any, Union, cast, get_args, get_origin, get_type_hints + +from .errors import TypedValueError + +_SCALARS: tuple[type, ...] = (str, int, float, bool, bytes) +_PASS_THROUGH = (Any, object, inspect.Parameter.empty) + + +def _type_name(annotation: Any) -> str: + """A short, module-free name for diagnostics, such as ``list[Item]``.""" + if annotation is type(None) or annotation is None: + return "None" + if _is_union(annotation): + options = get_args(annotation) + concrete = [option for option in options if option is not type(None)] + if len(concrete) == 1 and len(options) == 2: + return f"Optional[{_type_name(concrete[0])}]" + return " | ".join(_type_name(option) for option in options) + origin = get_origin(annotation) + if origin is not None: + arguments = ", ".join(_type_name(argument) for argument in get_args(annotation)) + name = getattr(origin, "__name__", repr(origin).replace("typing.", "")) + return f"{name}[{arguments}]" if arguments else name + if isinstance(annotation, type): + return annotation.__qualname__ + return repr(annotation).replace("typing.", "") + + +def _json_kind(value: Any) -> str: + """The neutral kind of a received value, never the value itself.""" + if value is None: + return "null" + if isinstance(value, bool): + return "boolean" + if isinstance(value, int): + return "integer" + if isinstance(value, float): + return "number" + if isinstance(value, str): + return "string" + if isinstance(value, bytes): + return "bytes" + if isinstance(value, list): + return "array" + if isinstance(value, dict): + return "object" + return type(value).__name__ + + +def _optional_member(annotation: Any) -> Any: + """Return T for Optional[T]; refuse every other union as ambiguous.""" + options = get_args(annotation) + concrete = [option for option in options if option is not type(None)] + if len(concrete) != 1 or len(concrete) == len(options): + raise TypeError( + f"typed values support only Optional[T] unions; {_type_name(annotation)} is ambiguous" + ) + return concrete[0] + + +def _is_union(annotation: Any) -> bool: + return get_origin(annotation) in (Union, types.UnionType) + + +def _dataclass_hints(cls: type) -> dict[str, Any]: + try: + return get_type_hints(cls) + except (NameError, TypeError) as error: + raise TypeError(f"typed values cannot resolve the annotations of {cls.__qualname__}: {error}") from None + + +def validate_annotation(annotation: Any, where: str, _seen: frozenset[type] = frozenset()) -> None: + """Refuse an annotation this contract cannot decode unambiguously. + + Called when a typed activity is declared or a ``result_type`` is given, so + an unsupported type fails at definition time instead of passing a plain + value through silently. + """ + if annotation in _PASS_THROUGH or annotation is None or annotation is type(None): + return + if _is_union(annotation): + validate_annotation(_optional_member(annotation), where, _seen) + return + origin = get_origin(annotation) + if annotation is list or origin is list: + for item in get_args(annotation): + validate_annotation(item, where, _seen) + return + if annotation is dict or origin is dict: + arguments = get_args(annotation) + if arguments: + if arguments[0] is not str: + raise TypeError(f"{where}: typed maps need str keys, not {_type_name(arguments[0])}") + validate_annotation(arguments[1], where, _seen) + return + if isinstance(annotation, type) and issubclass(annotation, Enum): + for member in annotation: + if not isinstance(member.value, str | int | float): + raise TypeError( + f"{where}: {annotation.__qualname__} members must have str, int or float values" + ) + return + if isinstance(annotation, type) and dataclasses.is_dataclass(annotation): + if annotation in _seen: + return + hints = _dataclass_hints(annotation) + for item in dataclasses.fields(annotation): + if item.init: + validate_annotation(hints.get(item.name, Any), f"{where}.{item.name}", _seen | {annotation}) + return + if isinstance(annotation, type) and annotation in _SCALARS: + return + raise TypeError(f"{where}: typed values do not support {_type_name(annotation)}") + + +def from_payload_value(value: Any, annotation: Any, path: str = "value") -> Any: + """Return *value* converted to *annotation*, which must be validated.""" + if annotation in _PASS_THROUGH: + return value + if annotation is None or annotation is type(None): + if value is not None: + raise TypedValueError(path, "None", f"got {_json_kind(value)}") + return None + if _is_union(annotation): + if value is None: + return None + return from_payload_value(value, _optional_member(annotation), path) + if value is None: + raise TypedValueError(path, _type_name(annotation), "got null") + + origin = get_origin(annotation) + if annotation is list or origin is list: + if not isinstance(value, list): + raise TypedValueError(path, _type_name(annotation), f"got {_json_kind(value)}") + (item_type,) = get_args(annotation) or (Any,) + return [from_payload_value(item, item_type, f"{path}[{index}]") for index, item in enumerate(value)] + + if annotation is dict or origin is dict: + if not isinstance(value, dict): + raise TypedValueError(path, _type_name(annotation), f"got {_json_kind(value)}") + _, item_type = get_args(annotation) or (str, Any) + return {key: from_payload_value(item, item_type, f"{path}[{key!r}]") for key, item in value.items()} + + if isinstance(annotation, type) and issubclass(annotation, Enum): + return _enum_from_payload(value, annotation, path) + + if isinstance(annotation, type) and dataclasses.is_dataclass(annotation): + return _dataclass_from_payload(value, annotation, path) + + if annotation is float and isinstance(value, int) and not isinstance(value, bool): + return float(value) + if annotation is int and isinstance(value, bool): + raise TypedValueError(path, "int", "got boolean") + if isinstance(annotation, type) and annotation in _SCALARS: + if not isinstance(value, annotation): + raise TypedValueError(path, annotation.__qualname__, f"got {_json_kind(value)}") + return value + raise TypeError(f"{path}: typed values do not support {_type_name(annotation)}") + + +def _enum_from_payload(value: Any, cls: type[Enum], path: str) -> Enum: + if isinstance(value, cls): + return value + for member in cls: + expected = member.value + same_kind = type(expected) is type(value) or (type(expected) is float and type(value) is int) + if same_kind and expected == value: + return member + # The diagnostic must not echo the received value. + raise TypedValueError(path, cls.__qualname__, f"the {_json_kind(value)} is not a member") + + +def _dataclass_from_payload(value: Any, cls: type, path: str) -> Any: + if isinstance(value, cls): + return value + if not isinstance(value, dict): + raise TypedValueError(path, cls.__qualname__, f"got {_json_kind(value)}") + hints = _dataclass_hints(cls) + arguments: dict[str, Any] = {} + for item in dataclasses.fields(cls): + if not item.init: + # Computed fields are derived again by the dataclass itself. + continue + field_path = f"{path}.{item.name}" + annotation = hints.get(item.name, Any) + if item.name in value: + arguments[item.name] = from_payload_value(value[item.name], annotation, field_path) + elif item.default is dataclasses.MISSING and item.default_factory is dataclasses.MISSING: + raise TypedValueError(field_path, _type_name(annotation), "the field is missing") + # Keys the receiver does not declare are ignored for rolling deploys. + return cls(**arguments) + + +def to_payload_value(value: Any) -> Any: + """Return the neutral value for dataclass and ``Enum`` values, recursively. + + Every other value is returned unchanged, so the codec keeps refusing what + it refuses today (for example datetimes and tuples). The result encodes + to the same bytes as ``dataclasses.asdict()`` with each ``Enum`` replaced + by its ``.value``. + """ + if isinstance(value, Enum): + return to_payload_value(value.value) + if dataclasses.is_dataclass(value) and not isinstance(value, type): + return {item.name: to_payload_value(getattr(value, item.name)) for item in dataclasses.fields(value)} + if isinstance(value, list): + return [to_payload_value(item) for item in value] + if isinstance(value, dict): + return {key: to_payload_value(item) for key, item in value.items()} + return value + + +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. + + 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. + """ + 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 + positional: list[tuple[str, Any]] = [] + variadic: tuple[str, Any] | None = None + for parameter in signature.parameters.values(): + annotation = hints.get(parameter.name, Any) + if parameter.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD): + validate_annotation(annotation, parameter.name) + positional.append((parameter.name, annotation)) + elif parameter.kind is inspect.Parameter.VAR_POSITIONAL: + validate_annotation(annotation, parameter.name) + variadic = (parameter.name, annotation) + + def convert(args: tuple[Any, ...]) -> list[Any]: + converted: list[Any] = [] + for index, value in enumerate(args): + if index < len(positional): + name, annotation = positional[index] + converted.append(from_payload_value(value, annotation, name)) + elif variadic is not None: + name, annotation = variadic + converted.append(from_payload_value(value, annotation, f"{name}[{index - len(positional)}]")) + else: + converted.append(value) + return converted + + wrapper: Any + if inspect.iscoroutinefunction(fn): + + @functools.wraps(fn) + async def typed_async(*args: Any, **kwargs: Any) -> Any: + return to_payload_value(await fn(*convert(args), **kwargs)) + + wrapper = typed_async + else: + + @functools.wraps(fn) + def typed_sync(*args: Any, **kwargs: Any) -> Any: + return to_payload_value(fn(*convert(args), **kwargs)) + + wrapper = typed_sync + wrapper.__typed_activity__ = fn + return cast(Callable[..., Any], wrapper) diff --git a/src/durable_workflow/activity.py b/src/durable_workflow/activity.py index 5334e4f5..27bf7588 100644 --- a/src/durable_workflow/activity.py +++ b/src/durable_workflow/activity.py @@ -206,10 +206,25 @@ def _set_context(ctx: ActivityContext | None) -> contextvars.Token[ActivityConte return _current_context.set(ctx) -def defn(*, name: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]: - """Register a callable as an activity type under a language-neutral name.""" +def defn(*, name: str, typed: bool = False) -> Callable[[Callable[..., Any]], Callable[..., Any]]: + """Register a callable as an activity type under a language-neutral name. + + With ``typed=True``, positional arguments are decoded to the callable's + parameter annotations: dataclasses, ``Enum`` (including ``IntEnum`` and + ``StrEnum``), ``Optional[T]``, ``list[T]``, ``dict[str, T]`` and the + scalars ``str``, ``int``, ``float``, ``bool`` and ``bytes``. Keys a + dataclass does not declare are ignored and missing fields take their + defaults. A value that does not match fails the activity without retries + with :class:`~durable_workflow.errors.TypedValueError`. An annotation + outside this list raises ``TypeError`` here, when the activity is + declared. Untyped activities keep receiving plain values. + """ def wrap(fn: Callable[..., Any]) -> Callable[..., Any]: + if typed: + from ._typed_values import typed_callable + + fn = typed_callable(fn) fn.__activity_name__ = name # type: ignore[attr-defined] _REGISTRY[name] = fn return fn diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 86916cd3..e5b01b60 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -881,6 +881,26 @@ def __init__(self, message: str, *, cause: Exception | None = None) -> None: self.__cause__ = cause +class TypedValueError(NonRetryableError, TypeError): + """A payload value does not match the receiver's declared type. + + Raised by opt-in typed values: a ``typed=True`` activity's arguments, or an + activity result decoded to ``result_type``. ``path`` names the field (for + example ``order.items[2].sku``) and ``expected`` the declared type. The + message never contains the received value. An activity whose arguments do + not match fails without retries, because a retry cannot change the payload. + """ + + def __init__(self, path: str, expected: str, detail: str) -> None: + self.path = path + self.expected = expected + self.detail = detail + super().__init__(f"{path}: expected {expected}; {detail}") + + def __reduce__(self) -> tuple[Any, ...]: + return (type(self), (self.path, self.expected, self.detail)) + + class AvroNotInstalledError(DurableWorkflowError, ImportError): """Raised when the core ``avro`` runtime dependency is unavailable.""" diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index dc4a3615..dae62da0 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -49,6 +49,7 @@ from ._prepared_local_activity import PreparedAttempt, PreparedCancellationObserved, PreparedLocalRunner from ._query_task_authority import QueryTaskAuthority, current_query_authority from ._sticky_workflow_cache import CacheKey, StickyWorkflowCache, complete_history +from ._typed_values import to_payload_value from .activity import ActivityContext, ActivityInfo, _set_context from .auth_composition import ( AUTH_COMPOSITION_CONTRACT_SCHEMA, @@ -2505,6 +2506,9 @@ async def heartbeat( "duration_ms": max(0, round((time.monotonic() - attempt_started_at) * 1000)), "heartbeats": heartbeats, }) + # The workflow receives the same neutral value on this first + # run as on replay of the recorded result. + result = to_payload_value(result) try: inline_result = serializer.envelope(result, codec=payload_codec) except Exception: @@ -3270,6 +3274,7 @@ async def _run_activity_task_body(self, task: dict[str, Any]) -> str: finally: _set_context(None) + result = to_payload_value(result) log.info("completing activity task %s (%s) -> %r", task_id, activity_type, result) try: await self.client.complete_activity_task( diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 78b28783..86fb9157 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -44,6 +44,7 @@ ) from ._cooperative_cancellation import CancellationDelivery, read_cancellation_history from ._failure_reporting import exception_message +from ._typed_values import from_payload_value, to_payload_values, validate_annotation from .cancellation import ( CancellationContext, CancellationPolicy, @@ -66,6 +67,7 @@ NonDeterministicReplayError, QueryFailed, SagaCompensationFailed, + TypedValueError, WorkflowCancelled, WorkflowEntryArgumentsMismatchError, WorkflowPayloadDecodeError, @@ -583,6 +585,9 @@ class ScheduleActivity: heartbeat_timeout: int | None = None worker_session: WorkerSessionOptions | None = None cancellation_policy: str | CancellationPolicy | None = None + # Opt-in local decoding of the completed result. It is never encoded into + # the command, compared, or written to history. + result_type: Any = field(default=None, repr=False, compare=False) _cancellation_scope_id: str = field( default_factory=_authored_cancellation_scope, init=False, repr=False, compare=False, ) @@ -594,6 +599,12 @@ class ScheduleActivity: ) def __post_init__(self) -> None: + # Dataclass and Enum arguments travel as their neutral values, the + # bytes dataclasses.asdict() and .value give. The codec refuses those + # values, so no existing history depends on this conversion. + self.arguments = to_payload_values(list(self.arguments)) + if self.result_type is not None: + validate_annotation(self.result_type, "result_type") self._validate_cancellation_policy() def _validate_cancellation_policy(self) -> None: @@ -2396,7 +2407,19 @@ def schedule_activity( heartbeat_timeout: int | None = None, worker_session: WorkerSessionOptions | None = None, cancellation_policy: str | CancellationPolicy | None = None, + result_type: Any = None, ) -> ScheduleActivity: + """Yield an activity request. + + Dataclass and ``Enum`` arguments are sent as their neutral values. + ``result_type`` opts in to decoding the completed result to that + annotation (the types :func:`durable_workflow.activity.defn` lists for + ``typed=True``), in single calls, parallel groups and selections. It + changes only the value this workflow receives; history keeps the + neutral payload. A result that does not match raises + :class:`~durable_workflow.errors.TypedValueError` at this yield, where + workflow code can catch it. + """ return ScheduleActivity( activity_type=activity_type, arguments=list(arguments), @@ -2408,6 +2431,7 @@ def schedule_activity( heartbeat_timeout=heartbeat_timeout, worker_session=worker_session, cancellation_policy=cancellation_policy, + result_type=result_type, ) def local_activity( @@ -3100,6 +3124,13 @@ def _history_event_time(event: Mapping[str, Any]) -> datetime | None: return parsed if parsed.tzinfo is not None and parsed.utcoffset() is not None else None +def _typed_activity_result(command: Any, value: Any) -> Any: + """Decode an activity result to the call's opt-in result_type, if any.""" + if not isinstance(command, ScheduleActivity) or command.result_type is None or isinstance(value, BaseException): + return value + return from_payload_value(value, command.result_type, "activity result") + + def _decode_history_result( payload: dict[str, Any], fallback_codec: str | None, @@ -6345,6 +6376,7 @@ def _selection_member_value( *, outcome: str | None = None, resolution_sequence: int | None = None, + leaves: Sequence[Any] | None = None, ) -> tuple[bool, Any, BaseException | None]: if outcome == "failed": if resolution_sequence is None: @@ -6370,6 +6402,9 @@ def _selection_member_value( if isinstance(value, BaseException): return True, None, value values.append(value) + if leaves is not None: + # May raise TypedValueError; callers deliver it at the yield. + values = [_typed_activity_result(leaf, value) for leaf, value in zip(leaves, values, strict=True)] if shape is None: return True, values[0], None return True, _parallel_result_shape(shape, values), None @@ -7293,13 +7328,21 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None min(handle.base_sequence for handle in handles.values()), len(annotated), ) - resolved, value, failure = _selection_member_value( - winner.base_sequence, - winner.size, - shapes[winner_key], - outcome=str(marker.get("outcome")), - resolution_sequence=resolution_sequence, - ) + winner_offset = winner.base_sequence - group_base + typed_failure: TypedValueError | None = None + try: + resolved, value, failure = _selection_member_value( + winner.base_sequence, + winner.size, + shapes[winner_key], + outcome=str(marker.get("outcome")), + resolution_sequence=resolution_sequence, + leaves=annotated[winner_offset:winner_offset + winner.size], + ) + except TypedValueError as error: + # The committed winner stands; only its typed decode failed. + typed_failure = error + resolved, value, failure = True, cast(Any, None), None if marker.get("outcome") == "completed" and failure is not None: raise NonDeterministicReplayError( winner.base_sequence, @@ -7328,7 +7371,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None kind=winner.kind, identity=winner.identity, value=value, - failure=failure, + failure=failure if typed_failure is None else typed_failure, winner=winner, handles=handles, ) @@ -7373,10 +7416,16 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if index is not None and failure_index is not None and index <= failure_index ] _advance_cancellation_clock(consumed_indexes) - if failed is not None: + group_typed_failure: TypedValueError | None = None + if failed is None: + try: + vals = [_typed_activity_result(leaf, value) for leaf, value in zip(cmd, vals, strict=True)] + except TypedValueError as error: + group_typed_failure = error + if failed is not None or group_typed_failure is not None: try: _refresh_replay_logging() - advanced_cmd = gen.throw(failed) + advanced_cmd = gen.throw(failed if failed is not None else group_typed_failure) continue except StopIteration as stop: return _terminal_state(stop.value, include_pending=False) @@ -7416,19 +7465,25 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None ) else: selection_leaves = [cmd.operation] - resolved, value, failure = _selection_member_value( - cmd.base_sequence, - cmd.size, - selection_shape, - ) + handle_typed_failure: TypedValueError | None = None + try: + resolved, value, failure = _selection_member_value( + cmd.base_sequence, + cmd.size, + selection_shape, + leaves=selection_leaves, + ) + except TypedValueError as error: + handle_typed_failure = error + resolved, value, failure = True, cast(Any, None), None if not resolved: _set_replaying(False) return _state(pending) _advance_selection_clock(cmd.base_sequence, cmd.size, failure) - if failure is not None: + if failure is not None or handle_typed_failure is not None: try: _refresh_replay_logging() - advanced_cmd = gen.throw(failure) + advanced_cmd = gen.throw(failure if failure is not None else handle_typed_failure) continue except StopIteration as stop: return _terminal_state(stop.value, include_pending=True) @@ -7705,7 +7760,12 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None if not isinstance(cmd, RecordLocalActivity) or prepare_local_activities: _advance_cancellation_clock([recorded_steps[result_cursor].history_index]) result_cursor += 1 - if isinstance(val, ActivityFailed | ChildWorkflowFailed): + if not isinstance(val, ActivityFailed | ChildWorkflowFailed): + try: + val = _typed_activity_result(cmd, val) + except TypedValueError as error: + val = error + if isinstance(val, ActivityFailed | ChildWorkflowFailed | TypedValueError): try: _refresh_replay_logging() advanced_cmd = gen.throw(val) diff --git a/tests/fixtures/codec_regressions/typed-dataclass-enum-neutral-map.json b/tests/fixtures/codec_regressions/typed-dataclass-enum-neutral-map.json new file mode 100644 index 00000000..c62af433 --- /dev/null +++ b/tests/fixtures/codec_regressions/typed-dataclass-enum-neutral-map.json @@ -0,0 +1,81 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.codec-regression/v1", + "id": "typed-dataclass-enum-neutral-map", + "protocol": { + "codec": "avro", + "schema": "durable_workflow.protocol.Value", + "version": "1", + "fingerprint": "e2a33dff55802237" + }, + "bindings": [ + "python" + ], + "value": { + "type": "map", + "entries": [ + { + "key": "order_id", + "value": { + "type": "string", + "value": "o-1" + } + }, + { + "key": "status", + "value": { + "type": "string", + "value": "shipped" + } + }, + { + "key": "priority", + "value": { + "type": "long", + "value": "2" + } + }, + { + "key": "items", + "value": { + "type": "array", + "items": [ + { + "type": "map", + "entries": [ + { + "key": "sku", + "value": { + "type": "string", + "value": "a" + } + }, + { + "key": "qty", + "value": { + "type": "long", + "value": "2" + } + } + ] + } + ] + } + }, + { + "key": "note", + "value": { + "type": "null" + } + } + ] + }, + "framing": { + "encoding": "avro-single-object", + "wire_base64": "wwHioz3/VYAiNw4KEG9yZGVyX2lkCgZvLTEMc3RhdHVzCg5zaGlwcGVkEHByaW9yaXR5BAQKaXRlbXMMAg4EBnNrdQoCYQZxdHkEBAAACG5vdGUAAA==" + }, + "failure_policy": { + "operation": "round_trip", + "error": null + } +} diff --git a/tests/fixtures/replay_regressions/typed-activity-results-opt-in.json b/tests/fixtures/replay_regressions/typed-activity-results-opt-in.json new file mode 100644 index 00000000..9924728f --- /dev/null +++ b/tests/fixtures/replay_regressions/typed-activity-results-opt-in.json @@ -0,0 +1,144 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "typed-activity-results-opt-in", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.typed-activity-results", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "payload": { + "workflow_type": "tests.replay.typed-activity-results" + } + }, + { + "sequence": 2, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_type": "typed.lookup" + } + }, + { + "sequence": 3, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "typed.lookup", + "result": "wwHioz3/VYAiNw4IEG9yZGVyX2lkCgZvLTEMc3RhdHVzCghvcGVuCml0ZW1zDAIOBAZza3UKAmEGcXR5BAQAABZhZGRlZF9pbl92MgIBAA==", + "payload_codec": "avro" + } + }, + { + "sequence": 4, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "typed.lookup", + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + } + }, + { + "sequence": 5, + "event_type": "ActivityScheduled", + "payload": { + "sequence": 3, + "activity_type": "typed.count", + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + } + }, + { + "sequence": 6, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 3, + "activity_type": "typed.count", + "result": "wwHioz3/VYAiNwQG", + "payload_codec": "avro", + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + } + }, + { + "sequence": 7, + "event_type": "ActivityCompleted", + "payload": { + "sequence": 2, + "activity_type": "typed.lookup", + "result": "wwHioz3/VYAiNw4GEG9yZGVyX2lkCgZvLTIMc3RhdHVzCg5zaGlwcGVkCml0ZW1zDAIOAgZza3UKAmIAAAA=", + "payload_codec": "avro", + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:2:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 2, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + } + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "first_status": "open", + "skus": [ + "a" + ], + "second_qty": 1, + "count": 3 + } + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 496a2de2..fe741ddf 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -55,6 +55,7 @@ RedriveVisibilityCleanupWorkflow, RedriveVisibilityWorkflow, ) +from tests.test_typed_values import TypedActivityResultsWorkflow from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -415,6 +416,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: YieldedContinueAfterMetadataWorkflow, StatusRecoveryWorkflow, RemainingTimeoutApproval, + TypedActivityResultsWorkflow, ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow diff --git a/tests/test_typed_values.py b/tests/test_typed_values.py new file mode 100644 index 00000000..c604ac98 --- /dev/null +++ b/tests/test_typed_values.py @@ -0,0 +1,414 @@ +"""Opt-in typed values at the activity boundary (#184, slice 1).""" + +from __future__ import annotations + +import base64 +import json +import pickle +from dataclasses import asdict, dataclass, field +from enum import Enum, IntEnum +from pathlib import Path +from typing import Any, Literal, Optional +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import NonRetryableError, TypedValueError, activity, serializer, workflow +from durable_workflow._typed_values import from_payload_value, to_payload_value, validate_annotation +from durable_workflow.workflow import CompleteWorkflow, ScheduleActivity, WorkflowContext, replay + +try: + from enum import StrEnum +except ImportError: # Python 3.10 + class StrEnum(str, Enum): # type: ignore[no-redef] + pass + +FIXTURES = Path(__file__).parent / "fixtures" + + +class Status(Enum): + OPEN = "open" + SHIPPED = "shipped" + + +class Priority(IntEnum): + LOW = 1 + HIGH = 2 + + +class Channel(StrEnum): + WEB = "web" + STORE = "store" + + +@dataclass +class Item: + sku: str + qty: int = 1 + + +@dataclass +class Shipment: + order_id: str + status: Status + items: list[Item] + priority: Priority = Priority.LOW + channel: Channel = Channel.WEB + note: Optional[str] = None # noqa: UP045 - typing.Optional is part of the contract + weights: dict[str, float] = field(default_factory=dict) + + +SHIPMENT = Shipment("o-1", Status.SHIPPED, [Item("a", 2), Item("b")], Priority.HIGH, Channel.STORE, "gift", + {"a": 1.5}) +NEUTRAL = { + "order_id": "o-1", "status": "shipped", "items": [{"sku": "a", "qty": 2}, {"sku": "b", "qty": 1}], + "priority": 2, "channel": "store", "note": "gift", "weights": {"a": 1.5}, +} + + +def by_hand(value: Shipment) -> dict[str, Any]: + """What a user passes today: dataclasses.asdict() with each Enum's .value.""" + plain = asdict(value) + plain.update(status=value.status.value, priority=value.priority.value, channel=value.channel.value) + return plain + + +# Conversion. +def test_a_typed_value_round_trips_through_the_neutral_wire_format() -> None: + assert to_payload_value(SHIPMENT) == NEUTRAL + blob = serializer.encode(to_payload_value(SHIPMENT)) + decoded = from_payload_value(serializer.decode(blob, codec="avro"), Shipment) + assert decoded == SHIPMENT + assert type(decoded.priority) is Priority and type(decoded.channel) is Channel + + +def test_automatic_encoding_gives_the_bytes_of_asdict_and_value() -> None: + assert serializer.encode(to_payload_value(SHIPMENT)) == serializer.encode(by_hand(SHIPMENT)) + + +def test_a_newer_payload_with_an_extra_field_decodes() -> None: + newer = {**NEUTRAL, "added_in_v2": True, "items": [{"sku": "a", "qty": 2, "weight_kg": 3.0}]} + decoded = from_payload_value(newer, Shipment) + assert decoded.items == [Item("a", 2)] + assert not hasattr(decoded, "added_in_v2") + + +def test_an_older_payload_missing_optional_fields_takes_defaults() -> None: + older = {"order_id": "o-1", "status": "open", "items": [{"sku": "a"}]} + assert from_payload_value(older, Shipment) == Shipment("o-1", Status.OPEN, [Item("a")]) + + +def test_optional_accepts_null_and_scalars_keep_their_kinds() -> None: + assert from_payload_value({**NEUTRAL, "note": None}, Shipment).note is None + assert from_payload_value(3, float) == 3.0 and type(from_payload_value(3, float)) is float + assert from_payload_value(None, Optional[Item]) is None # noqa: UP045 + assert from_payload_value([{"sku": "x"}], list[Item]) == [Item("x")] + assert from_payload_value({"k": "web"}, dict[str, Channel]) == {"k": Channel.WEB} + + +@pytest.mark.parametrize(("payload", "path", "expected", "detail"), [ + ({**NEUTRAL, "items": [{"qty": 2}]}, "shipment.items[0].sku", "str", "the field is missing"), + ({k: v for k, v in NEUTRAL.items() if k != "status"}, "shipment.status", "Status", "the field is missing"), + ({**NEUTRAL, "status": "lost-in-transit"}, "shipment.status", "Status", "the string is not a member"), + ({**NEUTRAL, "priority": True}, "shipment.priority", "Priority", "the boolean is not a member"), + ({**NEUTRAL, "items": "secret-sku"}, "shipment.items", "list[", "got string"), + ({**NEUTRAL, "weights": {"a": "heavy"}}, "shipment.weights['a']", "float", "got string"), + ({**NEUTRAL, "order_id": None}, "shipment.order_id", "str", "got null"), + ({**NEUTRAL, "items": [{"sku": "a", "qty": True}]}, "shipment.items[0].qty", "int", "got boolean"), + ("secret-order", "shipment", "Shipment", "got string"), +]) +def test_a_value_that_cannot_be_converted_names_the_path_never_the_value( + payload: Any, path: str, expected: str, detail: str, +) -> None: + with pytest.raises(TypedValueError) as raised: + from_payload_value(payload, Shipment, "shipment") + assert raised.value.path == path + assert raised.value.expected.startswith(expected) + assert raised.value.detail == detail + for secret in ("secret", "lost-in-transit", "heavy"): + assert secret not in str(raised.value) + + +@dataclass +class Nested: + inner: Item + + +@pytest.mark.parametrize("annotation", [ + int | str, Optional[int | str], dict[int, str], Literal["a"], tuple[int, ...], set[str], Path, # noqa: UP045 +]) +def test_ambiguous_or_unsupported_annotations_are_refused_up_front(annotation: Any) -> None: + with pytest.raises(TypeError): + validate_annotation(annotation, "value") + + +def test_an_enum_with_non_scalar_values_is_refused() -> None: + class Pair(Enum): + A = (1, 2) + + with pytest.raises(TypeError, match="str, int or float values"): + validate_annotation(Pair, "value") + + +def test_supported_annotations_validate() -> None: + for annotation in (Shipment, Nested, list[Item], dict[str, Item], Optional[Shipment], Any, int, bytes): # noqa: UP045 + validate_annotation(annotation, "value") + + +# Typed activities. +@activity.defn(name="tests.typed.ship", typed=True) +def ship(shipment: Shipment, channel: Channel, *extra: Item) -> Shipment: + return Shipment(shipment.order_id, Status.SHIPPED, shipment.items + list(extra), channel=channel) + + +@activity.defn(name="tests.typed.untyped") +def untyped(value: Any) -> Shipment: + return SHIPMENT + + +def test_a_typed_activity_decodes_its_arguments_and_returns_the_neutral_value() -> None: + result = ship({"order_id": "o-9", "status": "open", "items": []}, "store", {"sku": "z"}) + assert result == { + "order_id": "o-9", "status": "shipped", "items": [{"sku": "z", "qty": 1}], "priority": 1, + "channel": "store", "note": None, "weights": {}, + } + + +def test_an_untyped_activity_keeps_receiving_plain_values() -> None: + assert untyped({"order_id": "o-1"}) is SHIPMENT + + +def test_a_typed_activity_argument_mismatch_is_non_retryable() -> None: + with pytest.raises(TypedValueError) as raised: + ship({"order_id": "o-9", "status": "open"}, "store") + assert raised.value.path == "shipment.items" + assert isinstance(raised.value, NonRetryableError) and isinstance(raised.value, TypeError) + with pytest.raises(TypedValueError, match=r"extra\[0\]\.sku"): + ship({"order_id": "o-9", "status": "open", "items": []}, "web", {"qty": 1}) + + +def test_a_typed_activity_with_an_unsupported_annotation_fails_when_declared() -> None: + with pytest.raises(TypeError, match="does not support|do not support"): + @activity.defn(name="tests.typed.bad", typed=True) + def bad(value: set[str]) -> None: + return None + + +def test_a_typed_activity_and_its_error_pickle_by_reference() -> None: + # Supervised callbacks pickle the activity callable by reference. + assert pickle.loads(pickle.dumps(ship)) is ship + error = TypedValueError("shipment.items", "list[Item]", "got string") + restored = pickle.loads(pickle.dumps(error)) + assert (restored.path, restored.expected, restored.detail, str(restored)) == ( + error.path, error.expected, error.detail, str(error)) + + +def _worker_and_client(fn: Any) -> tuple[Any, AsyncMock]: + from durable_workflow.client import Client + from durable_workflow.worker import Worker + + client = AsyncMock(spec=Client) + client.complete_activity_task = AsyncMock(return_value={"outcome": "completed"}) + client.fail_activity_task = AsyncMock(return_value={"outcome": "failed"}) + return Worker(client, task_queue="q1", workflows=[], activities=[fn]), client + + +def _activity_task(activity_type: str, arguments: list[Any]) -> dict[str, Any]: + return {"task_id": "t1", "activity_attempt_id": "a1", "activity_type": activity_type, + "arguments": serializer.encode(arguments, codec="avro"), "payload_codec": "avro"} + + +@pytest.mark.asyncio +async def test_a_worker_fails_a_typed_activity_mismatch_without_retries() -> None: + worker, client = _worker_and_client(ship) + status = await worker._run_activity_task(_activity_task("tests.typed.ship", [{"order_id": "o-9", "status": "open"}, + "web"])) + assert status == "failed_non_retryable" + failure = client.fail_activity_task.await_args.kwargs + assert failure["non_retryable"] is True and failure["failure_type"] == "TypedValueError" + assert failure["message"] == "shipment.items: expected list[Item]; the field is missing" + client.complete_activity_task.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_a_worker_completes_a_dataclass_result_as_its_neutral_value() -> None: + worker, client = _worker_and_client(untyped) + assert await worker._run_activity_task(_activity_task("tests.typed.untyped", [None])) == "completed" + assert client.complete_activity_task.await_args.kwargs["result"] == NEUTRAL + + +# Workflow side. +def test_scheduled_dataclass_arguments_are_sent_as_their_neutral_value() -> None: + command = ScheduleActivity("tests.typed.ship", [SHIPMENT, Channel.STORE, Status.OPEN]) + assert command.arguments == [NEUTRAL, "store", "open"] + server = command.to_server_command("queue") + assert server["arguments"] == serializer.envelope([by_hand(SHIPMENT), "store", "open"]) + + +def test_result_type_is_local_only_and_validated() -> None: + plain = ScheduleActivity("x", []) + typed = ScheduleActivity("x", [], result_type=Shipment) + assert plain == typed + assert plain.to_server_command("queue") == typed.to_server_command("queue") + assert "result_type" not in json.dumps(typed.to_server_command("queue")) + with pytest.raises(TypeError): + ScheduleActivity("x", [], result_type=int | str) + + +# The interop codec fixture pins the neutral bytes PHP and Rust read. +INTEROP = FIXTURES / "codec_regressions" / "typed-dataclass-enum-neutral-map.json" + + +@dataclass +class InteropItem: + sku: str + qty: int + + +@dataclass +class InteropOrder: + order_id: str + status: Status + priority: Priority + items: list[InteropItem] + note: Optional[str] = None # noqa: UP045 + + +INTEROP_ORDER = InteropOrder("o-1", Status.SHIPPED, Priority.HIGH, [InteropItem("a", 2)]) + + +def test_interop_fixture_pins_the_bytes_of_asdict_and_value() -> None: + wire = json.loads(INTEROP.read_text(encoding="utf-8"))["framing"]["wire_base64"] + by_hand_value = {**asdict(INTEROP_ORDER), "status": "shipped", "priority": 2} + assert serializer.encode(to_payload_value(INTEROP_ORDER)) == wire + assert serializer.encode(by_hand_value) == wire + assert base64.b64decode(wire)[:2] == b"\xc3\x01" + assert from_payload_value(serializer.decode(wire, codec="avro"), InteropOrder) == INTEROP_ORDER + + +# Replay: the corpus fixture was recorded by an untyped workflow. +def summary(first_status: str, skus: list[str], second_qty: int, count: int) -> dict[str, Any]: + return {"first_status": first_status, "skus": skus, "second_qty": second_qty, "count": count} + + +@workflow.defn(name="tests.replay.typed-activity-results") +class TypedActivityResultsWorkflow: + """Opts in to result_type for a single call and a parallel group.""" + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + first = yield ctx.schedule_activity("typed.lookup", ["o-1"], result_type=Shipment) + second, count = yield [ + ctx.schedule_activity("typed.lookup", ["o-2"], result_type=Shipment), + ctx.schedule_activity("typed.count", [], result_type=int), + ] + return summary(first.status.value, [item.sku for item in first.items], + sum(item.qty for item in second.items), count) + + +@workflow.defn(name="tests.replay.untyped-activity-results") +class UntypedActivityResultsWorkflow: + """The workflow that recorded the corpus history, reading plain maps.""" + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + first = yield ctx.schedule_activity("typed.lookup", ["o-1"]) + second, count = yield [ctx.schedule_activity("typed.lookup", ["o-2"]), ctx.schedule_activity("typed.count", [])] + return summary(first["status"], [item["sku"] for item in first["items"]], + sum(item.get("qty", 1) for item in second["items"]), count) + + +CORPUS = FIXTURES / "replay_regressions" / "typed-activity-results-opt-in.json" + + +def corpus_history() -> list[dict[str, Any]]: + return json.loads(CORPUS.read_text(encoding="utf-8"))["history"] + + +def test_an_untyped_history_replays_unchanged_with_result_type() -> None: + fixture = json.loads(CORPUS.read_text(encoding="utf-8")) + typed = replay(TypedActivityResultsWorkflow, corpus_history(), [], payload_codec="avro") + untyped_outcome = replay(UntypedActivityResultsWorkflow, corpus_history(), [], payload_codec="avro") + assert list(typed.commands) == list(untyped_outcome.commands) == [CompleteWorkflow(fixture["expected"]["result"])] + + +@workflow.defn(name="tests.typed.catches-mismatch") +class CatchesMismatchWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield ctx.schedule_activity("typed.lookup", ["o-1"], result_type=list[Item]) + except TypedValueError as error: + return {"path": error.path, "expected": error.expected} + return "decoded" + + +def test_a_result_type_mismatch_is_raised_at_the_yield() -> None: + outcome = replay(CatchesMismatchWorkflow, corpus_history()[:3], [], payload_codec="avro") + assert list(outcome.commands) == [CompleteWorkflow({"path": "activity result", "expected": "list[Item]"})] + + +@workflow.defn(name="tests.typed.parallel-mismatch") +class ParallelMismatchWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + first = yield ctx.schedule_activity("typed.lookup", ["o-1"]) + try: + yield [ctx.schedule_activity("typed.lookup", ["o-2"], result_type=Shipment), + ctx.schedule_activity("typed.count", [], result_type=str)] + except TypedValueError as error: + return {"first": first["order_id"], "path": error.path, "detail": error.detail} + return "decoded" + + +def test_a_parallel_member_mismatch_is_raised_at_the_group_yield() -> None: + outcome = replay(ParallelMismatchWorkflow, corpus_history(), [], payload_codec="avro") + assert list(outcome.commands) == [ + CompleteWorkflow({"first": "o-1", "path": "activity result", "detail": "got integer"}), + ] + + +def _selection_history() -> list[dict[str, Any]]: + from tests.test_durable_selection import _activity_completed, _activity_scheduled, _winner_marker + + return [ + _activity_scheduled(0, "slow"), + _activity_scheduled(1, "fast"), + _activity_completed(1, "fast", {"sku": "fast-sku", "unknown": True}), + _winner_marker(), + _activity_completed(0, "slow", {"sku": "slow-sku", "qty": 4}), + ] + + +@workflow.defn(name="tests.typed.selection") +class TypedSelectionWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + selected = yield ctx.select({ + "slow": ctx.schedule_activity("slow-activity", [], result_type=Item), + "fast": ctx.schedule_activity("fast-activity", [], result_type=Item), + }) + slow = yield selected.handles["slow"].await_result() + return {"winner": selected.key, "item": asdict(selected.result()), "slow": asdict(slow)} + + +@workflow.defn(name="tests.typed.selection-mismatch") +class MismatchedSelectionWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + selected = yield ctx.select({ + "slow": ctx.schedule_activity("slow-activity", [], result_type=int), + "fast": ctx.schedule_activity("fast-activity", [], result_type=int), + }) + assert isinstance(selected.failure, TypedValueError) + try: + yield selected.handles["slow"].await_result() + except TypedValueError as error: + return {"winner": selected.key, "winner_detail": selected.failure.detail, "slow_detail": error.detail} + return "decoded" + + +def test_selection_members_are_decoded_to_their_result_type() -> None: + (command,) = replay(TypedSelectionWorkflow, _selection_history(), [], payload_codec="avro").commands + assert command == CompleteWorkflow({ + "winner": "fast", "item": {"sku": "fast-sku", "qty": 1}, "slow": {"sku": "slow-sku", "qty": 4}, + }) + + +def test_a_selection_mismatch_is_the_winner_failure_and_raised_at_a_handle_await() -> None: + (command,) = replay(MismatchedSelectionWorkflow, _selection_history(), [], payload_codec="avro").commands + assert command == CompleteWorkflow({"winner": "fast", "winner_detail": "got object", "slow_detail": "got object"})