Preserve patch decisions across old and duplicate-marker histories

CHANGELOG.md

@@ -6,6 +6,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.5.1] - 2026-10-10
+
+### Fixed
+- Select the legacy patch branch when an existing unmarked history has already
+  reached its next durable operation, preserving pending activities without
+  scheduling them again.
+- Reuse one version decision per change ID and validate each call's supported
+  range. Repeated patch calls now record one marker on new histories.
+- Replay consistent duplicate markers written by earlier Python SDKs without
+  rewriting their histories or shifting committed cancellation boundaries.
+  Conflicting decisions remain explicit replay errors.
+
 ## [2.5.0] - 2026-10-08
 
 ### Added

docs/sdk-reference.md

@@ -511,6 +511,38 @@ dispatch-rate limits, so automation can detect whether local worker slots,
 queue caps, namespace caps, or downstream dispatch budget groups are
 constraining throughput.
 
+## Evolve workflow code with patches
+
+Use a stable change ID when adding a branch to a workflow that already has
+running executions:
+
+```python
+def run(self, ctx):
+    if (yield ctx.patched("require-review")):
+        yield ctx.schedule_activity("review", [])
+    return (yield ctx.schedule_activity("fulfil", []))
+```
+
+New histories record version `1` and take the patched branch. An old unmarked
+history that already reached the next operation takes the legacy branch, with
+version `-1`, and keeps that operation's original sequence. A pending activity
+is awaited without scheduling it again.
+
+Each change ID has one decision per replay. Repeated `patched()` calls reuse
+that decision. `get_version()` also keeps the first selected version, even if a
+later call raises its maximum. Every call's supported range must include the
+recorded decision. Keep change IDs stable and do not switch an ID between
+`get_version()` and the patch helpers.
+
+After retiring the legacy branch, use `yield ctx.deprecate_patch("require-review")`
+at the same code boundary. It retains the marker without returning a branch
+decision. Keep it while retained histories still need that boundary.
+
+Earlier Python SDKs could record repeated markers for one change ID. Replay
+accepts consistent duplicates and retains their stored positions, including
+committed cancellation boundaries. Conflicting decisions remain a typed replay
+error. Existing histories are not rewritten or discarded.
+
 ## Replay captured histories
 
 Signals can update workflow state while a timer, remote activity or child is

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.5.0"
+version = "2.5.1"
 description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -71,9 +71,9 @@ 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.5.0"
-registry-version = "2.5.0"
-supported-server-versions = "2.5.0"
+product-train = "2.5.1"
+registry-version = "2.5.1"
+supported-server-versions = "2.5.16"
 worker-protocol-version = "1.19"
 control-plane-version = "2"
 durable-selection = true

src/durable_workflow/workflow.py

@@ -5185,6 +5185,96 @@ def _same_logical_condition_wait(opened: Mapping[str, Any], cmd: WaitCondition)
     pending: list[Command] = []
     advanced_cmd: Any = None
     terminal_condition_reopen_cmd: WaitCondition | None = None
+    version_decisions: dict[str, tuple[int, str]] = {}
+
+    def _version_error(detail: str, step: _RecordedStep | None = None) -> NonDeterministicReplayError:
+        return NonDeterministicReplayError(
+            step.workflow_sequence if step is not None else authored_sequence,
+            "compatible version marker",
+            step.event_types if step is not None else [],
+            detail=detail,
+        )
+
+    def _recorded_version(step: _RecordedStep) -> int:
+        if step.history_index is None:
+            raise _version_error("Version marker is missing its history position.", step)
+        payload = events[step.history_index].get("payload") or {}
+        version = payload.get("version")
+        if type(version) is not int:
+            raise _version_error("Version marker must contain an integer version.", step)
+        change_id = payload.get("change_id")
+        if "change_id" in payload and (not isinstance(change_id, str) or not change_id.strip()):
+            raise _version_error("Version marker must retain a non-empty change ID.", step)
+        if "min_supported" in payload or "max_supported" in payload:
+            minimum, maximum = payload.get("min_supported"), payload.get("max_supported")
+            if (type(minimum) is not int or type(maximum) is not int
+                    or not minimum <= version <= maximum):
+                raise _version_error("Recorded version marker has an invalid supported range.", step)
+        return version
+
+    def _consume_duplicate_version_markers() -> None:
+        nonlocal result_cursor, authored_sequence
+        if not version_decisions:
+            return
+        while result_cursor < len(recorded_steps):
+            step = recorded_steps[result_cursor]
+            change_id = step.details.get("change_id")
+            if step.shape != "version marker" or change_id not in version_decisions:
+                return
+            if _next_unconsumed_recorded_step() is not step:
+                return
+            version = _recorded_version(step)
+            if version != version_decisions[change_id][0]:
+                raise _version_error(f"Conflicting recorded decisions for change ID {change_id!r}.", step)
+            # Earlier SDKs emitted a marker for every invocation. Consume these
+            # aliases while retaining their physical positions for cancellation
+            # and prepared operations. The stored history itself is unchanged.
+            result_cursor += 1
+            authored_sequence = max(authored_sequence, step.workflow_sequence + 1)
+
+    def _resolve_version(command: RecordVersionMarker) -> Any:
+        nonlocal result_cursor, authored_sequence
+        change_id = command.change_id
+        if not isinstance(change_id, str) or not change_id.strip():
+            raise _version_error("Version markers require a stable non-empty change ID.")
+        if (type(command.min_supported) is not int or type(command.max_supported) is not int
+                or command.min_supported > command.max_supported):
+            raise _version_error(f"Invalid supported range for change ID {change_id!r}.")
+        family = "version" if command.result_kind == "version" else "patch"
+        decision = version_decisions.get(change_id)
+        step = _next_unconsumed_recorded_step()
+        if decision is not None:
+            version, original_family = decision
+            if original_family != family:
+                raise _version_error(f"Change ID {change_id!r} cannot switch between version and patch helpers.", step)
+        elif step is not None and step.shape == "version marker":
+            _assert_step_matches(command, step)
+            version = _recorded_version(step)
+            result_cursor += 1
+            authored_sequence = max(authored_sequence, step.workflow_sequence + 1)
+        elif step is not None or (cancellation.delivery is not None and not cancellation_consumed) or any(
+            sequence not in consumed_scope_deliveries for sequence in committed_scopes.deliveries
+        ) or any(_history_event_type(event) in {
+            "WorkflowCompleted", "WorkflowFailed", "WorkflowCancelled", "WorkflowTerminated",
+            "WorkflowContinuedAsNew", "WorkflowTimedOut",
+        } for event in events):
+            # An unmarked recorded command proves this history passed the new
+            # patch site. Leave that command available for its original call.
+            version = -1
+        else:
+            version = command.version
+            if type(version) is not int:
+                raise _version_error(f"Invalid version for change ID {change_id!r}.")
+            ctx.logger._set_replaying(False)
+            pending.append(command)
+            authored_sequence += 1
+        if not command.min_supported <= version <= command.max_supported:
+            raise _version_error(
+                f"Recorded version {version} for change ID {change_id!r} is outside "
+                f"the supported range {command.min_supported}..{command.max_supported}.", step,
+            )
+        version_decisions.setdefault(change_id, (version, family))
+        return _version_marker_result(command, version)
 
     def _parallel_leaf_kind(command: Any) -> str:
         if isinstance(command, ScheduleActivity | RecordLocalActivity):
@@ -6064,6 +6154,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
     replay_token = _ACTIVE_WORKFLOW_REPLAY.set(ctx)
     try:
         while True:
+            _consume_duplicate_version_markers()
             # 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
@@ -6081,6 +6172,9 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                 first = False
             _apply_due_receivers()
             _validate_local_activity_policies(cmd)
+            if isinstance(cmd, RecordVersionMarker):
+                next_value = _resolve_version(cmd)
+                continue
             if prepare_local_activities and isinstance(cmd, list | SelectGroup) and _contains_local_activity(cmd) and (
                 not prepare_local_activity_groups or isinstance(cmd, SelectGroup)
             ):
@@ -6497,18 +6591,6 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                 pending.append(cmd)
                 next_value = None
                 continue
-            if isinstance(cmd, RecordVersionMarker):
-                if result_cursor < len(resolved_results):
-                    _assert_next_step_matches(cmd)
-                    val = resolved_results[result_cursor]
-                    result_cursor += 1
-                    next_value = _version_marker_result(cmd, val)
-                    continue
-                ctx.logger._set_replaying(False)
-                _assert_pending_step_matches(cmd)
-                pending.append(cmd)
-                next_value = _version_marker_result(cmd, cmd.version)
-                continue
             if isinstance(cmd, WaitCondition):
                 # A condition that is already true is not persisted. Apply
                 # receivers at the yield boundary before deciding whether the

tests/fixtures/replay_regressions/patch-cancellation-boundary-after-one-marker.json

@@ -0,0 +1,32 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "patch-cancellation-boundary-after-one-marker",
+  "protocol_version": "1.20",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.patch.cancellation", "input": [], "payload_codec": "avro"},
+  "history": [
+    {"event_type": "VersionMarkerRecorded", "payload": {
+      "sequence": 1, "change_id": "same", "version": 1, "min_supported": -1, "max_supported": 1
+    }},
+    {"event_type": "ActivityScheduled", "payload": {"sequence": 2, "activity_type": "old"}},
+    {
+      "event_type": "CooperativeCancellationRequested",
+      "workflow_command_id": "request-1",
+      "recorded_at": "2026-09-30T12:00:00.000120Z",
+      "payload": {
+        "workflow_command_id": "request-1", "workflow_run_id": "run-1",
+        "cleanup_deadline_at": "2026-09-30T12:10:00Z"
+      }
+    },
+    {
+      "event_type": "CooperativeCancellationDelivered",
+      "workflow_command_id": "request-1",
+      "payload": {
+        "workflow_command_id": "request-1", "workflow_run_id": "run-1",
+        "sequence": 2, "call_kind": "activity"
+      }
+    }
+  ],
+  "expected": {"type": "complete_workflow", "result": [[true, true], "cancelled"]}
+}

tests/fixtures/replay_regressions/retained-duplicate-patch-markers.json

@@ -0,0 +1,22 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "retained-duplicate-patch-markers",
+  "protocol_version": "1.19",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.patch.repeated", "input": [false], "payload_codec": "avro"},
+  "history": [
+    {"event_type": "VersionMarkerRecorded", "payload": {
+      "sequence": 1, "change_id": "same", "version": 1, "min_supported": -1, "max_supported": 1
+    }},
+    {"event_type": "VersionMarkerRecorded", "payload": {
+      "sequence": 2, "change_id": "same", "version": 1, "min_supported": -1, "max_supported": 1
+    }},
+    {"event_type": "ActivityScheduled", "payload": {"sequence": 3, "activity_type": "old"}},
+    {"event_type": "ActivityCompleted", "payload": {
+      "sequence": 3, "activity_type": "old", "payload_codec": "avro",
+      "result": {"codec": "avro", "blob": "wwHioz3/VYAiNwTGAQ=="}
+    }}
+  ],
+  "expected": {"type": "complete_workflow", "result": [true, true, 99]}
+}

tests/fixtures/replay_regressions/unmarked-patch-before-completed-activity.json

@@ -0,0 +1,16 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "unmarked-patch-before-completed-activity",
+  "protocol_version": "1.19",
+  "bindings": ["python"],
+  "workflow": {"type": "tests.patch.insertion", "input": [], "payload_codec": "avro"},
+  "history": [
+    {"event_type": "ActivityScheduled", "payload": {"sequence": 1, "activity_type": "old"}},
+    {"event_type": "ActivityCompleted", "payload": {
+      "sequence": 1, "activity_type": "old", "payload_codec": "avro",
+      "result": {"codec": "avro", "blob": "wwHioz3/VYAiNwTGAQ=="}
+    }}
+  ],
+  "expected": {"type": "complete_workflow", "result": 99}
+}

tests/test_cooperative_remote_worker.py

@@ -47,22 +47,29 @@ def stop_receipt() -> dict[str, Any]:
             "heartbeat_recorded": False, "history_event_id": "receipt-event"}
 
 
+def write_pid_marker(marker: str) -> None:
+    target = Path(marker)
+    pending = target.with_name(target.name + ".pending")
+    pending.write_text(str(os.getpid()))
+    pending.replace(target)
+
+
 async def blocked_async(marker: str, heartbeat: bool = False) -> None:
     if heartbeat:
         await activity.context().heartbeat({"authored": True})
-    Path(marker).write_text(str(os.getpid()))
+    write_pid_marker(marker)
     await asyncio.Event().wait()
 
 
 def blocked_sync(marker: str, heartbeat: bool = False) -> None:
     if heartbeat:
         asyncio.run(activity.context().heartbeat({"authored": True}))
-    Path(marker).write_text(str(os.getpid()))
+    write_pid_marker(marker)
     time.sleep(60)
 
 
 def released_sync(marker: str) -> bytes:
-    Path(marker).write_text(str(os.getpid()))
+    write_pid_marker(marker)
     while not Path(marker + ".release").exists():
         time.sleep(0.02)
     return b"late result"