From 6b23a6e045bff12db6a85979935b7b6363dd1b95 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 22:34:15 -0600 Subject: [PATCH 1/4] WIP #184 slice 2: workflow.defn(typed=True) and typed_method --- src/durable_workflow/_typed_values.py | 60 ++++++++++++++++++++++----- src/durable_workflow/workflow.py | 31 +++++++++++++- 2 files changed, 80 insertions(+), 11 deletions(-) diff --git a/src/durable_workflow/_typed_values.py b/src/durable_workflow/_typed_values.py index 3f43de4..de76e58 100644 --- a/src/durable_workflow/_typed_values.py +++ b/src/durable_workflow/_typed_values.py @@ -255,22 +255,16 @@ 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. - """ +def _positional_decoder(fn: Callable[..., Any], skip: int, where: str) -> Callable[[tuple[Any, ...]], list[Any]]: + """Build a decoder for *fn*'s positional arguments after the first *skip*.""" 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) @@ -292,6 +286,18 @@ 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. + """ + convert = _positional_decoder(fn, 0, "typed activity") wrapper: Any if inspect.iscoroutinefunction(fn): @@ -309,3 +315,37 @@ 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. + """ + convert = _positional_decoder(fn, skip, "typed workflow method") + 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) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index a017aec..468de7d 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -189,16 +189,45 @@ def _validate_workflow_handler_result(result: Any, cls: type, method: str, *, al return result -def defn(*, name: str) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]: +_TYPED_HANDLER_KINDS = ("signal", "query", "update", "update_validator") + + +def _type_workflow_methods(cls: type) -> None: + """Wrap run and every handler so their arguments decode to annotations.""" + from ._typed_values import typed_method + + entrypoint = getattr(cls, "run", None) + if inspect.isfunction(entrypoint): + cls.run = typed_method(entrypoint, 2) # type: ignore[attr-defined] + for attr in dir(cls): + if attr.startswith("_"): + continue + member = getattr(cls, attr, None) + if inspect.isfunction(member) and any( + isinstance(getattr(member, f"__{kind}_name__", None), str) for kind in _TYPED_HANDLER_KINDS + ): + setattr(cls, attr, typed_method(member, 1)) + + +def defn(*, name: str, typed: bool = False) -> Callable[[type[_WorkflowT]], type[_WorkflowT]]: """Register a class as a workflow type under a language-neutral name. Scans the class for ``@signal``, ``@query``, and ``@update`` decorated methods and builds registries at decoration time so worker-side dispatch can use stable receiver names without re-inspecting the class on every history event or control-plane request. + + With ``typed=True``, the arguments of ``run`` (after ``ctx``) and of every + signal, query, update and update-validator handler are decoded to their + annotations, with the rules :func:`durable_workflow.activity.defn` + documents for ``typed=True``. A mismatch raises + :class:`~durable_workflow.errors.TypedValueError` when the method is + called. Without it, methods keep receiving plain values. """ def wrap(cls: type[_WorkflowT]) -> type[_WorkflowT]: + if typed: + _type_workflow_methods(cls) cls.__workflow_name__ = name # type: ignore[attr-defined] _validate_workflow_entrypoint(cls) signals, queries, updates, update_validators = _workflow_handler_registries(cls) From 43bc24fe71752000caf5e70e9ffef98f26730884 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 22:40:16 -0600 Subject: [PATCH 2/4] Add opt-in typed workflow, child and client values (#184 slice 2) workflow.defn(typed=True) decodes run, signal, query, update and update-validator arguments to their annotations; start_child_workflow (result_type=) and WorkflowHandle.result / Client.get_result (result_type=) decode results. Dataclass and Enum values in client argument payloads, child and continue-as-new arguments, and workflow and update results are sent as their neutral values at encoding time; commands keep the authored value, so the test harness is unchanged. A replay corpus fixture recorded by an untyped workflow replays to the same completion under typed=True. Refs #184. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + docs/sdk-reference.md | 9 + src/durable_workflow/client.py | 32 ++- src/durable_workflow/workflow.py | 31 ++- .../typed-workflow-values-opt-in.json | 107 +++++++++ tests/test_replay_regression_corpus.py | 2 + tests/test_typed_workflow_values.py | 203 ++++++++++++++++++ 7 files changed, 375 insertions(+), 10 deletions(-) create mode 100644 tests/fixtures/replay_regressions/typed-workflow-values-opt-in.json create mode 100644 tests/test_typed_workflow_values.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9aed71b..fbeea53 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. +- 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.13.2] - 2026-11-04 diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 26ebb71..043f16a 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -1020,6 +1020,15 @@ 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. Supported annotations: dataclasses, `Enum` (including `IntEnum` and `StrEnum`, which come back as their class), `Optional[T]` / `T | None`, `list[T]`, diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index abb6066..1f96f9b 100644 --- a/src/durable_workflow/client.py +++ b/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 @@ def from_dict(cls, data: dict[str, Any]) -> BridgeAdapterOutcome: ) +# 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 @@ def __init__(self, client: Client, workflow_id: str, run_id: str | None = None, 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 @@ def _payload_envelope( 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,9 +4774,16 @@ async def get_result( *, 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 @@ -4770,6 +4791,8 @@ async def get_result( 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 @@ async def get_result( 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"), diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 468de7d..eed85f4 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -44,7 +44,7 @@ ) from ._cooperative_cancellation import CancellationDelivery, read_cancellation_history from ._failure_reporting import exception_message -from ._typed_values import decode_payload_value, to_payload_values, validate_annotation +from ._typed_values import decode_payload_value, to_payload_value, to_payload_values, validate_annotation from .cancellation import ( CancellationContext, CancellationPolicy, @@ -962,7 +962,7 @@ def to_server_command( return { "type": "complete_workflow", "result": _payload_envelope( - self.result, + to_payload_value(self.result), payload_codec=payload_codec, size_warning=size_warning, warning_context=_payload_warning_context( @@ -1035,7 +1035,7 @@ def to_server_command( "type": "complete_update", "update_id": self.update_id, "result": _payload_envelope( - self.result, + to_payload_value(self.result), payload_codec=payload_codec, size_warning=size_warning, warning_context=_payload_warning_context( @@ -1105,7 +1105,7 @@ def to_server_command( if self.workflow_type is not None: cmd["workflow_type"] = self.workflow_type cmd["arguments"] = _payload_envelope( - self.arguments, + to_payload_values(list(self.arguments)), payload_codec=payload_codec, size_warning=size_warning, warning_context=_payload_warning_context( @@ -1175,6 +1175,8 @@ class StartChildWorkflow: execution_timeout_seconds: int | None = None run_timeout_seconds: int | None = None cancellation_policy: str | CancellationPolicy | None = None + # Opt-in local decoding of the child's result; never sent or compared. + 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, ) @@ -1186,6 +1188,10 @@ class StartChildWorkflow: ) def __post_init__(self) -> None: + # Dataclass and Enum arguments travel as their neutral values. + self.arguments = to_payload_values(list(self.arguments)) + if self.result_type is not None: + validate_annotation(self.result_type, "result_type") _canonical_child_policies({ "parent_close_policy": self.parent_close_policy, "cancellation_policy": self.cancellation_policy, @@ -2676,7 +2682,14 @@ def start_child_workflow( retry_policy: ChildWorkflowRetryPolicyInput | None = None, execution_timeout_seconds: int | None = None, run_timeout_seconds: int | None = None, + result_type: Any = None, ) -> StartChildWorkflow: + """Yield a child workflow start. + + Dataclass and ``Enum`` arguments are sent as their neutral values. + ``result_type`` opts in to decoding the child's result, with the same + rules and failure behavior as ``schedule_activity(result_type=...)``. + """ return StartChildWorkflow( workflow_type=workflow_type, arguments=list(arguments) if arguments is not None else [], @@ -2686,6 +2699,7 @@ def start_child_workflow( retry_policy=retry_policy, execution_timeout_seconds=execution_timeout_seconds, run_timeout_seconds=run_timeout_seconds, + result_type=result_type, ) def call_nexus_service( @@ -3154,10 +3168,13 @@ def _history_event_time(event: Mapping[str, Any]) -> datetime | 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): + """Decode an activity or child result to the call's opt-in result_type.""" + if not isinstance(command, ScheduleActivity | StartChildWorkflow) or command.result_type is None or isinstance( + value, BaseException, + ): return value - return decode_payload_value(value, command.result_type, "activity result") + path = "activity result" if isinstance(command, ScheduleActivity) else "child workflow result" + return decode_payload_value(value, command.result_type, path) def _decode_history_result( diff --git a/tests/fixtures/replay_regressions/typed-workflow-values-opt-in.json b/tests/fixtures/replay_regressions/typed-workflow-values-opt-in.json new file mode 100644 index 0000000..64beedd --- /dev/null +++ b/tests/fixtures/replay_regressions/typed-workflow-values-opt-in.json @@ -0,0 +1,107 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "typed-workflow-values-opt-in", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.typed-workflow-values", + "input": [ + { + "name": "Ada", + "tier": "gold", + "tags": [ + "vip" + ] + } + ], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.typed-workflow-values" + }, + "sequence": 1 + }, + { + "event_type": "ChildWorkflowScheduled", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "sequence": 1, + "child_workflow_type": "tests.accounts.open" + }, + "sequence": 2 + }, + { + "event_type": "ChildRunStarted", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "sequence": 1, + "child_workflow_type": "tests.accounts.open" + }, + "sequence": 3 + }, + { + "event_type": "ChildRunCompleted", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "sequence": 1, + "child_workflow_type": "tests.accounts.open", + "output": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4ICG5hbWUKBkFkYQh0aWVyCghnb2xkCHRhZ3MMAgoGdmlwABRhY2NvdW50X2lkCgxhY2N0LTEA" + }, + "payload_codec": "avro" + }, + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "approval-1", + "condition_key": "approval", + "sequence": 2 + }, + "sequence": 5 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCDgQQYXBwcm92ZXIKCkdyYWNlFmFkZGVkX2luX3YyAgEAAA==" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "sequence": 6 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "condition_wait_id": "approval-1", + "condition_key": "approval", + "sequence": 2 + }, + "sequence": 7 + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "name": "Ada", + "tier": "gold", + "approver": "Grace", + "note": null + } + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index fe741dd..3ab1289 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -56,6 +56,7 @@ RedriveVisibilityWorkflow, ) from tests.test_typed_values import TypedActivityResultsWorkflow +from tests.test_typed_workflow_values import TypedOnboarding from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -417,6 +418,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: StatusRecoveryWorkflow, RemainingTimeoutApproval, TypedActivityResultsWorkflow, + TypedOnboarding, ] 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_workflow_values.py b/tests/test_typed_workflow_values.py new file mode 100644 index 0000000..5c46315 --- /dev/null +++ b/tests/test_typed_workflow_values.py @@ -0,0 +1,203 @@ +"""Opt-in typed values at the workflow, child and client boundaries (#184, slice 2).""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field +from enum import Enum +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import TypedValueError, serializer, workflow +from durable_workflow.workflow import ( + CompleteUpdate, + CompleteWorkflow, + ContinueAsNew, + FailWorkflow, + StartChildWorkflow, + WorkflowContext, + apply_update, + query_state, + replay, +) + +CORPUS = Path(__file__).parent / "fixtures" / "replay_regressions" / "typed-workflow-values-opt-in.json" + + +class Tier(Enum): + BASIC = "basic" + GOLD = "gold" + + +@dataclass +class Customer: + name: str + tier: Tier = Tier.BASIC + tags: list[str] = field(default_factory=list) + + +@dataclass +class Approval: + approver: str + note: str | None = None + + +ADA = Customer("Ada", Tier.GOLD, ["vip"]) +ADA_NEUTRAL = {"name": "Ada", "tier": "gold", "tags": ["vip"]} + + +@workflow.defn(name="tests.replay.typed-workflow-values", typed=True) +class TypedOnboarding: + """Typed run, signal, update and query arguments, and a typed child result.""" + + def __init__(self) -> None: + self.customer: Customer | None = None + self.approval: Approval | None = None + + @workflow.signal("approve") + def approve(self, approval: Approval) -> None: + self.approval = approval + + @workflow.update("retier") + def retier(self, tier: Tier) -> str: + assert self.customer is not None + self.customer.tier = tier + return tier.value + + @workflow.query("describe") + def describe(self, field_name: str) -> Any: + assert self.customer is not None + return getattr(self.customer, field_name) + + def run(self, ctx: WorkflowContext, customer: Customer): # type: ignore[no-untyped-def] + self.customer = customer + account = yield ctx.start_child_workflow("tests.accounts.open", [customer], result_type=Customer) + yield ctx.wait_condition(lambda: self.approval is not None, key="approval") + assert self.approval is not None + return {"name": account.name, "tier": account.tier.value, "approver": self.approval.approver, + "note": self.approval.note} + + +@workflow.defn(name="tests.replay.untyped-workflow-values") +class UntypedOnboarding: + """The workflow that recorded the corpus history, reading plain maps.""" + + def __init__(self) -> None: + self.approval: dict[str, Any] | None = None + + @workflow.signal("approve") + def approve(self, approval: dict[str, Any]) -> None: + self.approval = approval + + def run(self, ctx: WorkflowContext, customer: dict[str, Any]): # type: ignore[no-untyped-def] + account = yield ctx.start_child_workflow("tests.accounts.open", [customer]) + yield ctx.wait_condition(lambda: self.approval is not None, key="approval") + assert self.approval is not None + return {"name": account["name"], "tier": account["tier"], "approver": self.approval["approver"], + "note": self.approval.get("note")} + + +def corpus() -> dict[str, Any]: + return json.loads(CORPUS.read_text(encoding="utf-8")) + + +def test_an_untyped_history_replays_unchanged_with_typed_true() -> None: + fixture = corpus() + typed = replay(TypedOnboarding, fixture["history"], fixture["workflow"]["input"], payload_codec="avro") + untyped = replay(UntypedOnboarding, fixture["history"], fixture["workflow"]["input"], payload_codec="avro") + expected = [CompleteWorkflow(fixture["expected"]["result"])] + assert list(typed.commands) == list(untyped.commands) == expected + + +def test_typed_query_and_update_arguments_are_decoded() -> None: + fixture = corpus() + history, start = fixture["history"], fixture["workflow"]["input"] + assert query_state(TypedOnboarding, history, start, "describe", ["tier"], payload_codec="avro") == Tier.GOLD + accepted = {"event_type": "UpdateAccepted", "payload": { + "update_id": "update-1", "update_name": "retier", + "arguments": serializer.encode(["basic"], codec="avro"), "payload_codec": "avro"}} + result = apply_update(TypedOnboarding, [*history[:-2], accepted], start, "update-1", payload_codec="avro") + assert result == CompleteUpdate("update-1", "basic") + + +def test_a_typed_run_argument_mismatch_fails_the_workflow_without_the_value() -> None: + (command,) = replay(TypedOnboarding, [{"event_type": "WorkflowStarted", "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": {}}], [{"tier": "gold", "secret": "s3cr3t"}], + payload_codec="avro").commands + assert isinstance(command, FailWorkflow) + assert command.exception_type == "TypedValueError" + assert "customer.name" in command.message and "s3cr3t" not in command.message + + +def test_typed_workflow_methods_keep_their_signatures_and_registries() -> None: + import inspect + + assert list(inspect.signature(TypedOnboarding.run).parameters) == ["self", "ctx", "customer"] + assert TypedOnboarding.__workflow_signals__ == {"approve": "approve"} # type: ignore[attr-defined] + assert TypedOnboarding.__workflow_updates__ == {"retier": "retier"} # type: ignore[attr-defined] + + +def test_an_unsupported_handler_annotation_fails_when_the_workflow_is_declared() -> None: + with pytest.raises(TypeError): + @workflow.defn(name="tests.typed.bad-signal", typed=True) + class BadSignal: + @workflow.signal("bad") + def bad(self, value: set[str]) -> None: + return None + + def run(self, ctx: WorkflowContext) -> None: + return None + + +def test_child_arguments_and_terminal_values_travel_as_neutral_bytes() -> None: + child = StartChildWorkflow("tests.accounts.open", [ADA]) + assert child.arguments == [ADA_NEUTRAL] + assert child.to_server_command("q")["arguments"] == serializer.envelope([asdict(ADA) | {"tier": "gold"}]) + assert child == StartChildWorkflow("tests.accounts.open", [ADA], result_type=Customer) + with pytest.raises(TypeError): + StartChildWorkflow("tests.accounts.open", [], result_type=int | str) + completed = CompleteWorkflow(ADA).to_server_command("q") + assert completed["result"] == serializer.envelope(ADA_NEUTRAL) + assert CompleteUpdate("u", Tier.GOLD).to_server_command("q")["result"] == serializer.envelope("gold") + assert ContinueAsNew(arguments=[ADA]).to_server_command("q")["arguments"] == serializer.envelope([ADA_NEUTRAL]) + # The command keeps the authored value; only the wire form is neutral. + assert CompleteWorkflow(ADA).result is ADA + + +def test_a_child_result_type_mismatch_is_raised_at_the_yield() -> None: + @workflow.defn(name="tests.typed.child-mismatch") + class ChildMismatch: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield ctx.start_child_workflow("tests.accounts.open", [], result_type=list[Customer]) + except TypedValueError as error: + return {"path": error.path, "detail": error.detail} + return "decoded" + + history = [event for event in corpus()["history"] if event["event_type"] in { + "WorkflowStarted", "ChildWorkflowScheduled", "ChildRunStarted", "ChildRunCompleted"}] + (command,) = replay(ChildMismatch, history, [], payload_codec="avro").commands + assert command == CompleteWorkflow({"path": "child workflow result", "detail": "got object"}) + + +@pytest.mark.asyncio +async def test_the_client_decodes_a_result_type_and_sends_neutral_arguments() -> None: + from durable_workflow.client import Client, WorkflowExecution, WorkflowHandle + + client = Client("http://server.invalid", token="t") + client.describe_workflow = AsyncMock(return_value=WorkflowExecution( # type: ignore[method-assign] + workflow_id="wf", run_id="run", workflow_type="x", status="completed", payload_codec="avro")) + client.get_history = AsyncMock(return_value={"events": [ # type: ignore[method-assign] + {"event_type": "WorkflowCompleted", "payload": {"output": serializer.envelope(ADA_NEUTRAL), + "payload_codec": "avro"}}]}) + handle = WorkflowHandle(client, "wf", "run") + assert await handle.result(result_type=Customer, timeout=1) == ADA + assert await handle.result(timeout=1) == ADA_NEUTRAL + with pytest.raises(TypedValueError, match="workflow result: expected list"): + await handle.result(result_type=list[Customer], timeout=1) + envelope = client._payload_envelope([ADA, Tier.GOLD], kind="workflow_input", workflow_id="wf", task_queue="q") + assert envelope == serializer.envelope([ADA_NEUTRAL, "gold"]) + await client.aclose() From 9b50b645b5875a069b6a10cec6f083923882ed37 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:08:03 -0600 Subject: [PATCH 3/4] Keep predicate fingerprints under typed=True and refuse static handlers A handler with nothing to decode stays unwrapped, and a typed wrapper is fingerprinted as the method it wraps, so turning typed=True on no longer changes a recorded wait_condition predicate (fails at 43bc24fe with condition_fingerprint_changed). staticmethod and classmethod handlers are refused under typed=True. Docs list per-surface mismatch behaviour and the in-flight dataclass change caveat; the untyped corpus twin annotates its dataclasses so a leak is caught. Refs #184. Co-Authored-By: Claude Opus 5.5 --- docs/sdk-reference.md | 25 +++++ src/durable_workflow/_typed_values.py | 25 ++++- src/durable_workflow/workflow.py | 15 ++- .../typed-predicate-fingerprint-opt-in.json | 101 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 3 +- tests/test_typed_workflow_values.py | 62 ++++++++++- 6 files changed, 220 insertions(+), 11 deletions(-) create mode 100644 tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 043f16a..c746b5a 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -1030,6 +1030,31 @@ 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]`, `dict[str, T]`, `str`, `int`, `float` (an integer is accepted), `bool`, `bytes` diff --git a/src/durable_workflow/_typed_values.py b/src/durable_workflow/_typed_values.py index de76e58..78baf7f 100644 --- a/src/durable_workflow/_typed_values.py +++ b/src/durable_workflow/_typed_values.py @@ -255,8 +255,14 @@ def to_payload_values(values: list[Any]) -> list[Any]: return [to_payload_value(value) for value in values] -def _positional_decoder(fn: Callable[..., Any], skip: int, where: str) -> Callable[[tuple[Any, ...]], list[Any]]: - """Build a decoder for *fn*'s positional arguments after the first *skip*.""" +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*. + + 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) @@ -273,6 +279,10 @@ def _positional_decoder(fn: Callable[..., Any], skip: int, where: str) -> Callab 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): @@ -297,7 +307,11 @@ def typed_callable(fn: Callable[..., Any]) -> Callable[..., Any]: the decorator binds that name to the wrapper, so supervised callbacks that pickle by reference resolve to it. """ - convert = _positional_decoder(fn, 0, "typed activity") + 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): @@ -323,8 +337,13 @@ def typed_method(fn: Callable[..., Any], skip: int) -> Callable[..., Any]: ``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): diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index eed85f4..f6ed6bb 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -203,9 +203,13 @@ def _type_workflow_methods(cls: type) -> None: if attr.startswith("_"): continue member = getattr(cls, attr, None) - if inspect.isfunction(member) and any( - isinstance(getattr(member, f"__{kind}_name__", None), str) for kind in _TYPED_HANDLER_KINDS - ): + if not any(isinstance(getattr(member, f"__{kind}_name__", None), str) for kind in _TYPED_HANDLER_KINDS): + continue + if isinstance(inspect.getattr_static(cls, attr), staticmethod | classmethod): + raise TypeError( + f"typed workflow {cls.__qualname__}.{attr}: handlers must be instance methods to decode arguments" + ) + if inspect.isfunction(member): setattr(cls, attr, typed_method(member, 1)) @@ -1761,7 +1765,10 @@ def _condition_predicate_fingerprint(predicate: Callable[[], bool]) -> str: h.update(f"{getattr(predicate, '__module__', '')}\0".encode()) h.update(f"{getattr(predicate, '__qualname__', '')}\0".encode()) - code = getattr(predicate, "__code__", None) + # A typed=True wrapper is fingerprinted as the method it wraps, so turning + # typed on does not change a recorded predicate's identity (#184). + original = getattr(predicate, "__typed_workflow_method__", None) + code = getattr(original if original is not None else predicate, "__code__", None) if code is None: h.update(repr(predicate).encode()) else: diff --git a/tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json b/tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json new file mode 100644 index 0000000..147f6b9 --- /dev/null +++ b/tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json @@ -0,0 +1,101 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "typed-predicate-fingerprint-opt-in", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.typed-predicate-fingerprint", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.typed-predicate-fingerprint" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:01+00:00", + "payload": { + "condition_wait_id": "done-1", + "condition_key": "done", + "sequence": 1, + "condition_definition_fingerprint": "sha256:33fa64522b9d27e8504cbcf473ccbdbf4670de42e8ed9c6c4e10b682c7b5116b" + }, + "sequence": 2 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCDgIQYXBwcm92ZXIKCkdyYWNlAAA=" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:02+00:00", + "payload": { + "condition_wait_id": "done-1", + "condition_key": "done", + "sequence": 1 + }, + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:03+00:00", + "payload": { + "condition_wait_id": "ready-1", + "condition_key": "ready", + "sequence": 2, + "condition_definition_fingerprint": "sha256:c6d32cfa09110013cbbdd7c71169ff78d16069765dbb22dda6b18a5610b0a9ec" + }, + "sequence": 5 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "signal_name": "approve", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCDgQQYXBwcm92ZXIKBkFkYQhub3RlCgRvawAA" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "sequence": 6 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:04+00:00", + "payload": { + "condition_wait_id": "ready-1", + "condition_key": "ready", + "sequence": 2 + }, + "sequence": 7 + } + ], + "expected": { + "type": "complete_workflow", + "result": [ + "Grace", + "Ada" + ] + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 3ab1289..68932c3 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -56,7 +56,7 @@ RedriveVisibilityWorkflow, ) from tests.test_typed_values import TypedActivityResultsWorkflow -from tests.test_typed_workflow_values import TypedOnboarding +from tests.test_typed_workflow_values import TypedOnboarding, TypedPredicate from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -419,6 +419,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: RemainingTimeoutApproval, TypedActivityResultsWorkflow, TypedOnboarding, + TypedPredicate, ] 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_workflow_values.py b/tests/test_typed_workflow_values.py index 5c46315..c885b55 100644 --- a/tests/test_typed_workflow_values.py +++ b/tests/test_typed_workflow_values.py @@ -89,10 +89,11 @@ def __init__(self) -> None: self.approval: dict[str, Any] | None = None @workflow.signal("approve") - def approve(self, approval: dict[str, Any]) -> None: - self.approval = approval + def approve(self, approval: Approval) -> None: + # Annotated but untyped: it must still receive the plain map. + self.approval = approval # type: ignore[assignment] - def run(self, ctx: WorkflowContext, customer: dict[str, Any]): # type: ignore[no-untyped-def] + def run(self, ctx: WorkflowContext, customer: Customer): # type: ignore[no-untyped-def] account = yield ctx.start_child_workflow("tests.accounts.open", [customer]) yield ctx.wait_condition(lambda: self.approval is not None, key="approval") assert self.approval is not None @@ -201,3 +202,58 @@ async def test_the_client_decodes_a_result_type_and_sends_neutral_arguments() -> envelope = client._payload_envelope([ADA, Tier.GOLD], kind="workflow_input", workflow_id="wf", task_queue="q") assert envelope == serializer.envelope([ADA_NEUTRAL, "gold"]) await client.aclose() + + +# A predicate's recorded fingerprint must not change when typed=True is turned +# on (#255 review): zero-arg handlers stay unwrapped, and a wrapped handler is +# fingerprinted as the method it wraps. +@workflow.defn(name="tests.replay.typed-predicate-fingerprint", typed=True) +class TypedPredicate: + def __init__(self) -> None: + self.approvals: list[Approval] = [] + + @workflow.signal("approve") + def approve(self, approval: Approval) -> None: + self.approvals.append(approval) + + @workflow.query("is_done") + def is_done(self) -> bool: + return bool(self.approvals) + + @workflow.query("ready") + def ready(self, minimum: int = 2) -> bool: + return len(self.approvals) >= minimum + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.wait_condition(self.is_done, key="done") + yield ctx.wait_condition(self.ready, key="ready") + return [approval.approver for approval in self.approvals] + + +PREDICATE_CORPUS = CORPUS.parent / "typed-predicate-fingerprint-opt-in.json" + + +def test_typed_true_keeps_recorded_predicate_fingerprints() -> None: + from durable_workflow.workflow import _condition_predicate_fingerprint + + instance = TypedPredicate() + # A zero-arg handler is not wrapped at all. + assert not hasattr(TypedPredicate.is_done, "__typed_workflow_method__") + # A wrapped handler keeps the fingerprint of the method it wraps. + original = TypedPredicate.ready.__typed_workflow_method__ # type: ignore[attr-defined] + assert _condition_predicate_fingerprint(instance.ready) == _condition_predicate_fingerprint( + original.__get__(instance)) + fixture = json.loads(PREDICATE_CORPUS.read_text(encoding="utf-8")) + (command,) = replay(TypedPredicate, fixture["history"], [], payload_codec="avro").commands + assert command == CompleteWorkflow(fixture["expected"]["result"]) + + +@pytest.mark.parametrize("kind", [staticmethod, classmethod]) +def test_static_and_class_method_handlers_are_refused_when_typed(kind: Any) -> None: + def approve(*args: Any) -> None: + return None + + approve.__signal_name__ = "approve" # type: ignore[attr-defined] + namespace = {"approve": kind(approve), "run": lambda self, ctx: None} + with pytest.raises(TypeError, match="handlers must be instance methods"): + workflow.defn(name=f"tests.typed.{kind.__name__}-handler", typed=True)(type("StaticHandler", (), namespace)) From 9ac0210233ec093c598ac5f1e4d8755619c76e62 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:24:32 -0600 Subject: [PATCH 4/4] Build the predicate-fingerprint evidence at run time, not as a fixture Predicate fingerprints hash bytecode, which differs between Python versions, so the 3.13-recorded corpus fixture failed on 3.10-3.15 in CI. The test now builds the untyped history with fingerprints of the unwrapped methods (inspect.unwrap) on the running Python; it still fails at 43bc24fe with condition_fingerprint_changed and passes here. Refs #184. Co-Authored-By: Claude Opus 5.5 --- .../typed-predicate-fingerprint-opt-in.json | 101 ------------------ tests/test_replay_regression_corpus.py | 3 +- tests/test_typed_workflow_values.py | 47 +++++++- 3 files changed, 44 insertions(+), 107 deletions(-) delete mode 100644 tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json diff --git a/tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json b/tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json deleted file mode 100644 index 147f6b9..0000000 --- a/tests/fixtures/replay_regressions/typed-predicate-fingerprint-opt-in.json +++ /dev/null @@ -1,101 +0,0 @@ -{ - "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", - "fixture_schema": "durable-workflow.replay-regression/v1", - "id": "typed-predicate-fingerprint-opt-in", - "protocol_version": "1.20", - "bindings": [ - "python" - ], - "workflow": { - "type": "tests.replay.typed-predicate-fingerprint", - "input": [], - "payload_codec": "avro" - }, - "history": [ - { - "event_type": "WorkflowStarted", - "recorded_at": "2026-11-04T00:00:00+00:00", - "payload": { - "workflow_type": "tests.replay.typed-predicate-fingerprint" - }, - "sequence": 1 - }, - { - "event_type": "ConditionWaitOpened", - "recorded_at": "2026-11-04T00:00:01+00:00", - "payload": { - "condition_wait_id": "done-1", - "condition_key": "done", - "sequence": 1, - "condition_definition_fingerprint": "sha256:33fa64522b9d27e8504cbcf473ccbdbf4670de42e8ed9c6c4e10b682c7b5116b" - }, - "sequence": 2 - }, - { - "event_type": "SignalReceived", - "recorded_at": "2026-11-04T00:00:02+00:00", - "payload": { - "signal_name": "approve", - "value": { - "codec": "avro", - "blob": "wwHioz3/VYAiNwwCDgIQYXBwcm92ZXIKCkdyYWNlAAA=" - }, - "payload_codec": "avro", - "workflow_sequence": 1 - }, - "sequence": 3 - }, - { - "event_type": "ConditionWaitSatisfied", - "recorded_at": "2026-11-04T00:00:02+00:00", - "payload": { - "condition_wait_id": "done-1", - "condition_key": "done", - "sequence": 1 - }, - "sequence": 4 - }, - { - "event_type": "ConditionWaitOpened", - "recorded_at": "2026-11-04T00:00:03+00:00", - "payload": { - "condition_wait_id": "ready-1", - "condition_key": "ready", - "sequence": 2, - "condition_definition_fingerprint": "sha256:c6d32cfa09110013cbbdd7c71169ff78d16069765dbb22dda6b18a5610b0a9ec" - }, - "sequence": 5 - }, - { - "event_type": "SignalReceived", - "recorded_at": "2026-11-04T00:00:04+00:00", - "payload": { - "signal_name": "approve", - "value": { - "codec": "avro", - "blob": "wwHioz3/VYAiNwwCDgQQYXBwcm92ZXIKBkFkYQhub3RlCgRvawAA" - }, - "payload_codec": "avro", - "workflow_sequence": 2 - }, - "sequence": 6 - }, - { - "event_type": "ConditionWaitSatisfied", - "recorded_at": "2026-11-04T00:00:04+00:00", - "payload": { - "condition_wait_id": "ready-1", - "condition_key": "ready", - "sequence": 2 - }, - "sequence": 7 - } - ], - "expected": { - "type": "complete_workflow", - "result": [ - "Grace", - "Ada" - ] - } -} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index d846652..b799968 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -56,7 +56,7 @@ RedriveVisibilityWorkflow, ) from tests.test_typed_values import TypedActivityResultsWorkflow -from tests.test_typed_workflow_values import TypedOnboarding, TypedPredicate +from tests.test_typed_workflow_values import TypedOnboarding from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -424,7 +424,6 @@ def run(self, ctx: WorkflowContext, message: str) -> None: OrphanedDuplicateDecision, TypedActivityResultsWorkflow, TypedOnboarding, - TypedPredicate, ] 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_workflow_values.py b/tests/test_typed_workflow_values.py index c885b55..5e87ff7 100644 --- a/tests/test_typed_workflow_values.py +++ b/tests/test_typed_workflow_values.py @@ -230,7 +230,44 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] return [approval.approver for approval in self.approvals] -PREDICATE_CORPUS = CORPUS.parent / "typed-predicate-fingerprint-opt-in.json" +def _untyped_predicate_history() -> list[dict[str, Any]]: + """The history an untyped deploy of TypedPredicate records on this Python. + + Fingerprints hash the predicate's bytecode, which differs between Python + versions, so they are computed from the unwrapped methods at run time + rather than stored in a corpus fixture. + """ + import inspect + + from durable_workflow.workflow import _condition_predicate_fingerprint + + # inspect.unwrap reaches the authored method whether or not typed=True + # wrapped it, which is what an untyped deploy fingerprints. + instance = TypedPredicate() + done = _condition_predicate_fingerprint(inspect.unwrap(TypedPredicate.is_done).__get__(instance)) + ready = _condition_predicate_fingerprint(inspect.unwrap(TypedPredicate.ready).__get__(instance)) + at = "2026-11-04T00:00:0{}+00:00" + + def approve(value: dict[str, Any], sequence: int, second: int) -> dict[str, Any]: + return {"event_type": "SignalReceived", "recorded_at": at.format(second), "payload": { + "signal_name": "approve", "value": serializer.envelope([value]), "payload_codec": "avro", + "workflow_sequence": sequence}} + + def opened(wait_id: str, key: str, sequence: int, fingerprint: str, second: int) -> dict[str, Any]: + return {"event_type": "ConditionWaitOpened", "recorded_at": at.format(second), "payload": { + "condition_wait_id": wait_id, "condition_key": key, "sequence": sequence, + "condition_definition_fingerprint": fingerprint}} + + def satisfied(wait_id: str, key: str, sequence: int, second: int) -> dict[str, Any]: + return {"event_type": "ConditionWaitSatisfied", "recorded_at": at.format(second), "payload": { + "condition_wait_id": wait_id, "condition_key": key, "sequence": sequence}} + + return [ + {"event_type": "WorkflowStarted", "recorded_at": at.format(0), "payload": {}}, + opened("done-1", "done", 1, done, 1), approve({"approver": "Grace"}, 1, 2), satisfied("done-1", "done", 1, 2), + opened("ready-1", "ready", 2, ready, 3), approve({"approver": "Ada", "note": "ok"}, 2, 4), + satisfied("ready-1", "ready", 2, 4), + ] def test_typed_true_keeps_recorded_predicate_fingerprints() -> None: @@ -243,9 +280,11 @@ def test_typed_true_keeps_recorded_predicate_fingerprints() -> None: original = TypedPredicate.ready.__typed_workflow_method__ # type: ignore[attr-defined] assert _condition_predicate_fingerprint(instance.ready) == _condition_predicate_fingerprint( original.__get__(instance)) - fixture = json.loads(PREDICATE_CORPUS.read_text(encoding="utf-8")) - (command,) = replay(TypedPredicate, fixture["history"], [], payload_codec="avro").commands - assert command == CompleteWorkflow(fixture["expected"]["result"]) + + +def test_an_untyped_history_with_predicate_fingerprints_replays_under_typed_true() -> None: + (command,) = replay(TypedPredicate, _untyped_predicate_history(), [], payload_codec="avro").commands + assert command == CompleteWorkflow(["Grace", "Ada"]) @pytest.mark.parametrize("kind", [staticmethod, classmethod])