Conformance finding: signals/queries smoke hits Python nondeterministic replay

Author: durable-workflow-ops

CHANGELOG.md

@@ -12,6 +12,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   those signals before the task's `ConditionWaitOpened` row. This avoids
   applying rapid signal batches to the previous wait and prevents replay from
   completing with a later wait/timer history step left unconsumed.
+- Condition-wait replay now prefers the recorded event-order wait binding over
+  a signal's stored `workflow_sequence` when several signals arrive before a
+  reopened wait. This handles server histories where later signals are accepted
+  while the previous wait is still open but must replay against the next
+  physical wait.
 - Workflow workers now report unhandled workflow-task execution errors back to
   the server instead of leaving the leased task pending until the lease or CLI
   wait times out. This lets the server observe and retry or fail the task

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "0.4.58"
+version = "0.4.59"
 description = "Python SDK for the Durable Workflow server (language-neutral HTTP protocol)"
 readme = "README.md"
 requires-python = ">=3.10"

src/durable_workflow/workflow.py

@@ -2278,7 +2278,9 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
                     # When multiple signals arrive while the task woken by the
                     # first signal is still leased, the server records those
                     # later SignalReceived rows before the task's next
-                    # ConditionWaitOpened row. Replay them at that next wait.
+                    # ConditionWaitOpened row. Replay them at that next wait
+                    # even if their stored workflow_sequence still points at
+                    # the wait that was open when the signal was accepted.
                     for receiver_index in receivers_since_wait[1:]:
                         if receiver_index is not None:
                             bindings[receiver_index] = wait_id
@@ -2296,8 +2298,8 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
                         prefix_receivers.append(index)
                     continue
 
-                receivers_since_wait.append(None if explicit_sequence else index)
-                if len(receivers_since_wait) == 1 and not explicit_sequence:
+                receivers_since_wait.append(index)
+                if len(receivers_since_wait) == 1:
                     bindings[index] = current_wait_id
                 continue
 
@@ -2404,11 +2406,12 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
             signal_name = payload.get("signal_name")
             if isinstance(signal_name, str) and signal_name:
                 workflow_sequence = _workflow_sequence(payload)
-                condition_wait_id = (
-                    condition_wait_ids_by_sequence.get(workflow_sequence)
-                    if workflow_sequence is not None
-                    else receiver_condition_wait_ids.get(event_index)
-                )
+                if event_index in receiver_condition_wait_ids:
+                    condition_wait_id = receiver_condition_wait_ids[event_index]
+                elif workflow_sequence is not None:
+                    condition_wait_id = condition_wait_ids_by_sequence.get(workflow_sequence)
+                else:
+                    condition_wait_id = None
                 pending_receivers.append(_PendingReceiver(
                     result_index=len(resolved_results),
                     kind="signal",
@@ -2429,11 +2432,12 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
             update_name = payload.get("update_name")
             if isinstance(update_name, str) and update_name:
                 workflow_sequence = _workflow_sequence(payload)
-                condition_wait_id = (
-                    condition_wait_ids_by_sequence.get(workflow_sequence)
-                    if workflow_sequence is not None
-                    else receiver_condition_wait_ids.get(event_index)
-                )
+                if event_index in receiver_condition_wait_ids:
+                    condition_wait_id = receiver_condition_wait_ids[event_index]
+                elif workflow_sequence is not None:
+                    condition_wait_id = condition_wait_ids_by_sequence.get(workflow_sequence)
+                else:
+                    condition_wait_id = None
                 pending_receivers.append(_PendingReceiver(
                     result_index=len(resolved_results),
                     kind="update",

tests/test_wait_condition.py

@@ -513,6 +513,55 @@ def test_signal_received_before_reopened_wait_uses_next_wait_when_sequence_absen
         assert outcome.commands[0].result == expected
         assert query_state(SignalCounterUntilFinished, history, [], "current") == expected
 
+    def test_signal_received_before_reopened_wait_uses_next_wait_when_sequence_is_stale(self) -> None:
+        history = [
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {
+                    "condition_wait_id": "wait-count-3",
+                    "condition_key": "done",
+                    "sequence": 21,
+                },
+            },
+            _signal_received_event("increment", [3], workflow_sequence=21),
+            _signal_received_event("finish", [], workflow_sequence=21),
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {
+                    "condition_wait_id": "wait-finish",
+                    "condition_key": "done",
+                    "sequence": 22,
+                    "timeout_seconds": 30,
+                },
+            },
+            {
+                "event_type": "TimerScheduled",
+                "payload": {
+                    "timer_kind": "condition_timeout",
+                    "condition_wait_id": "wait-finish",
+                    "condition_key": "done",
+                    "sequence": 22,
+                    "delay_seconds": 30,
+                },
+            },
+        ]
+
+        expected = {
+            "count": 3,
+            "done": True,
+            "events": [
+                {"signal": "increment", "amount": 3, "count": 3},
+                {"signal": "finish", "count": 3},
+            ],
+        }
+
+        outcome = replay(SignalCounterUntilFinished, history, [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == expected
+        assert query_state(SignalCounterUntilFinished, history, [], "current") == expected
+
     def test_repeated_wait_after_activity_can_be_satisfied_by_later_signal(self) -> None:
         history = [
             {