Conformance: signals/queries finish signal stays waiting on sdk-python 0.4.42 (#87)

durable-workflow-ops · durable-workflow-ops · commit 9e6bf9d28829 · 2026-05-18T14:05:37.000Z
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
+- Repeated condition-wait openings for the same logical wait now replay through
+  every matching signal before deciding whether the wait is still pending, so
+  long-running signal/query workflows do not get stuck on the first signal.
 - Signal and update receivers recorded while a condition wait is open now
   replay at that specific wait, so later signal-driven waits are not satisfied
   or consumed too early when no activity or timer result separates them.
diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py
@@ -2120,6 +2120,49 @@ def _apply_condition_wait_receivers(condition_wait_id: str | None) -> None:
                 break
             _apply_receiver(pending_receivers.pop(0))
 
+    def _condition_wait_mismatch(opened: Mapping[str, Any], cmd: WaitCondition) -> FailWorkflow | None:
+        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: "
+                    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=(
+                    "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
+
+    def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition) -> bool:
+        if _condition_wait_mismatch(opened, cmd) is not None:
+            return False
+
+        opened_key = opened.get("condition_key")
+        if isinstance(opened_key, str) and opened_key and opened_key == (cmd.condition_key or ""):
+            return True
+
+        opened_fingerprint = opened.get("condition_definition_fingerprint")
+        return (
+            isinstance(opened_fingerprint, str)
+            and bool(opened_fingerprint)
+            and opened_fingerprint == cmd.condition_definition_fingerprint
+        )
+
     result_cursor = 0
     gen = instance.run(ctx, *start_input)
     if not hasattr(gen, "__next__"):
@@ -2208,56 +2251,52 @@ def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState:
                 next_value = _version_marker_result(cmd, cmd.version)
                 continue
             if isinstance(cmd, WaitCondition):
-                resolution: str | None = None
-                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)
-                        _apply_condition_wait_receivers(opened_id)
-                    opened_key = opened.get("condition_key")
-                    if isinstance(opened_key, str) and opened_key != (cmd.condition_key or ""):
+                while True:
+                    resolution: str | None = None
+                    opened: dict[str, Any] | None = None
+                    if wait_yield_count < len(wait_opened):
+                        opened = wait_opened[wait_yield_count]
+                        mismatch = _condition_wait_mismatch(opened, cmd)
+                        if mismatch is not None:
+                            return _state([mismatch])
+                        opened_id = opened.get("condition_wait_id")
+                        if isinstance(opened_id, str):
+                            resolution = wait_resolutions.get(opened_id)
+                            _apply_condition_wait_receivers(opened_id)
+                    if resolution == "timed_out":
+                        next_value = False
+                        wait_yield_count += 1
+                        break
+                    try:
+                        satisfied = bool(cmd.predicate())
+                    except Exception as exc:
                         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",
+                            message=f"wait_condition predicate raised: {exc}",
+                            exception_type=type(exc).__name__,
                         )])
-                    opened_fingerprint = opened.get("condition_definition_fingerprint")
+                    if satisfied or resolution == "satisfied":
+                        next_value = True
+                        wait_yield_count += 1
+                        break
+
+                    next_wait_index = wait_yield_count + 1
+                    # A single logical wait can be re-opened in history after
+                    # non-satisfying signals. Consume repeated physical opens
+                    # with the same key/fingerprint before declaring the
+                    # logical wait still pending.
                     if (
-                        isinstance(opened_fingerprint, str)
-                        and cmd.condition_definition_fingerprint != opened_fingerprint
+                        opened is not None
+                        and next_wait_index < len(wait_opened)
+                        and _same_logical_condition_wait(wait_opened[next_wait_index], cmd)
                     ):
-                        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
-                    continue
-                try:
-                    satisfied = bool(cmd.predicate())
-                except Exception as exc:
-                    return _state([FailWorkflow(
-                        message=f"wait_condition predicate raised: {exc}",
-                        exception_type=type(exc).__name__,
-                    )])
-                if satisfied or resolution == "satisfied":
-                    next_value = True
-                    wait_yield_count += 1
-                    continue
-                ctx.logger._set_replaying(False)
-                pending.append(cmd)
-                wait_yield_count += 1
-                return _state(pending)
+                        wait_yield_count = next_wait_index
+                        continue
+
+                    ctx.logger._set_replaying(False)
+                    pending.append(cmd)
+                    wait_yield_count = next_wait_index
+                    return _state(pending)
+                continue
             if isinstance(cmd, (ScheduleActivity, StartTimer, StartChildWorkflow)):
                 if result_cursor < len(resolved_results):
                     val = resolved_results[result_cursor]
diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py
@@ -135,6 +135,36 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         return self.state()
 
 
+@workflow.defn(name="signal-counter-until-finished")
+class SignalCounterUntilFinished:
+    def __init__(self) -> None:
+        self.count = 0
+        self.done = False
+        self.events: list[dict[str, object]] = []
+
+    @workflow.signal("increment")
+    def increment(self, amount: int) -> None:
+        self.count += amount
+        self.events.append({"signal": "increment", "amount": amount, "count": self.count})
+
+    @workflow.signal("finish")
+    def finish(self) -> None:
+        self.done = True
+        self.events.append({"signal": "finish", "count": self.count})
+
+    @workflow.query("current")
+    def current(self) -> dict[str, object]:
+        return {
+            "count": self.count,
+            "done": self.done,
+            "events": list(self.events),
+        }
+
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        yield ctx.wait_condition(lambda: self.done, key="done")
+        return self.current()
+
+
 class TestCtxWaitCondition:
     def test_wait_condition_returns_dataclass_with_predicate_and_key(self) -> None:
         ctx = WorkflowContext(run_id="x")
@@ -321,6 +351,99 @@ def test_signals_scope_to_the_condition_wait_open_when_they_arrived(self) -> Non
         assert outcome.commands[0].result == expected
         assert query_state(TwoStageSignalWait, history, [], "state") == expected
 
+    def test_repeated_physical_waits_for_one_logical_condition_replay_all_signals(self) -> None:
+        history = [
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-count-3", "condition_key": "done"},
+            },
+            _signal_received_event("increment", [3]),
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-count-8", "condition_key": "done"},
+            },
+            _signal_received_event("increment", [5]),
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-finish", "condition_key": "done"},
+            },
+            _signal_received_event("finish", []),
+        ]
+
+        expected = {
+            "count": 8,
+            "done": True,
+            "events": [
+                {"signal": "increment", "amount": 3, "count": 3},
+                {"signal": "increment", "amount": 5, "count": 8},
+                {"signal": "finish", "count": 8},
+            ],
+        }
+
+        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_physical_waits_keep_query_state_current_before_finish(self) -> None:
+        history = [
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-count-3", "condition_key": "done"},
+            },
+            _signal_received_event("increment", [3]),
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-count-8", "condition_key": "done"},
+            },
+            _signal_received_event("increment", [5]),
+        ]
+
+        outcome = replay(SignalCounterUntilFinished, history, [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], WaitCondition)
+        assert query_state(SignalCounterUntilFinished, history, [], "current") == {
+            "count": 8,
+            "done": False,
+            "events": [
+                {"signal": "increment", "amount": 3, "count": 3},
+                {"signal": "increment", "amount": 5, "count": 8},
+            ],
+        }
+
+    def test_repeated_wait_after_activity_can_be_satisfied_by_later_signal(self) -> None:
+        history = [
+            {
+                "event_type": "ActivityCompleted",
+                "payload": {"result": '"loaded"'},
+            },
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-before-query", "condition_key": "approval"},
+            },
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {"condition_wait_id": "wait-after-query", "condition_key": "approval"},
+            },
+            _signal_received_event("approve", ["alice"]),
+        ]
+
+        outcome = replay(ActivityThenWaitForApproval, history, [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == {
+            "activity_result": "loaded",
+            "approved_by": "alice",
+        }
+        assert query_state(ActivityThenWaitForApproval, history, [], "state") == {
+            "activity_result": "loaded",
+            "approved_by": "alice",
+        }
+
     def test_open_with_no_resolution_and_predicate_false_re_emits_wait_condition(self) -> None:
         history = [
             {
diff --git a/tests/test_worker.py b/tests/test_worker.py
@@ -93,7 +93,7 @@ def current(self) -> int:
         return self.count
 
     def run(self, ctx):  # type: ignore[no-untyped-def]
-        yield ctx.wait_condition(lambda: False)
+        yield ctx.wait_condition(lambda: False, key="done")
 
 
 @workflow.defn(name="async-query-wf")
@@ -758,6 +758,93 @@ async def test_query_task_replays_signal_arguments_from_history_export(
         )
         mock_client.fail_query_task.assert_not_called()
 
+    @pytest.mark.asyncio
+    async def test_query_task_replays_repeated_condition_wait_signal_arguments(
+        self, mock_client: AsyncMock
+    ) -> None:
+        worker = Worker(mock_client, task_queue="q1", workflows=[CounterQueryWorkflow], activities=[])
+        first_signal_arguments = serializer.encode([3], codec="json")
+        second_signal_arguments = serializer.encode([5], codec="json")
+        task = {
+            "query_task_id": "qt-repeated-wait-signals",
+            "query_task_attempt": 1,
+            "workflow_type": "counter-query-wf",
+            "workflow_id": "wf-counter",
+            "run_id": "run-counter",
+            "query_name": "current",
+            "history_events": [
+                {
+                    "event_type": "ConditionWaitOpened",
+                    "payload": {
+                        "condition_wait_id": "wait-count-3",
+                        "condition_key": "done",
+                    },
+                },
+                {
+                    "event_type": "SignalReceived",
+                    "workflow_command_id": "cmd-increment-3",
+                    "payload": {
+                        "signal_id": "sig-increment-3",
+                        "workflow_command_id": "cmd-increment-3",
+                        "signal_name": "increment",
+                    },
+                },
+                {
+                    "event_type": "ConditionWaitOpened",
+                    "payload": {
+                        "condition_wait_id": "wait-count-8",
+                        "condition_key": "done",
+                    },
+                },
+                {
+                    "event_type": "SignalReceived",
+                    "workflow_command_id": "cmd-increment-5",
+                    "payload": {
+                        "signal_id": "sig-increment-5",
+                        "workflow_command_id": "cmd-increment-5",
+                        "signal_name": "increment",
+                    },
+                },
+            ],
+            "history_export": {
+                "payloads": {"codec": "json"},
+                "signals": [
+                    {
+                        "id": "sig-increment-3",
+                        "command_id": "cmd-increment-3",
+                        "name": "increment",
+                        "payload_codec": "json",
+                        "arguments": first_signal_arguments,
+                    },
+                    {
+                        "id": "sig-increment-5",
+                        "command_id": "cmd-increment-5",
+                        "name": "increment",
+                        "payload_codec": "json",
+                        "arguments": second_signal_arguments,
+                    },
+                ],
+            },
+            "workflow_arguments": serializer.envelope([], codec="json"),
+            "query_arguments": serializer.envelope([], codec="json"),
+            "payload_codec": "json",
+        }
+
+        outcome = await worker._run_query_task(task)
+
+        assert outcome == "completed"
+        mock_client.complete_query_task.assert_awaited_once_with(
+            query_task_id="qt-repeated-wait-signals",
+            lease_owner=worker.worker_id,
+            query_task_attempt=1,
+            result=8,
+            codec="json",
+            workflow_id="wf-counter",
+            run_id="run-counter",
+            query_name="current",
+        )
+        mock_client.fail_query_task.assert_not_called()
+
     @pytest.mark.asyncio
     async def test_query_task_awaits_async_query_result(self, mock_client: AsyncMock) -> None:
         worker = Worker(mock_client, task_queue="q1", workflows=[AsyncQueryWorkflow], activities=[])
