Keep condition replay drift recoverable and prepare Python 2.5.3

docs/sdk-reference.md

@@ -488,8 +488,15 @@ Use `yield ctx.wait_condition(lambda: self.approved, key="approved",
 timeout=30)` to wait for signal- or update-mutated workflow state without
 polling timers by hand. The SDK sends a stable predicate fingerprint with the
 durable wait command and rejects replay if history records a different wait
-key or predicate fingerprint, so condition changes fail visibly instead of
-silently resolving a different wait.
+key or predicate fingerprint. A mismatch raises `NonDeterministicReplayError`
+with the original wait identity, workflow sequence and both definitions. The
+worker fails the current task without recording a terminal workflow failure.
+Deploy compatible code, then use `await handle.repair()` to release the blocked
+task for the original run. The Server holds replay failures instead of
+continuously retrying incompatible code. Accepted updates also remain available
+for repair when their state replay encounters this error. Query replay
+reports a query failure with the mismatch as its cause. Exceptions raised by
+application predicates remain ordinary workflow failures.
 
 Workers fingerprint registered workflow class definitions and advertise those
 fingerprints during registration. Re-registering the same `worker_id` with a

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.5.2"
+version = "2.5.3"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.5.2"
-registry-version = "2.5.2"
+product-train = "2.5.3"
+registry-version = "2.5.3"
 supported-server-versions = "2.5.16"
 worker-protocol-version = "1.19"
 control-plane-version = "2"

src/durable_workflow/worker.py

@@ -69,6 +69,7 @@
     DurableWorkflowError,
     ExternalPayloadError,
     InvalidArgument,
+    NonDeterministicReplayError,
     NonRetryableError,
     QueryFailed,
     ServerError,
@@ -2502,20 +2503,35 @@ async def _run_workflow_task_core(self, task: dict[str, Any]) -> list[dict[str,
 
         update_id = task.get("workflow_update_id")
         if isinstance(update_id, str) and update_id:
-            update_command = apply_update(
-                cls,
-                history,
-                start_input,
-                update_id,
-                workflow_id=task.get("workflow_id"),
-                run_id=run_id,
-                payload_codec=codec,
-                external_storage=self.external_storage,
-                external_storage_cache=self.external_storage_cache,
-                prepare_local_activities=self._prepared_local_activities_supported,
-                prepare_local_activity_groups=self._prepared_local_activity_groups_supported,
-                local_activity_cancellation_policies=self._local_activity_cancellation_policies,
-            )
+            try:
+                update_command = apply_update(
+                    cls,
+                    history,
+                    start_input,
+                    update_id,
+                    workflow_id=task.get("workflow_id"),
+                    run_id=run_id,
+                    payload_codec=codec,
+                    external_storage=self.external_storage,
+                    external_storage_cache=self.external_storage_cache,
+                    prepare_local_activities=self._prepared_local_activities_supported,
+                    prepare_local_activity_groups=self._prepared_local_activity_groups_supported,
+                    local_activity_cancellation_policies=self._local_activity_cancellation_policies,
+                )
+            except NonDeterministicReplayError as exc:
+                log.exception("workflow update task %s replay failed for run %s", task_id, run_id)
+                try:
+                    await self.client.fail_workflow_task(
+                        task_id=task_id,
+                        lease_owner=self.worker_id,
+                        workflow_task_attempt=attempt,
+                        message=f"workflow replay failed before update {update_id!r}: {exc}",
+                        failure_type=type(exc).__name__,
+                        stack_trace=traceback.format_exc(),
+                    )
+                except Exception as failure_error:
+                    log.warning("failed to report update replay failure: %s", failure_error)
+                return None
             command = update_command.to_server_command(
                 self.task_queue,
                 payload_codec=command_codec,

src/durable_workflow/workflow.py

@@ -3067,6 +3067,8 @@ def apply_update(
             prepare_local_activity_groups=prepare_local_activity_groups,
             local_activity_cancellation_policies=local_activity_cancellation_policies,
         )
+    except NonDeterministicReplayError:
+        raise
     except Exception as exc:
         return _fail_update_from_exception(
             update_id,
@@ -5144,30 +5146,35 @@ def _apply_condition_wait_receivers(condition_wait_id: str | None) -> None:
         for _ in range(boundary):
             _apply_receiver(pending_receivers.pop(0))
 
-    def _condition_wait_mismatch(opened: Mapping[str, Any], cmd: WaitCondition) -> FailWorkflow | None:
+    def _condition_wait_mismatch(
+        opened: Mapping[str, Any], cmd: WaitCondition,
+    ) -> NonDeterministicReplayError | None:
+        step = recorded_wait_steps[wait_yield_count]
+        identity = f"condition wait {opened.get('condition_wait_id')!r}"
         opened_key = opened.get("condition_key")
         if isinstance(opened_key, str) and opened_key != (cmd.condition_key or ""):
-            return FailWorkflow(
-                message=(
-                    "wait_condition key changed during replay: "
+            return NonDeterministicReplayError(
+                step.workflow_sequence, "condition wait", step.event_types,
+                detail=(
+                    f"condition_key_changed: {identity} wait_condition key changed during replay: "
                     f"history has {opened_key!r}, workflow yielded "
                     f"{cmd.condition_key!r}"
                 ),
-                exception_type="NonDeterministicWaitCondition",
             )
 
         opened_fingerprint = opened.get("condition_definition_fingerprint")
         if (
             isinstance(opened_fingerprint, str)
             and cmd.condition_definition_fingerprint != opened_fingerprint
         ):
-            return FailWorkflow(
-                message=(
+            return NonDeterministicReplayError(
+                step.workflow_sequence, "condition wait", step.event_types,
+                detail=(
+                    f"condition_fingerprint_changed: {identity} "
                     "wait_condition predicate fingerprint changed during replay: "
                     f"history has {opened_fingerprint!r}, workflow yielded "
                     f"{cmd.condition_definition_fingerprint!r}"
                 ),
-                exception_type="NonDeterministicWaitCondition",
             )
 
         return None
@@ -6655,7 +6662,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                         next_value = True
                         continue
                     if next_opened_mismatch is not None:
-                        return _state([next_opened_mismatch])
+                        raise next_opened_mismatch
                     step = _next_unconsumed_recorded_step()
                     if step is not None:
                         _assert_step_matches(cmd, step)
@@ -6707,7 +6714,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                         opened = wait_opened[wait_yield_count]
                         mismatch = _condition_wait_mismatch(opened, cmd)
                         if mismatch is not None:
-                            return _state([mismatch])
+                            raise mismatch
                         opened_id = opened.get("condition_wait_id")
                         if isinstance(opened_id, str):
                             resolution = wait_resolutions.get(opened_id)

tests/fixtures/replay_regressions/condition-fingerprint-drift-task-failure.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": "python-condition-fingerprint-drift-task-failure",
+  "protocol_version": "1.19",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.replay.condition-identity", "input": [], "payload_codec": "avro"},
+  "history": [
+    {"event_type": "ConditionWaitOpened", "payload": {"sequence": 7, "condition_wait_id": "wait-original", "condition_key": "approved", "condition_definition_fingerprint": "sha256:different"}}
+  ],
+  "expected": {"command_sequence": []},
+  "expected_replay_error": {"type": "NonDeterministicReplayError", "message_contains": "condition_fingerprint_changed", "workflow_sequence": 7}
+}

tests/fixtures/replay_regressions/condition-key-drift-task-failure.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": "python-condition-key-drift-task-failure",
+  "protocol_version": "1.19",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.replay.condition-identity", "input": [], "payload_codec": "avro"},
+  "history": [
+    {"event_type": "ConditionWaitOpened", "payload": {"sequence": 7, "condition_wait_id": "wait-original", "condition_key": "old-key"}}
+  ],
+  "expected": {"command_sequence": []},
+  "expected_replay_error": {"type": "NonDeterministicReplayError", "message_contains": "condition_key_changed", "workflow_sequence": 7}
+}

tests/integration/test_condition_recovery.py

@@ -0,0 +1,185 @@
+"""A rejected condition deployment must leave the durable run recoverable."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import importlib.metadata
+import inspect
+import json
+import os
+import sys
+import uuid
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from durable_workflow import Client, Worker, workflow
+
+
+@workflow.defn(name="tests.condition-deployment-recovery")
+class ConditionRecoveryWorkflow:
+    def __init__(self) -> None:
+        self.approved = False
+
+    @workflow.signal("wake")
+    def wake(self) -> None:
+        pass
+
+    @workflow.signal("approve")
+    def approve(self) -> None:
+        self.approved = True
+
+    @workflow.update("acknowledge")
+    def acknowledge(self) -> str:
+        return "acknowledged"
+
+    def run(self, ctx: Any) -> Any:
+        # Deliberately unsafe configuration drift keeps the class fingerprint
+        # stable, exercising the recorded condition's own identity protection.
+        mode = os.environ.get("CONDITION_RECOVERY_MODE", "correct")
+        key = "changed" if mode == "key" else "approved"
+        predicate = (lambda: bool(self.approved)) if mode == "fingerprint" else (lambda: self.approved)
+        yield ctx.wait_condition(predicate, key=key)
+        return "approved"
+
+
+async def execute_one() -> None:
+    async with Client(
+        os.environ["DURABLE_WORKFLOW_SERVER_URL"],
+        token=os.environ["DURABLE_WORKFLOW_AUTH_TOKEN"], namespace="default",
+    ) as client:
+        worker = Worker(
+            client, task_queue=os.environ["CONDITION_RECOVERY_QUEUE"],
+            workflows=[ConditionRecoveryWorkflow],
+            worker_id=os.environ["CONDITION_RECOVERY_WORKER"],
+            max_concurrent_workflow_tasks=1,
+        )
+        await worker._register()
+        if os.environ.get("CONDITION_RECOVERY_REGISTER_ONLY") == "1":
+            print(json.dumps({"registered": True}), flush=True)
+            return
+        expect_empty = os.environ.get("CONDITION_RECOVERY_EXPECT_EMPTY") == "1"
+        task = await client.poll_workflow_task(
+            worker_id=worker.worker_id, task_queue=worker.task_queue, timeout=1 if expect_empty else 20,
+        )
+        if expect_empty:
+            assert task is None
+            print(json.dumps({"empty": True}), flush=True)
+            return
+        assert task is not None
+        commands = await worker._run_workflow_task(task)
+        print(json.dumps({
+            "task_id": task["task_id"], "run_id": task["run_id"],
+            "attempt": task["workflow_task_attempt"], "commands": commands,
+            "update_id": task.get("workflow_update_id"),
+            "fingerprint": worker.workflow_definition_fingerprints,
+            "sdk_version": importlib.metadata.version("durable-workflow"),
+            "replay_source_sha256": hashlib.sha256(
+                Path(inspect.getfile(workflow)).read_bytes(),
+            ).hexdigest(),
+        }), flush=True)
+
+
+async def cold_worker(
+    server_url: str, token: str, queue: str, mode: str, *, expect_empty: bool = False,
+    python_executable: str | None = None, register_only: bool = False,
+) -> dict[str, Any]:
+    environment = {
+        **os.environ, "DURABLE_WORKFLOW_SERVER_URL": server_url,
+        "DURABLE_WORKFLOW_AUTH_TOKEN": token, "CONDITION_RECOVERY_QUEUE": queue,
+        "CONDITION_RECOVERY_MODE": mode,
+        "CONDITION_RECOVERY_EXPECT_EMPTY": "1" if expect_empty else "0",
+        "CONDITION_RECOVERY_REGISTER_ONLY": "1" if register_only else "0",
+        "CONDITION_RECOVERY_WORKER": f"condition-worker-{uuid.uuid4().hex}",
+    }
+    process = await asyncio.create_subprocess_exec(
+        python_executable or sys.executable, "-c",
+        "import asyncio; from tests.integration.test_condition_recovery import execute_one; asyncio.run(execute_one())",
+        env=environment, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=45)
+    finally:
+        if process.returncode is None:
+            process.kill()
+            await process.wait()
+    assert process.returncode == 0, stderr.decode()
+    return json.loads(stdout.decode())
+
+
+def history_events(document: dict[str, Any]) -> list[dict[str, Any]]:
+    events = document.get("events", document.get("history_events"))
+    assert isinstance(events, list)
+    return events
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("drift", ["key", "fingerprint"])
+@pytest.mark.parametrize("consumer", ["workflow", "update"])
+async def test_bad_deployment_then_cold_replacement_resumes_original_run(
+    server_url: str, server_token: str, drift: str, consumer: str,
+) -> None:
+    suffix = uuid.uuid4().hex
+    queue = f"condition-recovery-{suffix}"
+    async with Client(server_url, token=server_token, namespace="default") as client:
+        assert await cold_worker(
+            server_url, server_token, queue, "correct", register_only=True,
+            python_executable=os.environ.get("CONDITION_RECOVERY_INITIAL_PYTHON"),
+        ) == {"registered": True}
+        handle = await client.start_workflow(
+            workflow_type="tests.condition-deployment-recovery", task_queue=queue,
+            workflow_id=f"condition-recovery-{suffix}", input=[],
+        )
+        initial = await cold_worker(
+            server_url, server_token, queue, "correct",
+            python_executable=os.environ.get("CONDITION_RECOVERY_INITIAL_PYTHON"),
+        )
+        assert [command["type"] for command in initial["commands"]] == ["open_condition_wait"]
+        original_history = history_events(await handle.get_history())
+        if consumer == "update":
+            await handle.update("acknowledge", wait_for="accepted", request_id=f"ack-{suffix}")
+        else:
+            await handle.signal("wake")
+
+        rejected = await cold_worker(server_url, server_token, queue, drift)
+        assert rejected["commands"] is None
+        assert rejected["run_id"] == initial["run_id"] == handle.run_id
+        assert rejected["fingerprint"] == initial["fingerprint"]
+        assert (await handle.describe()).status not in {"failed", "completed", "terminated"}
+        after_rejection = history_events(await handle.get_history())
+        assert after_rejection[:len(original_history)] == original_history
+        assert all(event["event_type"] != "WorkflowFailed" for event in after_rejection)
+        assert all(event["event_type"] != "UpdateFailed" for event in after_rejection)
+
+        assert await cold_worker(server_url, server_token, queue, "correct", expect_empty=True) == {"empty": True}
+        await handle.repair()
+        resumed = await cold_worker(server_url, server_token, queue, "correct")
+        assert resumed["run_id"] == handle.run_id
+        assert resumed["fingerprint"] == initial["fingerprint"]
+        expected_command = "complete_update" if consumer == "update" else "open_condition_wait"
+        assert [command["type"] for command in resumed["commands"]] == [expected_command]
+        if consumer == "update":
+            assert resumed["update_id"] == rejected["update_id"]
+            assert resumed["update_id"] is not None
+        assert resumed["attempt"] > rejected["attempt"]
+        await handle.signal("approve")
+        finished = await cold_worker(server_url, server_token, queue, "correct")
+        assert [command["type"] for command in finished["commands"]] == ["complete_workflow"]
+        assert await handle.result(timeout=15) == "approved"
+        final_history = history_events(await handle.get_history())
+        assert final_history[:len(original_history)] == original_history
+        kinds = [event["event_type"] for event in final_history]
+        assert "WorkflowFailed" not in kinds
+        assert kinds.count("WorkflowCompleted") == 1
+        if consumer == "update":
+            assert kinds.count("UpdateCompleted") == 1
+            assert "UpdateFailed" not in kinds
+        print(json.dumps({
+            "drift": drift, "consumer": consumer, "initial": initial, "rejected": rejected,
+            "resumed": resumed, "finished": finished,
+            "preserved_prefix_events": len(original_history),
+            "workflow_failed_events": 0, "workflow_completed_events": 1,
+            "explicit_repair": True, "same_run": True,
+        }), flush=True)