Keep post-condition receivers on their replay boundary

CHANGELOG.md

@@ -91,6 +91,10 @@ Earlier SDK versions remain historical releases and are not alternate supported
   `NamespaceDescription.deleted`.
 
 ### Fixed
+- Cold replay now treats an intervening durable step as the authoritative
+  boundary for later signals and updates, even when they retain an older
+  condition sequence. Post-condition activities can therefore complete before
+  those receivers advance the workflow without rebinding them to a closed wait.
 - The README and PyPI project description now require Server worker protocol
   `>=1.19,<2.0`. Package and API-reference release validation derive that
   interval from the shipped SDK protocol and reject semantic guidance drift.

src/durable_workflow/workflow.py

@@ -3925,6 +3925,11 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
                 if current_wait_id is None:
                     if prefix_can_bind_to_first_wait or explicit_sequence is None:
                         prefix_receivers.append(index)
+                    else:
+                        # A durable step separated this receiver from the last
+                        # condition. Record that boundary so a stale sequence
+                        # cannot bind it back to the completed wait.
+                        bindings[index] = None
                     continue
 
                 receivers_since_wait.append(index)

tests/fixtures/replay_regressions/post-condition-receivers-stale-sequence-avro.json

@@ -0,0 +1,113 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "post-condition-receivers-stale-sequence-avro",
+  "protocol_version": "1.19",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.post-condition-receivers",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "WorkflowStarted",
+      "payload": {
+        "workflow_type": "tests.replay.post-condition-receivers",
+        "payload_codec": "avro"
+      }
+    },
+    {
+      "event_type": "ConditionWaitOpened",
+      "payload": {
+        "sequence": 1,
+        "condition_wait_id": "condition:1",
+        "condition_key": "gate"
+      }
+    },
+    {
+      "event_type": "SignalReceived",
+      "payload": {
+        "workflow_sequence": 1,
+        "signal_name": "open-gate",
+        "value": "wwHioz3/VYAiNwwA",
+        "payload_codec": "avro"
+      }
+    },
+    {
+      "event_type": "ConditionWaitSatisfied",
+      "payload": {
+        "sequence": 1,
+        "condition_wait_id": "condition:1",
+        "condition_key": "gate"
+      }
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": 2,
+        "activity_type": "after-condition"
+      }
+    },
+    {
+      "event_type": "ActivityStarted",
+      "payload": {
+        "sequence": 2,
+        "activity_type": "after-condition"
+      }
+    },
+    {
+      "event_type": "UpdateApplied",
+      "payload": {
+        "sequence": 1,
+        "update_id": "update-after-condition",
+        "update_name": "set-note",
+        "arguments": "wwHioz3/VYAiNwwCChBhY2NlcHRlZAA=",
+        "payload_codec": "avro"
+      }
+    },
+    {
+      "event_type": "UpdateCompleted",
+      "payload": {
+        "sequence": 1,
+        "update_id": "update-after-condition",
+        "update_name": "set-note",
+        "result": "wwHioz3/VYAiNwoQYWNjZXB0ZWQ=",
+        "payload_codec": "avro"
+      }
+    },
+    {
+      "event_type": "SignalReceived",
+      "payload": {
+        "workflow_sequence": 1,
+        "signal_name": "finish",
+        "value": "wwHioz3/VYAiNwwA",
+        "payload_codec": "avro"
+      }
+    },
+    {
+      "event_type": "ActivityCompleted",
+      "payload": {
+        "sequence": 2,
+        "activity_type": "after-condition",
+        "result": "wwHioz3/VYAiNwoSY29tcGxldGVk"
+      }
+    },
+    {
+      "event_type": "WorkflowCompleted",
+      "payload": {
+        "result": "wwHioz3/VYAiNw4GHmFjdGl2aXR5X3Jlc3VsdAoSY29tcGxldGVkCG5vdGUKEGFjY2VwdGVkEGZpbmlzaGVkAgEA"
+      }
+    }
+  ],
+  "expected": {
+    "command_type": "CompleteWorkflow",
+    "result": {
+      "activity_result": "completed",
+      "note": "accepted",
+      "finished": true
+    }
+  }
+}

tests/test_replay_regression_corpus.py

@@ -18,7 +18,10 @@
     GoldenTimeoutWaitWorkflow,
     GoldenVersionMarkerWorkflow,
 )
-from tests.test_update_signal_condition_replay import UpdateSignalConditionTimerWorkflow
+from tests.test_update_signal_condition_replay import (
+    PostConditionReceiversWorkflow,
+    UpdateSignalConditionTimerWorkflow,
+)
 
 FIXTURE_SCHEMA = "durable-workflow.replay-regression/v1"
 FIXTURE_DIR = Path(__file__).parent / "fixtures" / "replay_regressions"
@@ -193,6 +196,7 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
     NestedParallelPathWorkflow,
     ParallelMetadataProducerWorkflow,
     ParallelResultBindingWorkflow,
+    PostConditionReceiversWorkflow,
     SelectionAwaitMarkerWorkflow,
     SelectionCancellationQueryWorkflow,
     SelectionMissingNestedOpeningWorkflow,

tests/test_update_signal_condition_replay.py

@@ -92,6 +92,40 @@ def run(self, ctx: WorkflowContext) -> Generator[Any, Any, dict[str, Any]]:
         }
 
 
+@workflow.defn(name="tests.replay.post-condition-receivers")
+class PostConditionReceiversWorkflow:
+    def __init__(self) -> None:
+        self.gate_open = False
+        self.note: str | None = None
+        self.finished = False
+
+    @workflow.signal("open-gate")
+    def open_gate(self) -> None:
+        self.gate_open = True
+
+    @workflow.update("set-note")
+    def set_note(self, note: str) -> str:
+        self.note = note
+        return note
+
+    @workflow.signal("finish")
+    def finish(self) -> None:
+        self.finished = True
+
+    def run(self, ctx: WorkflowContext) -> Generator[Any, Any, dict[str, Any]]:
+        yield ctx.wait_condition(lambda: self.gate_open, key="gate")
+        activity_result = yield ctx.schedule_activity("after-condition", [])
+        yield ctx.wait_condition(
+            lambda: self.note is not None and self.finished,
+            key="post-activity-receivers",
+        )
+        return {
+            "activity_result": activity_result,
+            "note": self.note,
+            "finished": self.finished,
+        }
+
+
 def _event(event_type: str, payload: dict[str, Any]) -> dict[str, Any]:
     return {"event_type": event_type, "payload": payload}
 
@@ -273,7 +307,102 @@ def _legacy_history() -> list[dict[str, Any]]:
     return history
 
 
+def _post_condition_receiver_history() -> list[dict[str, Any]]:
+    return [
+        _event("WorkflowStarted", {"workflow_type": "tests.replay.post-condition-receivers"}),
+        _event(
+            "ConditionWaitOpened",
+            {
+                "sequence": 1,
+                "condition_wait_id": "condition:1",
+                "condition_key": "gate",
+            },
+        ),
+        _event(
+            "SignalReceived",
+            {
+                "workflow_sequence": 1,
+                "signal_name": "open-gate",
+                "value": _payload([]),
+            },
+        ),
+        _event(
+            "ConditionWaitSatisfied",
+            {
+                "sequence": 1,
+                "condition_wait_id": "condition:1",
+                "condition_key": "gate",
+            },
+        ),
+        _event(
+            "ActivityScheduled",
+            {"sequence": 2, "activity_type": "after-condition"},
+        ),
+        _event(
+            "ActivityStarted",
+            {"sequence": 2, "activity_type": "after-condition"},
+        ),
+        _event(
+            "UpdateApplied",
+            {
+                "sequence": 1,
+                "update_id": "update-after-condition",
+                "update_name": "set-note",
+                "arguments": _payload(["accepted"]),
+            },
+        ),
+        _event(
+            "UpdateCompleted",
+            {
+                "sequence": 1,
+                "update_id": "update-after-condition",
+                "update_name": "set-note",
+                "result": _payload("accepted"),
+            },
+        ),
+        _event(
+            "SignalReceived",
+            {
+                "workflow_sequence": 1,
+                "signal_name": "finish",
+                "value": _payload([]),
+            },
+        ),
+        _event(
+            "ActivityCompleted",
+            {
+                "sequence": 2,
+                "activity_type": "after-condition",
+                "result": _payload("completed"),
+            },
+        ),
+        _event(
+            "WorkflowCompleted",
+            {
+                "result": _payload(
+                    {
+                        "activity_result": "completed",
+                        "note": "accepted",
+                        "finished": True,
+                    }
+                )
+            },
+        ),
+    ]
+
+
 class TestUpdateSignalConditionReplay:
+    def test_post_condition_step_boundary_overrides_stale_receiver_sequences(self) -> None:
+        outcome = replay(PostConditionReceiversWorkflow, _post_condition_receiver_history(), [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == {
+            "activity_result": "completed",
+            "note": "accepted",
+            "finished": True,
+        }
+
     def test_current_history_uses_explicit_timeout_identity_without_legacy_classification(self) -> None:
         with patch(
             "durable_workflow.workflow._legacy_condition_timeout_timer_sequence_aliases",