Preserve pending operations across signal wakeups

CHANGELOG.md

@@ -6,6 +6,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.4.1] - 2026-10-07
+
+### Fixed
+- Keep already scheduled scalar timers, remote activities and child workflows
+  pending during signal wakeups and cold worker replay. Replay no longer emits
+  another scheduling command for the same recorded operation, including
+  shielded cancellation cleanup timers.
+- Reject changed timer delays when history records the original delay, while
+  preserving compatibility with older histories that omit it.
+
 ## [2.3.9] - 2026-10-04
 
 ### Fixed

docs/sdk-reference.md

@@ -506,6 +506,21 @@ constraining throughput.
 
 ## Replay captured histories
 
+Signals can update workflow state while a timer, remote activity or child is
+pending. Replaying that state waits for the recorded operation. It keeps the
+original timer deadline and does not schedule another activity or child.
+
+Python SDK 2.4.1 fixes duplicate scheduling during these wakeups. Upgrade every
+Python worker on an affected queue. For a run created by an older worker,
+inspect its history for extra `TimerScheduled`, `ActivityScheduled` or
+`ChildWorkflowScheduled` events before the original operation completed.
+The fix prevents new duplicates and does not rewrite existing history. An
+extra completed timer can still produce `NonDeterministicReplayError` when
+replay reaches the next authored operation. Capture that history with the
+workflow code and reconcile any activity or child side effects before
+terminating and replacing an affected run. Do not delete history rows or skip
+recorded operations to force replay to succeed.
+
 Use `Replayer` to debug a captured history without connecting to a live server:
 
 ```python

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.4.0"
+version = "2.4.1"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main"
 durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main"
 
 [tool.durable-workflow]
-product-train = "2.4.0"
-registry-version = "2.4.0"
+product-train = "2.4.1"
+registry-version = "2.4.1"
 supported-server-versions = "2.5.0"
 worker-protocol-version = "1.19"
 control-plane-version = "2"

src/durable_workflow/workflow.py

@@ -3624,6 +3624,8 @@ def _activity_type_from_payload(payload: Mapping[str, Any]) -> str | None:
 
 def _recorded_step_details(payload: Mapping[str, Any]) -> dict[str, Any]:
     details: dict[str, Any] = {}
+    if "delay_seconds" in payload:
+        details["delay_seconds"] = payload["delay_seconds"]
     if "cancellation_cleanup" in payload:
         details["cancellation_cleanup"] = payload["cancellation_cleanup"]
     local_preparation = payload.get("local_preparation")
@@ -3846,6 +3848,12 @@ def _recorded_detail_mismatch(command: Any, step: _RecordedStep) -> str | None:
                     f"child_workflow_policy_changed: recorded {field} {recorded_policy!r}, "
                     f"but current workflow requested {actual_policy!r}."
                 )
+    elif isinstance(command, StartTimer):
+        if "delay_seconds" in step.details and step.details["delay_seconds"] != command.delay_seconds:
+            return (
+                f"Recorded delay_seconds {step.details['delay_seconds']!r}, but current workflow "
+                f"requested {command.delay_seconds!r}."
+            )
     elif isinstance(command, RecordVersionMarker):
         recorded = step.details.get("change_id")
         if isinstance(recorded, str) and recorded != command.change_id:
@@ -6667,6 +6675,11 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     continue
                 ctx.logger._set_replaying(False)
                 _assert_pending_step_matches(cmd)
+                if not isinstance(cmd, RecordLocalActivity) and _next_unconsumed_recorded_step() is not None:
+                    # A signal may wake a workflow while its existing operation is
+                    # still pending. Reissuing it allocates a new durable sequence
+                    # and can move a timer's deadline or duplicate external work.
+                    return _state(pending)
                 if isinstance(cmd, RecordLocalActivity) and prepare_local_activities:
                     return _state(pending, _prepared_call(cmd, current_call_sequence))
                 if isinstance(cmd, RecordLocalActivity) and local_activity_executor is not None:

tests/fixtures/replay_regressions/pending-timer-after-signal-avro.json

@@ -0,0 +1,52 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "pending-timer-after-signal-avro",
+  "protocol_version": "1.19",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.pending-operation",
+    "input": [
+      "timer"
+    ],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "ActivityCompleted",
+      "payload": {
+        "sequence": 1,
+        "activity_type": "echo",
+        "result": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwoMc2FtcGxl"
+        }
+      }
+    },
+    {
+      "event_type": "TimerScheduled",
+      "payload": {
+        "sequence": 2,
+        "timer_id": "original-timer",
+        "delay_seconds": 120,
+        "fire_at": "2026-10-07T06:02:00Z"
+      }
+    },
+    {
+      "event_type": "SignalReceived",
+      "payload": {
+        "signal_name": "increment",
+        "arguments": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwwCBA4A"
+        },
+        "payload_codec": "avro"
+      }
+    }
+  ],
+  "expected": {
+    "command_sequence": []
+  }
+}

tests/test_child_workflow_policies.py

@@ -104,9 +104,8 @@ def test_existing_positional_constructor_and_omitted_wire_fields_are_preserved()
     result = replay(workflow("sequential", {
         "parent_close_policy": ParentClosePolicy.ABANDON, "cancellation_policy": CancellationPolicy.ABANDON,
     }), scheduled([StartChildWorkflow("child", ["argument"])]), [])
-    assert len(result.commands) == 1
-    assert isinstance(result.commands[0], StartChildWorkflow)
-    assert result.commands[0].cancellation_policy == CancellationPolicy.ABANDON
+    # Omitted historical fields retain their defaults, without starting another child.
+    assert result.commands == []
 
 
 @pytest.mark.parametrize("mode", ["sequential", "parallel", "selection"])
@@ -120,7 +119,7 @@ def test_changed_policies_fail_cold_replay_and_cancellation_delivery(mode: str,
         assert replay(workflow(mode, policies), history, [], run_id="run-1").commands == [CompleteWorkflow("request-1")]
     else:
         cold = replay(workflow(mode, policies), history, [])
-        if mode == "selection":
+        if mode in ("sequential", "selection"):
             assert cold.commands == []
         else:
             assert scheduled(cold.commands) == history

tests/test_committed_cancellation_scope_descendants.py

@@ -227,7 +227,8 @@ def test_cleanup_timer_retains_original_ancestor_receipt_and_each_scope_authorit
         "2026-10-04T00:00:10.123456Z",
     )
     value["task"].update({"lease_owner": "replacement", "workflow_task_attempt": 17})
-    assert workflow.commands_to_server_commands(run(cls, value).commands, "queue") == fresh
+    assert run(cls, value).commands == []
+    assert run(cls, value).commands == []
     append_cleanup_event(
         value,
         "TimerFired",

tests/test_committed_cancellation_scope_groups.py

@@ -217,7 +217,8 @@ def test_group_cleanup_timer_replays_original_receipt_and_narrower_authority(lay
         "cancellation_cleanup": snapshot,
     }, "2026-10-04T00:00:09.123456Z")
     value["task"].update({"workflow_task_attempt": 23, "lease_owner": "replacement"})
-    assert workflow.commands_to_server_commands(run(cls, value).commands, "queue") == first
+    assert run(cls, value).commands == []
+    assert run(cls, value).commands == []
     append_cleanup_event(value, "TimerFired", {
         "sequence": 8, "timer_id": "cleanup-timer", "delay_seconds": 1,
         "cancellation_scope_id": original.scope_id,

tests/test_committed_cancellation_scope_replay.py

@@ -76,7 +76,10 @@ def outer():  # type: ignore[no-untyped-def]
         field: snapshot[field] for field in ("scope_id", "request_id", "delivery_history_event_id")
     }
     assert wire[0]["cancellation_scope_id"] == snapshot["scope_id"]
-    assert workflow.commands_to_server_commands(run(Probe, value).commands, "queue") == wire
+    # The original cleanup timer and its authority are already durable.
+    # Cold replay waits for it rather than starting a second timer.
+    assert run(Probe, value).commands == []
+    assert run(Probe, value).commands == []
     append_cleanup_event(value, "TimerFired", {
         "sequence": 4, "timer_id": "cleanup-timer", "delay_seconds": 1,
         "cancellation_scope_id": snapshot["scope_id"],

tests/test_pending_operation_replay.py

@@ -0,0 +1,154 @@
+"""Scheduled scalar work stays the same operation when signals wake replay."""
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from durable_workflow import serializer, workflow
+from durable_workflow.errors import NonDeterministicReplayError
+from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay
+
+
+@workflow.defn(name="tests.replay.pending-operation")
+class PendingOperation:
+    def __init__(self) -> None:
+        self.total = 0
+        self.released = False
+
+    @workflow.signal("increment")
+    def increment(self, amount: int) -> None:
+        self.total += amount
+
+    @workflow.signal("release")
+    def release(self) -> None:
+        self.released = True
+
+    @workflow.query("total")
+    def current(self) -> int:
+        return self.total
+
+    def run(self, ctx: WorkflowContext, kind: str):  # type: ignore[no-untyped-def]
+        first = yield ctx.schedule_activity("echo", ["sample"])
+        if kind == "timer":
+            yield ctx.sleep(120)
+        elif kind == "activity":
+            yield ctx.schedule_activity("work", [first])
+        else:
+            yield ctx.start_child_workflow("child", [first])
+        yield ctx.wait_condition(lambda: self.released, key="release")
+        return self.total
+
+
+def initial_history() -> list[dict[str, Any]]:
+    return [
+        {"event_type": "ActivityCompleted", "payload": {
+            "sequence": 1, "activity_type": "echo", "result": serializer.envelope("sample"),
+        }},
+    ]
+
+
+OPENINGS = {
+    "timer": {"event_type": "TimerScheduled", "payload": {
+        "sequence": 2, "timer_id": "original-timer", "delay_seconds": 120,
+        "fire_at": "2026-10-07T06:02:00Z",
+    }},
+    "activity": {"event_type": "ActivityScheduled", "payload": {
+        "sequence": 2, "activity_type": "work", "activity_execution_id": "original-activity",
+    }},
+    "child": {"event_type": "ChildWorkflowScheduled", "payload": {
+        "sequence": 2, "workflow_type": "child", "child_workflow_instance_id": "original-child",
+    }},
+}
+
+
+@pytest.mark.parametrize("kind", OPENINGS)
+def test_signal_replay_does_not_reschedule_pending_operation(kind: str) -> None:
+    history = initial_history() + [OPENINGS[kind]]
+    for amount in (7, 11):
+        history.append({"event_type": "SignalReceived", "payload": {
+            "signal_name": "increment", "arguments": serializer.envelope([amount]),
+        }})
+        # Each call constructs a fresh workflow instance, as a replacement worker does.
+        outcome = replay(PendingOperation, history, [kind])
+        assert outcome.commands == []
+        assert query_state(PendingOperation, history, [kind], "total") == sum(
+            serializer.decode_envelope(event["payload"]["arguments"])[0]
+            for event in history if event["event_type"] == "SignalReceived"
+        )
+
+
+@pytest.mark.parametrize("kind", OPENINGS)
+def test_new_operation_is_still_scheduled(kind: str) -> None:
+    outcome = replay(PendingOperation, initial_history(), [kind])
+    assert len(outcome.commands) == 1
+    assert outcome.commands[0].to_server_command("queue")["type"] == {
+        "timer": "start_timer", "activity": "schedule_activity", "child": "start_child_workflow",
+    }[kind]
+
+
+def test_original_timer_completion_consumes_signals_once() -> None:
+    history = initial_history() + [OPENINGS["timer"],
+        {"event_type": "SignalReceived", "payload": {
+            "signal_name": "increment", "arguments": serializer.envelope([7]),
+        }},
+        {"event_type": "TimerFired", "payload": {"sequence": 2, "timer_id": "original-timer"}},
+        {"event_type": "SignalReceived", "payload": {
+            "signal_name": "release", "arguments": serializer.envelope([]),
+        }},
+    ]
+    outcome = replay(PendingOperation, history, ["timer"])
+    assert len(outcome.commands) == 1
+    assert isinstance(outcome.commands[0], CompleteWorkflow)
+    assert outcome.commands[0].result == 7
+
+
+def test_pending_timer_still_rejects_changed_delay() -> None:
+    changed = {"event_type": "TimerScheduled", "payload": {**OPENINGS["timer"]["payload"], "delay_seconds": 121}}
+    with pytest.raises(NonDeterministicReplayError) as captured:
+        replay(PendingOperation, initial_history() + [changed], ["timer"])
+    assert captured.value.workflow_sequence == 2
+    assert captured.value.recorded_event_types == ["TimerScheduled"]
+
+
+@pytest.mark.parametrize("kind", ["activity", "child"])
+def test_original_external_operation_completion_consumes_signals_once(kind: str) -> None:
+    terminal = {
+        "event_type": "ActivityCompleted" if kind == "activity" else "ChildRunCompleted",
+        "payload": {**OPENINGS[kind]["payload"], "result": serializer.envelope("original-result")},
+    }
+    history = initial_history() + [OPENINGS[kind],
+        {"event_type": "SignalReceived", "payload": {
+            "signal_name": "increment", "arguments": serializer.envelope([7]),
+        }}, terminal,
+        {"event_type": "SignalReceived", "payload": {
+            "signal_name": "release", "arguments": serializer.envelope([]),
+        }},
+    ]
+    assert replay(PendingOperation, history, [kind]).commands == [CompleteWorkflow(7)]
+
+
+@pytest.mark.parametrize("kind", OPENINGS)
+def test_recorded_operation_cannot_be_replaced_by_a_different_kind(kind: str) -> None:
+    different = "activity" if kind == "timer" else "timer"
+    with pytest.raises(NonDeterministicReplayError) as captured:
+        replay(PendingOperation, initial_history() + [OPENINGS[kind]], [different])
+    assert captured.value.workflow_sequence == 2
+
+
+def test_legacy_timer_without_recorded_delay_still_waits_for_original_operation() -> None:
+    legacy = {"event_type": "TimerScheduled", "payload": {"sequence": 2, "timer_id": "original-timer"}}
+    assert replay(PendingOperation, initial_history() + [legacy], ["timer"]).commands == []
+
+
+def test_already_duplicated_timers_are_not_silently_skipped() -> None:
+    duplicate = {"event_type": "TimerScheduled", "payload": {
+        **OPENINGS["timer"]["payload"], "sequence": 3, "timer_id": "duplicate-timer",
+    }}
+    history = initial_history() + [OPENINGS["timer"], duplicate,
+        {"event_type": "TimerFired", "payload": {"sequence": 2, "timer_id": "original-timer"}},
+        {"event_type": "TimerFired", "payload": {"sequence": 3, "timer_id": "duplicate-timer"}},
+    ]
+    with pytest.raises(NonDeterministicReplayError) as captured:
+        replay(PendingOperation, history, ["timer"])
+    assert captured.value.workflow_sequence == 3

tests/test_replay_regression_corpus.py

@@ -24,6 +24,7 @@
     GoldenTimeoutWaitWorkflow,
     GoldenVersionMarkerWorkflow,
 )
+from tests.test_pending_operation_replay import PendingOperation
 from tests.test_update_signal_condition_replay import (
     PostConditionReceiversWorkflow,
     UpdateSignalConditionTimerWorkflow,
@@ -265,6 +266,7 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
     NestedParallelPathWorkflow,
     ParallelMetadataProducerWorkflow,
     ParallelResultBindingWorkflow,
+    PendingOperation,
     PagedRecordedSideEffectsWorkflow,
     PostConditionReceiversWorkflow,
     RecordedSideEffectWorkflow,