From 4e69152a4d163ddcb733b826a69ade434e031b8e Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 11:41:25 +0000 Subject: [PATCH 1/3] test: reproduce unsafe workflow entry binding failures --- tests/test_workflow_entry_arguments.py | 106 +++++++++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 tests/test_workflow_entry_arguments.py diff --git a/tests/test_workflow_entry_arguments.py b/tests/test_workflow_entry_arguments.py new file mode 100644 index 0000000..60e2fe0 --- /dev/null +++ b/tests/test_workflow_entry_arguments.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import Any + +import pytest + +from durable_workflow import workflow +from durable_workflow.errors import DurableWorkflowError +from durable_workflow.workflow import CompleteWorkflow, FailWorkflow, WorkflowContext, replay + + +@workflow.defn(name="entry.required") +class RequiredArgumentWorkflow: + calls = 0 + + def run(self, ctx: WorkflowContext, amount: int, required: int) -> int: + type(self).calls += 1 + return amount + required + + +@workflow.defn(name="entry.keyword-only") +class KeywordOnlyWorkflow: + def run(self, ctx: WorkflowContext, amount: int, *, required: int) -> int: + return amount + required + + +@workflow.defn(name="entry.excess") +class ExcessArgumentWorkflow: + def run(self, ctx: WorkflowContext) -> int: + return 7 + + +@workflow.defn(name="entry.optional") +class OptionalArgumentWorkflow: + def run(self, ctx: WorkflowContext, amount: int, extra: int = 0) -> int: + return amount + extra + + +@workflow.defn(name="entry.variadic") +class VariadicWorkflow: + def run(self, ctx: WorkflowContext, *amounts: int) -> int: + return sum(amounts) + + +@workflow.defn(name="entry.body-error") +class BodyTypeErrorWorkflow: + def run(self, ctx: WorkflowContext, amount: int) -> int: + raise TypeError("application body failed") + + +@workflow.defn(name="entry.generator-body-error") +class GeneratorBodyTypeErrorWorkflow: + def run(self, ctx: WorkflowContext, amount: int): # type: ignore[no-untyped-def] + if False: + yield ctx.start_timer(1) + raise TypeError("generator application body failed") + + +@pytest.mark.parametrize("definition", [RequiredArgumentWorkflow, KeywordOnlyWorkflow, ExcessArgumentWorkflow]) +@pytest.mark.parametrize("committed_wait", [False, True]) +def test_entry_binding_refuses_without_running_body_or_rewriting_input( + definition: type, committed_wait: bool, +) -> None: + history = [ + {"event_type": "ConditionWaitOpened", "payload": { + "sequence": 1, "condition_wait_id": "original-wait", "condition_key": "ready", + }}, + ] if committed_wait else [] + original = deepcopy(history) + arguments: list[Any] = [{"private": "input-must-not-appear-in-diagnostics"}] + RequiredArgumentWorkflow.calls = 0 + with pytest.raises(DurableWorkflowError, match="workflow_entry_arguments_mismatch") as captured: + replay(definition, history, arguments) + assert type(captured.value).__name__ == "WorkflowEntryArgumentsMismatchError" + assert getattr(captured.value, "reason") == "workflow_entry_arguments_mismatch" + assert "input-must-not-appear-in-diagnostics" not in str(captured.value) + assert RequiredArgumentWorkflow.calls == 0 + assert arguments == [{"private": "input-must-not-appear-in-diagnostics"}] + assert history == original + + +@pytest.mark.parametrize("definition", [BodyTypeErrorWorkflow, GeneratorBodyTypeErrorWorkflow]) +def test_genuine_body_type_errors_remain_workflow_failures(definition: type) -> None: + outcome = replay(definition, [], [7]) + assert len(outcome.commands) == 1 + failure = outcome.commands[0] + assert isinstance(failure, FailWorkflow) + assert failure.exception_type == "TypeError" + assert "application body failed" in failure.message + + +@pytest.mark.parametrize( + ("definition", "arguments", "expected"), + [(OptionalArgumentWorkflow, [7], 7), (VariadicWorkflow, [7, 2], 9)], +) +def test_compatible_entry_arguments_complete(definition: type, arguments: list[int], expected: int) -> None: + command = replay(definition, [], arguments).commands[0] + assert isinstance(command, CompleteWorkflow) + assert command.result == expected + + +def test_python_annotations_do_not_become_runtime_argument_checks() -> None: + command = replay(OptionalArgumentWorkflow, [], ["not-an-int", "-suffix"]).commands[0] + assert isinstance(command, CompleteWorkflow) + assert command.result == "not-an-int-suffix" From 6dcd59280af4fff4b7764f6f1b0f4f22a3f8693b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 11:52:09 +0000 Subject: [PATCH 2/3] fix: refuse incompatible entry arguments before workflow execution --- docs/sdk-reference.md | 16 ++++++++ src/durable_workflow/__init__.py | 2 + src/durable_workflow/client.py | 7 +++- src/durable_workflow/errors.py | 15 +++++++ src/durable_workflow/worker.py | 5 +++ src/durable_workflow/workflow.py | 16 +++++++- ...w-entry-arguments-required-after-wait.json | 13 ++++++ tests/test_replay_regression_corpus.py | 19 ++++++++- tests/test_workflow_entry_arguments.py | 41 +++++++++++++++++-- 9 files changed, 128 insertions(+), 6 deletions(-) create mode 100644 tests/fixtures/replay_regressions/workflow-entry-arguments-required-after-wait.json diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index a9c3553..d53245b 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -441,6 +441,22 @@ entry points are rejected with `InvalidWorkflowDefinition` during definition or worker registration. Async activity handlers remain supported. Direct replay also rejects async entry points and awaitable results. +Recorded start input must bind to the deployed `run` signature before any +workflow body or version branch executes. Missing required arguments, +unexpected positional arguments and required keyword-only parameters raise +`WorkflowEntryArgumentsMismatchError`. The Worker reports the existing task +failure boundary with reason `workflow_entry_arguments_mismatch`; it does not +submit a terminal workflow-failure command. Direct replay raises the same +typed error without rewriting input or history. Diagnostics contain the +workflow type and binding reason, not the recorded argument values. + +Restore a compatible worker, then use the Server's repair action for the +original run. Repair does not extend its original deadlines. An added optional +parameter or a compatible variadic signature remains supported. Python type +annotations are not enforced as runtime input validation. A `TypeError`, or +even a binding-error exception, raised inside a successfully bound body remains +an application workflow failure. + Workers reject duplicate workflow or activity wire names with `DuplicateRegistrationError` before contacting Server or polling work. Each Worker and offline Replayer has its own registration scope. Distinct names diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py index 7512447..dc9bf20 100644 --- a/src/durable_workflow/__init__.py +++ b/src/durable_workflow/__init__.py @@ -103,6 +103,7 @@ UpdateValidationFailed, WorkflowAlreadyStarted, WorkflowCancelled, + WorkflowEntryArgumentsMismatchError, WorkflowFailed, WorkflowNotFound, WorkflowPayloadDecodeError, @@ -319,6 +320,7 @@ "WorkflowHandle", "WorkflowList", "WorkflowPayloadDecodeError", + "WorkflowEntryArgumentsMismatchError", "WorkflowRun", "WorkflowRunList", "WorkflowStreamAppendItem", diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index db47a0d..442737a 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -5312,15 +5312,20 @@ async def fail_workflow_task( message: str, failure_type: str | None = None, stack_trace: str | None = None, + failure_reason: str | None = None, ) -> Any: - """Report a workflow task failure so the server can schedule a retry. + """Report a workflow task failure without closing the run. Worker-plane endpoint. Task failures (e.g. non-determinism) are distinct from workflow failures (``FailWorkflow`` commands). + The runtime may block incompatible replay until compatible code and + a supported repair are available instead of automatically retrying it. """ failure: dict[str, Any] = {"message": message} if failure_type is not None: failure["type"] = failure_type + if failure_reason is not None: + failure["reason"] = failure_reason if stack_trace is not None: failure["stack_trace"] = stack_trace body: dict[str, Any] = { diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index 85ce050..724d29e 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -465,6 +465,21 @@ def __init__( self.exception_type = exception_type +class WorkflowEntryArgumentsMismatchError(DurableWorkflowError): + """Recorded input cannot bind to the deployed entry method before it runs.""" + + reason = "workflow_entry_arguments_mismatch" + + def __init__(self, workflow_type: str, binding_reason: str) -> None: + self.workflow_type = workflow_type + self.entry_method = "run" + self.binding_reason = binding_reason + super().__init__( + f"{self.reason}: {workflow_type}.run cannot bind the recorded arguments " + f"({binding_reason}). Restore compatible code and repair the workflow task." + ) + + class NonDeterministicReplayError(DurableWorkflowError): """Current workflow code yielded a command shape that does not match history.""" diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index d99e963..7880f21 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -79,6 +79,7 @@ ServerError, UpdateRejected, UpdateValidationFailed, + WorkflowEntryArgumentsMismatchError, ) from .external_storage import ExternalPayloadCache, ExternalStorageDriver, store_external_payload from .interceptors import ( @@ -2619,6 +2620,9 @@ def execute_local(command: RecordLocalActivity) -> Any: return None except Exception as e: log.exception("replay failed") + failure_details = ( + {"failure_reason": e.reason} if isinstance(e, WorkflowEntryArgumentsMismatchError) else {} + ) try: await self.client.fail_workflow_task( task_id=task_id, @@ -2627,6 +2631,7 @@ def execute_local(command: RecordLocalActivity) -> Any: message=f"replay failed: {e}", failure_type=type(e).__name__, stack_trace=traceback.format_exc(), + **failure_details, ) except Exception as fe: log.warning("failed to report replay failure: %s", fe) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 8710ba3..f98ce98 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -66,6 +66,7 @@ QueryFailed, SagaCompensationFailed, WorkflowCancelled, + WorkflowEntryArgumentsMismatchError, WorkflowPayloadDecodeError, ) from .external_storage import ExternalPayloadCache, ExternalStorageDriver @@ -5386,8 +5387,21 @@ def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition) result_cursor = 0 pending_step_cursor = 0 wait_yield_count = 0 + entry = instance.run + workflow_type = str(getattr(workflow_cls, "__workflow_name__", workflow_cls.__name__)) try: - gen = instance.run(ctx, *start_input) + entry_signature = inspect.signature(entry) + except (TypeError, ValueError): + raise WorkflowEntryArgumentsMismatchError(workflow_type, "signature_unavailable") from None + try: + entry_signature.bind(ctx, *start_input) + except TypeError: + raise WorkflowEntryArgumentsMismatchError(workflow_type, "argument_binding_failed") from None + + # Binding happens outside the application-error path. A TypeError raised + # by a successfully bound body is still an actual workflow failure. + try: + gen = entry(ctx, *start_input) except NonDeterministicReplayError: raise except LocalActivityExecutionAborted: diff --git a/tests/fixtures/replay_regressions/workflow-entry-arguments-required-after-wait.json b/tests/fixtures/replay_regressions/workflow-entry-arguments-required-after-wait.json new file mode 100644 index 0000000..9431cbf --- /dev/null +++ b/tests/fixtures/replay_regressions/workflow-entry-arguments-required-after-wait.json @@ -0,0 +1,13 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "workflow-entry-arguments-required-after-wait", + "protocol_version": "1.20", + "bindings": ["python"], + "workflow": {"type": "tests.replay.entry-arguments-required", "input": [7], "payload_codec": "avro"}, + "history": [ + {"sequence": 3, "event_type": "ConditionWaitOpened", "payload": {"sequence": 1, "condition_wait_id": "original-condition-wait", "condition_key": "ready"}} + ], + "expected": {"command_sequence": []}, + "expected_replay_error": {"type": "WorkflowEntryArgumentsMismatchError", "message_contains": "workflow_entry_arguments_mismatch"} +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 3b8e5aa..c51d360 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -12,7 +12,12 @@ from durable_workflow import Replayer, Worker, serializer, workflow from durable_workflow.client import PROTOCOL_VERSION, Client, WorkflowExecution, WorkflowStreamAppendItem -from durable_workflow.errors import NonDeterministicReplayError, WorkflowCancelled, WorkflowPayloadDecodeError +from durable_workflow.errors import ( + DurableWorkflowError, + NonDeterministicReplayError, + WorkflowCancelled, + WorkflowPayloadDecodeError, +) from durable_workflow.replay_conformance import compose_report from durable_workflow.workflow import ( LocalActivityExecutionAborted, @@ -282,6 +287,12 @@ async def run(self, ctx: WorkflowContext) -> str: return "body was never executed" +@workflow.defn(name="tests.replay.entry-arguments-required") +class RequiredEntryArgumentWorkflow: + def run(self, ctx: WorkflowContext, amount: int, required: int) -> int: + raise AssertionError("incompatible entry body must never run") + + @workflow.defn(name="tests.replay.nexus-target") class NexusTargetWorkflow: def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] @@ -356,6 +367,7 @@ def run(self, ctx: WorkflowContext, cleanup_seconds: int = 0): # type: ignore[n ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} WORKFLOW_TYPES["tests.replay.condition-identity"] = ConditionIdentityWorkflow +WORKFLOW_TYPES["tests.replay.entry-arguments-required"] = RequiredEntryArgumentWorkflow def _fixture_paths() -> list[Path]: @@ -558,6 +570,11 @@ def test_checked_in_replay_regression_corpus_uses_official_replayer( message = expected_replay_error.get("message_contains") assert isinstance(message, str) and message assert expected.get("command_sequence") == [] + if expected_replay_error.get("type") == "WorkflowEntryArgumentsMismatchError": + with pytest.raises(DurableWorkflowError, match=message) as binding_error: + _execute_fixture(fixture) + assert type(binding_error.value).__name__ == "WorkflowEntryArgumentsMismatchError" + return if expected_replay_error.get("type") == "LocalActivityExecutionAborted": with pytest.raises(LocalActivityExecutionAborted, match=message): _execute_fixture(fixture) diff --git a/tests/test_workflow_entry_arguments.py b/tests/test_workflow_entry_arguments.py index 60e2fe0..3bd61fa 100644 --- a/tests/test_workflow_entry_arguments.py +++ b/tests/test_workflow_entry_arguments.py @@ -1,12 +1,14 @@ from __future__ import annotations +import json from copy import deepcopy from typing import Any +import httpx import pytest -from durable_workflow import workflow -from durable_workflow.errors import DurableWorkflowError +from durable_workflow import Client, workflow +from durable_workflow.errors import DurableWorkflowError, WorkflowEntryArgumentsMismatchError from durable_workflow.workflow import CompleteWorkflow, FailWorkflow, WorkflowContext, replay @@ -57,6 +59,12 @@ def run(self, ctx: WorkflowContext, amount: int): # type: ignore[no-untyped-def raise TypeError("generator application body failed") +@workflow.defn(name="entry.body-mismatch-error") +class BodyMismatchErrorWorkflow: + def run(self, ctx: WorkflowContext, amount: int) -> int: + raise WorkflowEntryArgumentsMismatchError("application-workflow", "application-control") + + @pytest.mark.parametrize("definition", [RequiredArgumentWorkflow, KeywordOnlyWorkflow, ExcessArgumentWorkflow]) @pytest.mark.parametrize("committed_wait", [False, True]) def test_entry_binding_refuses_without_running_body_or_rewriting_input( @@ -73,7 +81,8 @@ def test_entry_binding_refuses_without_running_body_or_rewriting_input( with pytest.raises(DurableWorkflowError, match="workflow_entry_arguments_mismatch") as captured: replay(definition, history, arguments) assert type(captured.value).__name__ == "WorkflowEntryArgumentsMismatchError" - assert getattr(captured.value, "reason") == "workflow_entry_arguments_mismatch" + assert isinstance(captured.value, WorkflowEntryArgumentsMismatchError) + assert captured.value.reason == "workflow_entry_arguments_mismatch" assert "input-must-not-appear-in-diagnostics" not in str(captured.value) assert RequiredArgumentWorkflow.calls == 0 assert arguments == [{"private": "input-must-not-appear-in-diagnostics"}] @@ -104,3 +113,29 @@ def test_python_annotations_do_not_become_runtime_argument_checks() -> None: command = replay(OptionalArgumentWorkflow, [], ["not-an-int", "-suffix"]).commands[0] assert isinstance(command, CompleteWorkflow) assert command.result == "not-an-int-suffix" + + +def test_binding_exception_thrown_by_a_valid_body_remains_an_application_failure() -> None: + command = replay(BodyMismatchErrorWorkflow, [], [7]).commands[0] + assert isinstance(command, FailWorkflow) + assert command.exception_type == "WorkflowEntryArgumentsMismatchError" + + +async def test_binding_refusal_sends_existing_structured_failure_reason() -> None: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json={"recorded": True}) + + async with Client("http://runtime.invalid", token="fixture-token") as client: + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(respond)) + await client.fail_workflow_task( + task_id="original-task", lease_owner="original-worker", workflow_task_attempt=1, + message="replay failed: entry binding", failure_type="WorkflowEntryArgumentsMismatchError", + failure_reason="workflow_entry_arguments_mismatch", + ) + failure = json.loads(requests[0].content)["failure"] + assert failure["reason"] == "workflow_entry_arguments_mismatch" + assert failure["type"] == "WorkflowEntryArgumentsMismatchError" From 404064ec55c89c3231703b180bfb6488cad5c4c2 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 20 Oct 2026 12:30:27 +0000 Subject: [PATCH 3/3] test: restore entry counter with the pytest fixture --- tests/test_workflow_entry_arguments.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_workflow_entry_arguments.py b/tests/test_workflow_entry_arguments.py index 3bd61fa..fc701fc 100644 --- a/tests/test_workflow_entry_arguments.py +++ b/tests/test_workflow_entry_arguments.py @@ -68,7 +68,7 @@ def run(self, ctx: WorkflowContext, amount: int) -> int: @pytest.mark.parametrize("definition", [RequiredArgumentWorkflow, KeywordOnlyWorkflow, ExcessArgumentWorkflow]) @pytest.mark.parametrize("committed_wait", [False, True]) def test_entry_binding_refuses_without_running_body_or_rewriting_input( - definition: type, committed_wait: bool, + definition: type, committed_wait: bool, monkeypatch: pytest.MonkeyPatch, ) -> None: history = [ {"event_type": "ConditionWaitOpened", "payload": { @@ -77,7 +77,7 @@ def test_entry_binding_refuses_without_running_body_or_rewriting_input( ] if committed_wait else [] original = deepcopy(history) arguments: list[Any] = [{"private": "input-must-not-appear-in-diagnostics"}] - RequiredArgumentWorkflow.calls = 0 + monkeypatch.setattr(RequiredArgumentWorkflow, "calls", 0) with pytest.raises(DurableWorkflowError, match="workflow_entry_arguments_mismatch") as captured: replay(definition, history, arguments) assert type(captured.value).__name__ == "WorkflowEntryArgumentsMismatchError"