Guard Python wait_condition replay fingerprints

Issue: zorporation/durable-workflow#397

Loop-ID: build-02

Author: durable-workflow-ops

README.md

@@ -144,6 +144,13 @@ server. Query routing and synchronous pre-accept update validator execution are
 still server-side follow-ups; use those paths only with deployments that
 advertise support for the target workflow type.
 
+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.
+
 Workers fingerprint registered workflow class definitions and advertise those
 fingerprints during registration. Re-registering the same `worker_id` with a
 changed class body for an already advertised workflow type raises immediately;

src/durable_workflow/workflow.py

@@ -678,6 +678,30 @@ def to_server_command(
         return cmd
 
 
+def _condition_predicate_fingerprint(predicate: Callable[[], bool]) -> str:
+    h = hashlib.sha256()
+    h.update(b"durable-workflow-python.wait-condition.v1\0")
+    h.update(f"{getattr(predicate, '__module__', '')}\0".encode())
+    h.update(f"{getattr(predicate, '__qualname__', '')}\0".encode())
+
+    code = getattr(predicate, "__code__", None)
+    if code is None:
+        h.update(repr(predicate).encode())
+    else:
+        h.update(repr((
+            code.co_argcount,
+            code.co_posonlyargcount,
+            code.co_kwonlyargcount,
+            code.co_code,
+            code.co_consts,
+            code.co_names,
+            code.co_varnames,
+            code.co_freevars,
+        )).encode())
+
+    return f"sha256:{h.hexdigest()}"
+
+
 Command = (
     ScheduleActivity | StartTimer | CompleteWorkflow | FailWorkflow
     | CompleteUpdate | FailUpdate | ContinueAsNew | RecordSideEffect | StartChildWorkflow
@@ -960,6 +984,7 @@ def wait_condition(
         return WaitCondition(
             predicate=predicate,
             condition_key=key,
+            condition_definition_fingerprint=_condition_predicate_fingerprint(predicate),
             timeout_seconds=timeout_seconds,
         )
 
@@ -1551,10 +1576,10 @@ def _state(commands: list[Command]) -> _ReplayState:
     # external receivers apply before the generator consumes the resolved_result
     # at the stored index, preserving history interleaving with activities.
     pending_receivers: list[tuple[int, str, str, list[Any]]] = []
-    # Ordered list of condition_wait_id strings from ConditionWaitOpened events,
-    # used by ``WaitCondition`` yields to match against their corresponding
-    # opened wait in history (Nth yield ↔ Nth opened).
-    wait_opened_ids: list[str] = []
+    # Ordered ``ConditionWaitOpened`` payloads, used by ``WaitCondition`` yields
+    # to match against their corresponding opened wait in history
+    # (Nth yield ↔ Nth opened).
+    wait_opened: list[dict[str, Any]] = []
     # Map condition_wait_id → resolution: 'satisfied' (from ConditionWaitSatisfied
     # in history, future server-recorded) or 'timed_out' (from a matching
     # condition_timeout TimerFired event).
@@ -1577,7 +1602,7 @@ def _state(commands: list[Command]) -> _ReplayState:
         elif etype == "ConditionWaitOpened":
             wait_id = payload.get("condition_wait_id")
             if isinstance(wait_id, str) and wait_id:
-                wait_opened_ids.append(wait_id)
+                wait_opened.append(dict(payload))
         elif etype == "ConditionWaitSatisfied":
             wait_id = payload.get("condition_wait_id")
             if isinstance(wait_id, str) and wait_id:
@@ -1734,8 +1759,35 @@ def _apply_due_receivers() -> None:
                 continue
             if isinstance(cmd, WaitCondition):
                 resolution: str | None = None
-                if wait_yield_count < len(wait_opened_ids):
-                    resolution = wait_resolutions.get(wait_opened_ids[wait_yield_count])
+                opened: dict[str, Any] | None = None
+                if wait_yield_count < len(wait_opened):
+                    opened = wait_opened[wait_yield_count]
+                    opened_id = opened.get("condition_wait_id")
+                    if isinstance(opened_id, str):
+                        resolution = wait_resolutions.get(opened_id)
+                    opened_key = opened.get("condition_key")
+                    if isinstance(opened_key, str) and opened_key != (cmd.condition_key or ""):
+                        return _state([FailWorkflow(
+                            message=(
+                                "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 _state([FailWorkflow(
+                            message=(
+                                "wait_condition predicate fingerprint changed during replay: "
+                                f"history has {opened_fingerprint!r}, workflow yielded "
+                                f"{cmd.condition_definition_fingerprint!r}"
+                            ),
+                            exception_type="NonDeterministicWaitCondition",
+                        )])
                 if resolution == "timed_out":
                     next_value = False
                     wait_yield_count += 1

tests/test_wait_condition.py

@@ -3,6 +3,7 @@
 from durable_workflow import serializer, workflow
 from durable_workflow.workflow import (
     CompleteWorkflow,
+    FailWorkflow,
     WaitCondition,
     WorkflowContext,
     replay,
@@ -67,6 +68,8 @@ def test_wait_condition_returns_dataclass_with_predicate_and_key(self) -> None:
         assert isinstance(cmd, WaitCondition)
         assert cmd.condition_key == "k"
         assert cmd.timeout_seconds == 5
+        assert cmd.condition_definition_fingerprint is not None
+        assert cmd.condition_definition_fingerprint.startswith("sha256:")
         assert callable(cmd.predicate)
         assert cmd.predicate() is False
 
@@ -95,6 +98,7 @@ def test_emits_open_condition_wait_with_optional_fields(self) -> None:
         cmd = WaitCondition(
             predicate=lambda: True,
             condition_key="order",
+            condition_definition_fingerprint="sha256:condition",
             timeout_seconds=60,
         )
 
@@ -103,6 +107,7 @@ def test_emits_open_condition_wait_with_optional_fields(self) -> None:
         assert server_cmd == {
             "type": "open_condition_wait",
             "condition_key": "order",
+            "condition_definition_fingerprint": "sha256:condition",
             "timeout_seconds": 60,
         }
 
@@ -196,6 +201,70 @@ def test_open_with_no_resolution_and_predicate_false_re_emits_wait_condition(sel
         assert len(outcome.commands) == 1
         assert isinstance(outcome.commands[0], WaitCondition)
 
+    def test_replay_rejects_changed_condition_key(self) -> None:
+        history = [
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-1", "condition_key": "old-key"},
+            },
+        ]
+
+        outcome = replay(WaitUntilApproved, history, [])
+
+        assert len(outcome.commands) == 1
+        assert not isinstance(outcome.commands[0], CompleteWorkflow)
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, FailWorkflow)
+        assert cmd.exception_type == "NonDeterministicWaitCondition"
+        assert "key changed" in cmd.message
+
+    def test_replay_rejects_changed_condition_fingerprint(self) -> None:
+        initial = replay(WaitUntilApproved, [], [])
+        assert isinstance(initial.commands[0], WaitCondition)
+        fingerprint = initial.commands[0].condition_definition_fingerprint
+        assert fingerprint is not None
+        history = [
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {
+                    "condition_wait_id": "wait-1",
+                    "condition_key": "approved",
+                    "condition_definition_fingerprint": "sha256:different",
+                },
+            },
+        ]
+
+        outcome = replay(WaitUntilApproved, history, [])
+
+        assert len(outcome.commands) == 1
+        cmd = outcome.commands[0]
+        assert isinstance(cmd, FailWorkflow)
+        assert cmd.exception_type == "NonDeterministicWaitCondition"
+        assert "predicate fingerprint changed" in cmd.message
+        assert fingerprint in cmd.message
+
+    def test_replay_accepts_matching_condition_fingerprint(self) -> None:
+        initial = replay(WaitUntilApproved, [], [])
+        assert isinstance(initial.commands[0], WaitCondition)
+        fingerprint = initial.commands[0].condition_definition_fingerprint
+        history = [
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {
+                    "condition_wait_id": "wait-1",
+                    "condition_key": "approved",
+                    "condition_definition_fingerprint": fingerprint,
+                },
+            },
+            _signal_received_event("approve", []),
+        ]
+
+        outcome = replay(WaitUntilApproved, history, [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == "approved"
+
     def test_condition_timeout_does_not_pollute_start_timer_resolved_results(self) -> None:
         @workflow.defn(name="wait-then-sleep")
         class WaitThenSleep: