Replay start-boundary receivers after workflow initialization

Replay start-boundary receivers after workflow init

Author: durable-workflow-ops

src/durable_workflow/workflow.py

@@ -1681,8 +1681,6 @@ def _apply_due_receivers() -> None:
             handler(*args)
 
     result_cursor = 0
-    _apply_due_receivers()
-
     gen = instance.run(ctx, *start_input)
     if not hasattr(gen, "__next__"):
         if isinstance(gen, ContinueAsNew):
@@ -1698,13 +1696,19 @@ def _apply_due_receivers() -> None:
     wait_yield_count = 0
     try:
         while True:
-            _apply_due_receivers()
+            # Cursor-0 receivers are start-boundary events. Enter run() once
+            # before applying them so workflow initialization observes the
+            # same WorkflowStarted-before-Signal/Update ordering the server
+            # records in history.
+            if not first:
+                _apply_due_receivers()
             if advanced_cmd is not None:
                 cmd = advanced_cmd
                 advanced_cmd = None
             else:
                 cmd = gen.send(None) if first else gen.send(next_value)
                 first = False
+            _apply_due_receivers()
             if isinstance(cmd, list):
                 needed = len(cmd)
                 if result_cursor + needed <= len(resolved_results):

tests/test_signals.py

@@ -71,6 +71,20 @@ def _activity_completed_event(result: Any) -> dict[str, Any]:
     }
 
 
+def _json_activity_completed_event(result: Any) -> dict[str, Any]:
+    return {
+        "event_type": "ActivityCompleted",
+        "payload": {"result": serializer.encode(result, codec="json"), "payload_codec": "json"},
+    }
+
+
+def _workflow_started_event() -> dict[str, Any]:
+    return {
+        "event_type": "WorkflowStarted",
+        "payload": {"timestamp": "2026-04-21T00:00:00Z"},
+    }
+
+
 def _signal_received_event(name: str, args: list[Any]) -> dict[str, Any]:
     return {
         "event_type": "SignalReceived",
@@ -82,7 +96,45 @@ def _signal_received_event(name: str, args: list[Any]) -> dict[str, Any]:
     }
 
 
+def _json_signal_received_event(name: str, args: list[Any]) -> dict[str, Any]:
+    return {
+        "event_type": "SignalReceived",
+        "payload": {
+            "signal_name": name,
+            "value": serializer.encode(args, codec="json"),
+            "payload_codec": "json",
+        },
+    }
+
+
 class TestSignalDispatchDuringReplay:
+    def test_start_boundary_signal_applies_after_workflow_initialization(self) -> None:
+        @workflow.defn(name="start-boundary-signal-workflow")
+        class StartBoundarySignalWorkflow:
+            def __init__(self) -> None:
+                self.messages: list[str] = []
+
+            @workflow.signal("message")
+            def on_message(self, value: str) -> None:
+                self.messages.append(value)
+
+            def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+                self.messages = ["started"]
+                yield ctx.schedule_activity("wait", [])
+                return list(self.messages)
+
+        events = [
+            _workflow_started_event(),
+            _json_signal_received_event("message", ["after-start"]),
+            _json_activity_completed_event(None),
+        ]
+
+        outcome = replay(StartBoundarySignalWorkflow, events, [], payload_codec="json")
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == ["started", "after-start"]
+
     def test_signal_mutates_workflow_state_before_completion(self) -> None:
         events = [
             _signal_received_event("approve", ["alice"]),

tests/test_updates.py

@@ -123,6 +123,13 @@ def _signal_received_event(name: str, args: list[object]) -> dict[str, object]:
     }
 
 
+def _workflow_started_event() -> dict[str, object]:
+    return {
+        "event_type": "WorkflowStarted",
+        "payload": {"timestamp": "2026-04-21T00:00:00Z"},
+    }
+
+
 def _update_accepted_event(
     update_id: str,
     name: str,
@@ -156,6 +163,39 @@ def _update_applied_event(
 
 
 class TestUpdateApplicationReplay:
+    def test_start_boundary_update_applies_after_workflow_initialization(self) -> None:
+        @workflow.defn(name="start-boundary-update-receiver")
+        class StartBoundaryUpdateReceiver:
+            def __init__(self) -> None:
+                self.messages: list[str] = []
+
+            @workflow.update("append")
+            def append(self, value: str) -> list[str]:
+                self.messages.append(value)
+                return list(self.messages)
+
+            @workflow.query("messages")
+            def read_messages(self) -> list[str]:
+                return list(self.messages)
+
+            def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+                self.messages = ["started"]
+                yield ctx.schedule_activity("wait", [])
+                return list(self.messages)
+
+        history = [
+            _workflow_started_event(),
+            _update_applied_event("upd-start", "append", ["after-start"]),
+        ]
+
+        assert query_state(
+            StartBoundaryUpdateReceiver,
+            history,
+            [],
+            "messages",
+            payload_codec="json",
+        ) == ["started", "after-start"]
+
     def test_replay_applies_committed_update_events_to_workflow_state(self) -> None:
         history = [
             _update_applied_event("upd-1", "increment", [3]),