feat: Implement workflow versioning support

cadence/_internal/workflow/context.py

@@ -3,10 +3,12 @@
 from contextlib import contextmanager
 from asyncio import get_running_loop
 from datetime import datetime, timedelta
+from json import JSONDecoder
 from math import ceil
 from typing import Iterator, Optional, Any, Unpack, Type, cast, Callable
 
 from cadence._internal.workflow.deterministic_event_loop import DeterministicEventLoop
+from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError
 from cadence._internal.workflow.memo import memo_to_proto
 from cadence._internal.workflow.retry_policy import retry_policy_to_proto
 from cadence._internal.workflow.statemachine.decision_manager import DecisionManager
@@ -28,7 +30,7 @@
     StartTimerDecisionAttributes,
 )
 from cadence.api.v1.tasklist_pb2 import TaskList, TaskListKind
-from cadence.data_converter import DataConverter
+from cadence.data_converter import DataConverter, DefaultDataConverter
 from cadence.workflow import (
     ActivityOptions,
     ChildWorkflowFuture,
@@ -37,6 +39,8 @@
     WorkflowCancellationInfo,
     WorkflowContext,
     WorkflowInfo,
+    VersioningOption,
+    DEFAULT_VERSION,
 )
 from cadence.api.v1.history_pb2 import WorkflowExecutionCancelRequestedEventAttributes
 
@@ -57,6 +61,9 @@ def __init__(
         self._replay_current_time: Optional[datetime] = None
         self._decision_manager = decision_manager
         self._cancellation_info: WorkflowCancellationInfo | None = None
+        # None represents a provisional DEFAULT_VERSION returned while replaying
+        # history before a Version marker becomes available.
+        self._versions: dict[str, int | None] = {}
 
     def info(self) -> WorkflowInfo:
         return self._info
@@ -333,6 +340,174 @@ def mutable_side_effect(
             self.data_converter().from_data(result_payload, [result_type])[0],
         )
 
+    def get_version(
+        self,
+        change_id: str,
+        min_supported: int,
+        max_supported: int,
+        *options: VersioningOption,
+    ) -> int:
+        self._validate_version_arguments(change_id, min_supported, max_supported)
+
+        if change_id in self._versions:
+            cached = self._versions[change_id]
+            if cached is None:
+                details = self._decision_manager.version_marker_details(change_id)
+                if details is not None:
+                    recorded = self._decode_recorded_version(change_id, details)
+                    self._validate_recorded_version(
+                        change_id, recorded, min_supported, max_supported, "recorded"
+                    )
+                    self._versions[change_id] = recorded
+                    if self._decision_manager.has_pending_version_marker(change_id):
+                        self._decision_manager.record_version_marker(change_id, details)
+                    return recorded
+                self._validate_recorded_version(
+                    change_id,
+                    DEFAULT_VERSION,
+                    min_supported,
+                    max_supported,
+                    "cached",
+                )
+                return DEFAULT_VERSION
+            self._validate_recorded_version(
+                change_id, cached, min_supported, max_supported, "cached"
+            )
+            return cached
+
+        details = self._decision_manager.version_marker_details(change_id)
+        if details is not None:
+            recorded = self._decode_recorded_version(change_id, details)
+            self._validate_recorded_version(
+                change_id, recorded, min_supported, max_supported, "recorded"
+            )
+            self._versions[change_id] = recorded
+            # Recreate the state machine so the preloaded marker is completed
+            # when its output event is replayed.
+            if self._decision_manager.has_pending_version_marker(change_id):
+                self._decision_manager.record_version_marker(change_id, details)
+            return recorded
+
+        # A version introduced while replaying old history must preserve the old
+        # code path. There is no marker to validate or state machine to create.
+        if self.is_replay_mode():
+            self._validate_recorded_version(
+                change_id,
+                DEFAULT_VERSION,
+                min_supported,
+                max_supported,
+                "markerless replay",
+            )
+            self._versions[change_id] = None
+            return DEFAULT_VERSION
+
+        selected = self._select_version(min_supported, max_supported, *options)
+        self._validate_selected_version(
+            change_id, selected, min_supported, max_supported
+        )
+        self._versions[change_id] = selected
+
+        # DEFAULT_VERSION is intentionally never written to history.
+        if selected == DEFAULT_VERSION:
+            return selected
+
+        self._decision_manager.record_version_marker(
+            change_id, self.data_converter().to_data([selected])
+        )
+        return selected
+
+    @staticmethod
+    def _validate_version_arguments(
+        change_id: str, min_supported: int, max_supported: int
+    ) -> None:
+        if not isinstance(change_id, str) or not change_id:
+            raise ValueError("change_id must be a non-empty str")
+        for name, value in (
+            ("min_supported", min_supported),
+            ("max_supported", max_supported),
+        ):
+            if isinstance(value, bool) or not isinstance(value, int):
+                raise ValueError(f"{name} must be an int")
+        if min_supported > max_supported:
+            raise ValueError("min_supported must not be greater than max_supported")
+
+    @staticmethod
+    def _select_version(
+        min_supported: int, max_supported: int, *options: VersioningOption
+    ) -> int:
+        custom_version: int | None = None
+        use_min_version = False
+        for option in options:
+            if not isinstance(option, VersioningOption):
+                raise ValueError("get_version options must be VersioningOption values")
+            if option._kind == "version":
+                custom_version = option._version
+            elif option._kind == "min":
+                use_min_version = True
+            else:
+                raise ValueError("invalid get_version option")
+        if custom_version is not None:
+            return custom_version
+        if use_min_version:
+            return min_supported
+        return max_supported
+
+    @staticmethod
+    def _validate_selected_version(
+        change_id: str, version: int, min_supported: int, max_supported: int
+    ) -> None:
+        if version < min_supported or version > max_supported:
+            raise ValueError(
+                f"selected version {version} for change_id {change_id!r} is outside "
+                f"the supported range [{min_supported}, {max_supported}]"
+            )
+
+    @staticmethod
+    def _validate_recorded_version(
+        change_id: str,
+        version: int,
+        min_supported: int,
+        max_supported: int,
+        source: str,
+    ) -> None:
+        if version < min_supported or version > max_supported:
+            raise FatalDecisionError(
+                f"{source} version {version} for change_id {change_id!r} is outside "
+                f"the supported range [{min_supported}, {max_supported}]"
+            )
+
+    def _decode_recorded_version(self, change_id: str, details: Payload) -> int:
+        if type(self.data_converter()) is DefaultDataConverter:
+            self._validate_default_version_details(change_id, details)
+        try:
+            version = self.data_converter().from_data(details, [int])[0]
+        except Exception as exc:
+            raise FatalDecisionError(
+                f"Unable to decode Version marker for change_id {change_id!r}"
+            ) from exc
+        if isinstance(version, bool) or not isinstance(version, int):
+            raise FatalDecisionError(
+                f"Version marker for change_id {change_id!r} did not contain an int"
+            )
+        return cast(int, version)
+
+    @staticmethod
+    def _validate_default_version_details(change_id: str, details: Payload) -> None:
+        if not details.data:
+            raise FatalDecisionError(
+                f"Version marker for change_id {change_id!r} had empty details"
+            )
+        try:
+            _, end = JSONDecoder(strict=False).raw_decode(details.data.decode())
+        except (UnicodeDecodeError, ValueError) as exc:
+            raise FatalDecisionError(
+                f"Version marker for change_id {change_id!r} had invalid details"
+            ) from exc
+        if end != len(details.data):
+            raise FatalDecisionError(
+                f"Version marker for change_id {change_id!r} had invalid details"
+            )
+
     def set_replay_current_time(self, current_time: datetime) -> None:
         """Set the current replay timestamp."""
         self._replay_current_time = current_time

cadence/_internal/workflow/statemachine/decision_manager.py

@@ -32,7 +32,9 @@
 )
 from cadence._internal.workflow.statemachine.marker_state_machine import (
     MUTABLE_SIDE_EFFECT_MARKER_NAME,
+    VERSION_MARKER_NAME,
     encode_marker_header,
+    has_marker_header,
     marker_context_id,
     marker_decision_id,
     KNOWN_MARKER_NAMES,
@@ -43,6 +45,7 @@
 )
 from cadence._internal.workflow.statemachine.nondeterminism import DeterminismTracker
 from cadence._internal.workflow.statemachine.cancellation import is_immediate_cancel
+from cadence._internal.workflow.deterministic_event_loop import FatalDecisionError
 from cadence._internal.workflow.statemachine.signal_external_workflow_state_machine import (
     signal_external_events,
     SignalExternalWorkflowStateMachine,
@@ -122,6 +125,8 @@ def __init__(self, event_loop: asyncio.AbstractEventLoop):
         self._determinism_tracker = DeterminismTracker()
         self._replaying = False
         self._recorded_marker_details: Dict[DecisionId, Payload] = {}
+        self._version_marker_event_ids: Dict[DecisionId, int] = {}
+        self._pending_version_marker_event_ids: Dict[DecisionId, int] = {}
         self._mutable_side_effects: dict[str, MutableSideEffectState] = {}
         self.state_machines: OrderedDict[DecisionId, DecisionStateMachine] = (
             OrderedDict()
@@ -282,6 +287,29 @@ def record_mutable_side_effect(
         state.value = result
         return result
 
+    def version_marker_details(self, change_id: str) -> Payload | None:
+        """Return a preloaded Python Version marker's details, if one exists."""
+        return self._recorded_marker_details.get(
+            marker_decision_id(VERSION_MARKER_NAME, change_id)
+        )
+
+    def has_pending_version_marker(self, change_id: str) -> bool:
+        """Whether a preloaded Version marker still awaits output-event routing."""
+        return (
+            marker_decision_id(VERSION_MARKER_NAME, change_id)
+            in self._pending_version_marker_event_ids
+        )
+
+    def record_version_marker(self, change_id: str, details: Payload) -> Payload:
+        """Record or replay a Version marker without consuming a sequence ID."""
+        return self._record_marker(
+            decision.RecordMarkerDecisionAttributes(
+                marker_name=VERSION_MARKER_NAME,
+                details=details,
+            ),
+            context_id=change_id,
+        )
+
     # ----- Workflow API -----
     def complete_workflow(self, decision: decision.Decision) -> None:
         if self._replaying:
@@ -311,8 +339,17 @@ def _add_state_machine(self, state: DecisionStateMachine) -> None:
 
     # ----- History routing -----
 
+    def preload_marker_event(self, event: history.HistoryEvent) -> None:
+        """Preload a marker before workflow code runs."""
+        self._handle_history_event(event, preloaded_marker=True)
+
     def handle_history_event(self, event: history.HistoryEvent) -> None:
         """Dispatch history event to typed handlers using the global transition map."""
+        self._handle_history_event(event, preloaded_marker=False)
+
+    def _handle_history_event(
+        self, event: history.HistoryEvent, *, preloaded_marker: bool
+    ) -> None:
         attr = event.WhichOneof("attributes")
         event_attributes = getattr(event, attr)
 
@@ -325,7 +362,10 @@ def handle_history_event(self, event: history.HistoryEvent) -> None:
             decision_type = event_action.decision_type
             action = event_action.action
             if decision_type is DecisionType.MARKER:
-                self._index_marker_details(event_attributes)
+                self._index_marker_details(event_attributes, event.event_id)
+                self._update_version_marker_preload_state(
+                    event_attributes, event.event_id, preloaded_marker
+                )
                 machine = self._state_machine_for_marker_event(event_attributes)
                 if machine is None:
                     return
@@ -341,6 +381,23 @@ def handle_history_event(self, event: history.HistoryEvent) -> None:
             if action.event_id_is_alias:
                 self.aliases[(decision_type, event.event_id)] = machine
 
+    def _update_version_marker_preload_state(
+        self,
+        attrs: history.MarkerRecordedEventAttributes,
+        event_id: int,
+        preloaded_marker: bool,
+    ) -> None:
+        if attrs.marker_name != VERSION_MARKER_NAME:
+            return
+        context_id = marker_context_id(attrs)
+        if context_id is None or not context_id:
+            return
+        marker_id = marker_decision_id(VERSION_MARKER_NAME, context_id)
+        if preloaded_marker:
+            self._pending_version_marker_event_ids[marker_id] = event_id
+        elif self._pending_version_marker_event_ids.get(marker_id) == event_id:
+            del self._pending_version_marker_event_ids[marker_id]
+
     def _state_machine_for_event(
         self,
         event_id: int,
@@ -401,7 +458,7 @@ def _state_machine_for_marker_event(
         return machine
 
     def _index_marker_details(
-        self, attrs: history.MarkerRecordedEventAttributes
+        self, attrs: history.MarkerRecordedEventAttributes, event_id: int
     ) -> None:
         """Store the user payload from a recorded marker event, keyed by its marker DecisionId.
 
@@ -413,8 +470,26 @@ def _index_marker_details(
             return
         context_id = marker_context_id(attrs)
         if context_id is None:
+            if attrs.marker_name == VERSION_MARKER_NAME and has_marker_header(attrs):
+                raise FatalDecisionError(
+                    "Version marker contains an invalid Python MarkerHeader"
+                )
             return
+        if attrs.marker_name == VERSION_MARKER_NAME and not context_id:
+            raise FatalDecisionError(
+                "Version marker contains an invalid Python MarkerHeader"
+            )
         marker_id = marker_decision_id(attrs.marker_name, context_id)
+        if attrs.marker_name == VERSION_MARKER_NAME:
+            previous_event_id = self._version_marker_event_ids.get(marker_id)
+            if previous_event_id is None:
+                self._version_marker_event_ids[marker_id] = event_id
+            elif previous_event_id != event_id:
+                raise FatalDecisionError(
+                    f"Received duplicate Version marker for change_id {context_id!r}"
+                )
+            else:
+                return
         details = Payload(data=attrs.details.data)
         self._recorded_marker_details[marker_id] = details
         mutable_info = mutable_side_effect_marker_info(attrs)
@@ -439,7 +514,9 @@ def _start_execution(self, replaying: bool, outcomes: List[history.HistoryEvent]
         for event in outcomes:
             self._determinism_tracker.add_expectation(event)
             if event.HasField("marker_recorded_event_attributes"):
-                self._index_marker_details(event.marker_recorded_event_attributes)
+                self._index_marker_details(
+                    event.marker_recorded_event_attributes, event.event_id
+                )
 
     def _end_execution(self) -> None:
         if self._replaying:

cadence/_internal/workflow/statemachine/marker_state_machine.py

@@ -76,6 +76,14 @@ def marker_header(
         return None
 
 
+def has_marker_header(
+    attrs: decision.RecordMarkerDecisionAttributes
+    | history.MarkerRecordedEventAttributes,
+) -> bool:
+    """Return whether this marker attempts to use the Python MarkerHeader format."""
+    return MARKER_HEADER_KEY in attrs.header.fields
+
+
 def marker_context_id(
     attrs: decision.RecordMarkerDecisionAttributes
     | history.MarkerRecordedEventAttributes,

cadence/_internal/workflow/workflow_engine.py

@@ -190,8 +190,7 @@ def _process_decision_events(
                             "replay_mode": decision_events.replay,
                         },
                     )
-                    # Process through state machines (DecisionsHelper now delegates to DecisionManager)
-                    self._decision_manager.handle_history_event(marker_event)
+                    self._decision_manager.preload_marker_event(marker_event)
 
                 # Phase 2: Apply input events in history order.
                 for event in decision_events.input: