Preserve runs when redeployed Python entry arguments cannot bind

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

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",

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] = {

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."""
 

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)

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:

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"}
+}

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)

tests/test_workflow_entry_arguments.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import json
+from copy import deepcopy
+from typing import Any
+
+import httpx
+import pytest
+
+from durable_workflow import Client, workflow
+from durable_workflow.errors import DurableWorkflowError, WorkflowEntryArgumentsMismatchError
+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")
+
+
+@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(
+    definition: type, committed_wait: bool, monkeypatch: pytest.MonkeyPatch,
+) -> 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"}]
+    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"
+    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"}]
+    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"
+
+
+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"