diff --git a/CHANGELOG.md b/CHANGELOG.md index 911923e..94ee15a 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] +### Fixed + +- Worker batch encoding converts dataclass and Enum workflow results, update results and continue-as-new arguments to neutral Avro values, matching individual command encoding. Typed results previously failed during real Worker submission. (#184) + ### Added - Unreleased typed run-activity visibility through `Client.get_run_activities` and `WorkflowHandle.get_activities`, with activity/attempt metadata, exact run/namespace binding, explicit unavailable fields and preserved diagnostic authority. Standalone activity listing is unchanged. Shared delivery remains tracked in durable-workflow/.github#235. diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 438019a..58ae9bd 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -1856,7 +1856,7 @@ def commands_to_server_commands( envelope_jobs.append(( len(server_commands) - 1, "result", - command.result, + to_payload_value(command.result), _payload_warning_context( warning_context, kind="workflow_result", @@ -1870,7 +1870,7 @@ def commands_to_server_commands( envelope_jobs.append(( len(server_commands) - 1, "result", - command.result, + to_payload_value(command.result), _payload_warning_context( warning_context, kind="update_result", @@ -1888,7 +1888,7 @@ def commands_to_server_commands( envelope_jobs.append(( len(server_commands) - 1, "arguments", - command.arguments, + to_payload_values(list(command.arguments)), _payload_warning_context( warning_context, kind="continue_as_new_input", diff --git a/tests/fixtures/replay_regressions/typed-terminal-result-batched-encoding.json b/tests/fixtures/replay_regressions/typed-terminal-result-batched-encoding.json new file mode 100644 index 0000000..311d705 --- /dev/null +++ b/tests/fixtures/replay_regressions/typed-terminal-result-batched-encoding.json @@ -0,0 +1,55 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "typed-terminal-result-batched-encoding", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.typed-terminal-result", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-05T00:00:00+00:00", + "payload": {}, + "sequence": 1 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-11-05T00:00:01+00:00", + "payload": { + "sequence": 1, + "activity_type": "tests.typed.echo" + }, + "sequence": 2 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-05T00:00:02+00:00", + "payload": { + "sequence": 1, + "activity_type": "tests.typed.echo", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4GCG5hbWUKBkFkYQh0aWVyCghnb2xkCHRhZ3MMAgoGdmlwAAA=" + }, + "payload_codec": "avro" + }, + "sequence": 3 + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "name": "Ada", + "tier": "gold", + "tags": [ + "vip" + ] + } + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 143ad7b..1fb07e1 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -58,7 +58,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, TypedTerminalResult from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -426,6 +426,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: OrphanedDuplicateDecision, TypedActivityResultsWorkflow, TypedOnboarding, + TypedTerminalResult, FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate, diff --git a/tests/test_typed_workflow_values.py b/tests/test_typed_workflow_values.py index 5e87ff7..d43b017 100644 --- a/tests/test_typed_workflow_values.py +++ b/tests/test_typed_workflow_values.py @@ -12,6 +12,7 @@ import pytest from durable_workflow import TypedValueError, serializer, workflow +from durable_workflow.external_storage import LocalFilesystemExternalStorage from durable_workflow.workflow import ( CompleteUpdate, CompleteWorkflow, @@ -20,6 +21,7 @@ StartChildWorkflow, WorkflowContext, apply_update, + commands_to_server_commands, query_state, replay, ) @@ -49,6 +51,12 @@ class Approval: ADA_NEUTRAL = {"name": "Ada", "tier": "gold", "tags": ["vip"]} +@workflow.defn(name="tests.replay.typed-terminal-result") +class TypedTerminalResult: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + return (yield ctx.schedule_activity("tests.typed.echo", [], result_type=Customer)) + + @workflow.defn(name="tests.replay.typed-workflow-values", typed=True) class TypedOnboarding: """Typed run, signal, update and query arguments, and a typed child result.""" @@ -168,6 +176,39 @@ def test_child_arguments_and_terminal_values_travel_as_neutral_bytes() -> None: assert CompleteWorkflow(ADA).result is ADA +@pytest.mark.parametrize("external", [False, True]) +@pytest.mark.parametrize(("command", "key", "neutral"), [ + (CompleteWorkflow(ADA), "result", ADA_NEUTRAL), + (CompleteUpdate("typed-update", {"customer": ADA, "tier": Tier.GOLD}), + "result", {"customer": ADA_NEUTRAL, "tier": "gold"}), + (ContinueAsNew(arguments=[ADA, Tier.GOLD]), "arguments", [ADA_NEUTRAL, "gold"]), + (CompleteWorkflow(ADA_NEUTRAL), "result", ADA_NEUTRAL), + (CompleteUpdate("plain-update", None), "result", None), + (ContinueAsNew(arguments=[ADA_NEUTRAL, "gold"]), "arguments", [ADA_NEUTRAL, "gold"]), +]) +def test_worker_batch_preserves_typed_terminal_wire_contract( + tmp_path: Path, external: bool, command: Any, key: str, neutral: Any, +) -> None: + storage = LocalFilesystemExternalStorage(tmp_path) if external else None + options = {"size_warning": None, "external_storage": storage, + "external_storage_threshold_bytes": 1 if external else None} + direct = command.to_server_command("q", **options) + authored = command.arguments if isinstance(command, ContinueAsNew) else command.result + before = CompleteWorkflow({"plain": [True, 0, b"bytes"]}) + after = CompleteUpdate("tail", [None, 1.5]) + batch = commands_to_server_commands([before, command, after], "q", **options) + assert len(batch) == 3 + assert serializer.decode_envelope(batch[0]["result"], external_storage=storage) == before.result + assert serializer.decode_envelope(batch[2]["result"], external_storage=storage) == after.result + assert serializer.decode_envelope(batch[1][key], external_storage=storage) == neutral + assert {k: v for k, v in batch[1].items() if k != key} == {k: v for k, v in direct.items() if k != key} + if external: + assert "external_storage" in batch[1][key] + else: + assert batch[1] == direct + assert (command.arguments if isinstance(command, ContinueAsNew) else command.result) is authored + + def test_a_child_result_type_mismatch_is_raised_at_the_yield() -> None: @workflow.defn(name="tests.typed.child-mismatch") class ChildMismatch: