feat: marker state machine

cadence/_internal/workflow/context.py

@@ -10,9 +10,18 @@
 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
+from cadence._internal.workflow.statemachine.marker_state_machine import (
+    SIDE_EFFECT_MARKER_NAME,
+)
 from cadence.api.v1 import workflow_pb2
-from cadence.api.v1.common_pb2 import ActivityType, WorkflowType, WorkflowExecution
+from cadence.api.v1.common_pb2 import (
+    ActivityType,
+    Payload,
+    WorkflowType,
+    WorkflowExecution,
+)
 from cadence.api.v1.decision_pb2 import (
+    RecordMarkerDecisionAttributes,
     ScheduleActivityTaskDecisionAttributes,
     SignalExternalWorkflowExecutionDecisionAttributes,
     StartChildWorkflowExecutionDecisionAttributes,
@@ -257,6 +266,25 @@ def is_replay_mode(self) -> bool:
         """Check if the workflow is currently in replay mode."""
         return self._replay_mode
 
+    def side_effect(
+        self,
+        fn: Callable[[], ResultType],
+        result_type: Type[ResultType],
+    ) -> ResultType:
+        details = Payload()
+        if not self.is_replay_mode():
+            details = self.data_converter().to_data([fn()])
+        result_payload = self._decision_manager.record_marker(
+            RecordMarkerDecisionAttributes(
+                marker_name=SIDE_EFFECT_MARKER_NAME,
+                details=details,
+            )
+        )
+        return cast(
+            ResultType,
+            self.data_converter().from_data(result_payload, [result_type])[0],
+        )
+
     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/cancellation.py

@@ -10,27 +10,26 @@
 from msgspec import json
 
 
-MARKER_PREFIX = "Cancel_"
+CANCEL_MARKER_NAME = "Cancel"
 
 
 def is_immediate_cancel(marker: history.MarkerRecordedEventAttributes) -> bool:
-    return marker.marker_name.startswith(MARKER_PREFIX)
+    return marker.marker_name == CANCEL_MARKER_NAME
 
 
 def to_marker(
     decision_id: DecisionId, props: Dict[str, Any]
 ) -> decision.RecordMarkerDecisionAttributes:
-    data = props | {"type": decision_id.decision_type.name}
+    data = props | {"id": decision_id.id, "type": decision_id.decision_type.name}
     return decision.RecordMarkerDecisionAttributes(
-        marker_name=MARKER_PREFIX + decision_id.id,
+        marker_name=CANCEL_MARKER_NAME,
         details=Payload(data=json.encode(data)),
     )
 
 
 def from_marker(
     marker: history.MarkerRecordedEventAttributes,
 ) -> Tuple[DecisionId, Dict[str, Any]]:
-    decision_id = marker.marker_name.replace(MARKER_PREFIX, "")
     props = json.decode(marker.details.data)
-    decision_type = DecisionType[props.pop("type")]
-    return DecisionId(decision_type, decision_id), props
+    decision_id = DecisionId(DecisionType[props.pop("type")], props.pop("id"))
+    return decision_id, props

cadence/_internal/workflow/statemachine/decision_manager.py

@@ -1,4 +1,5 @@
 import asyncio
+import logging
 from collections import OrderedDict
 from contextlib import contextmanager
 from dataclasses import dataclass
@@ -29,7 +30,17 @@
     Action,
     resolve_id_attr,
 )
+from cadence._internal.workflow.statemachine.marker_state_machine import (
+    encode_marker_header,
+    marker_context_id,
+    marker_decision_id,
+    KNOWN_MARKER_NAMES,
+    MARKER_HEADER_KEY,
+    marker_events,
+    MarkerStateMachine,
+)
 from cadence._internal.workflow.statemachine.nondeterminism import DeterminismTracker
+from cadence._internal.workflow.statemachine.cancellation import is_immediate_cancel
 from cadence._internal.workflow.statemachine.signal_external_workflow_state_machine import (
     signal_external_events,
     SignalExternalWorkflowStateMachine,
@@ -41,6 +52,8 @@
 from cadence.api.v1 import decision, history
 from cadence.api.v1.common_pb2 import Payload, WorkflowExecution
 
+logger = logging.getLogger(__name__)
+
 DecisionAlias = tuple[DecisionType, str | int]
 
 
@@ -85,6 +98,7 @@ class DecisionManager:
             DecisionType.TIMER: timer_events,
             DecisionType.CHILD_WORKFLOW: child_workflow_events,
             DecisionType.SIGNAL: signal_external_events,
+            DecisionType.MARKER: marker_events,
         }
     )
 
@@ -93,6 +107,7 @@ def __init__(self, event_loop: asyncio.AbstractEventLoop):
         self._id_counter = 0
         self._determinism_tracker = DeterminismTracker()
         self._replaying = False
+        self._recorded_marker_details: Dict[DecisionId, Payload] = {}
         self.state_machines: OrderedDict[DecisionId, DecisionStateMachine] = (
             OrderedDict()
         )
@@ -164,6 +179,35 @@ def signal_external_workflow(
         self._add_state_machine(machine)
         return future
 
+    # ----- Marker API -----
+
+    def record_marker(
+        self,
+        attrs: decision.RecordMarkerDecisionAttributes,
+    ) -> Payload:
+        if not attrs.marker_name:
+            raise ValueError("marker_name is required")
+        context_id = self._next_id()
+
+        # Metadata (the context_id) goes in the Header; Details stays the raw user payload.
+        # The header is set in-place so the state machine carries it on the wire.
+        attrs.header.fields[MARKER_HEADER_KEY].CopyFrom(
+            encode_marker_header(context_id)
+        )
+
+        marker_id = marker_decision_id(attrs.marker_name, context_id)
+        result = Payload(data=attrs.details.data)
+
+        if self._replaying:
+            self._determinism_tracker.validate_action(attrs)
+            history_value = self._recorded_marker_details.get(marker_id)
+            if history_value is not None:
+                result = Payload(data=history_value.data)
+
+        machine = MarkerStateMachine(attrs, attrs.marker_name, context_id)
+        self._add_state_machine(machine)
+        return result
+
     # ----- Workflow API -----
     def complete_workflow(self, decision: decision.Decision) -> None:
         if self._replaying:
@@ -206,9 +250,15 @@ def handle_history_event(self, event: history.HistoryEvent) -> None:
         if event_action is not None:
             decision_type = event_action.decision_type
             action = event_action.action
-            machine = self._state_machine_for_event(
-                event.event_id, decision_type, action, event_attributes
-            )
+            if decision_type is DecisionType.MARKER:
+                self._index_marker_details(event_attributes)
+                machine = self._state_machine_for_marker_event(event_attributes)
+                if machine is None:
+                    return
+            else:
+                machine = self._state_machine_for_event(
+                    event.event_id, decision_type, action, event_attributes
+                )
 
             action.fn(machine, event_attributes)
 
@@ -233,6 +283,66 @@ def _state_machine_for_event(
             raise KeyError(f"Event {event_id} references unknown state machine {alias}")
         return machine
 
+    def _state_machine_for_marker_event(
+        self,
+        event_attributes: history.MarkerRecordedEventAttributes,
+    ) -> DecisionStateMachine | None:
+        # Immediate-cancellation markers are matched by DeterminismTracker, not routed
+        # through a MarkerStateMachine — handle them explicitly rather than via decode-failure below.
+        if is_immediate_cancel(event_attributes):
+            logger.debug(
+                "Marker '%s' is the immediate-cancellation marker — handled by "
+                "DeterminismTracker, not routed through a MarkerStateMachine",
+                event_attributes.marker_name,
+            )
+            return None
+
+        # Marker events are preloaded before workflow code runs. If no marker
+        # decision has been requested yet, keep the preloaded event as a no-op.
+        context_id = marker_context_id(event_attributes)
+        if context_id is None:
+            logger.debug(
+                "Marker '%s' has no marker header — skipping routing "
+                "(produced by another SDK or pre-header history)",
+                event_attributes.marker_name,
+            )
+            return None
+        marker_id = marker_decision_id(event_attributes.marker_name, context_id)
+        machine = self.aliases.get((marker_id.decision_type, marker_id.id), None)
+        if machine is None:
+            if event_attributes.marker_name not in KNOWN_MARKER_NAMES:
+                logger.warning(
+                    "Marker event with unknown marker_name '%s' (key='%s') has no "
+                    "matching state machine and will be dropped",
+                    event_attributes.marker_name,
+                    marker_id.id,
+                )
+            else:
+                logger.debug(
+                    "No state machine for marker '%s' yet (key='%s') — "
+                    "marker is preloaded before workflow code runs",
+                    event_attributes.marker_name,
+                    marker_id.id,
+                )
+        return machine
+
+    def _index_marker_details(
+        self, attrs: history.MarkerRecordedEventAttributes
+    ) -> None:
+        """Store the user payload from a recorded marker event, keyed by its marker DecisionId.
+
+        Called for every MarkerRecordedEvent (both during preload and output replay) so
+        that record_marker can return the historical value on replay without routing
+        through Expectation.properties.
+        """
+        if is_immediate_cancel(attrs):
+            return
+        context_id = marker_context_id(attrs)
+        if context_id is None:
+            return
+        marker_id = marker_decision_id(attrs.marker_name, context_id)
+        self._recorded_marker_details[marker_id] = Payload(data=attrs.details.data)
+
     # ---- Non-determinism ----
     @contextmanager
     def track_nondeterminism(
@@ -246,6 +356,8 @@ def _start_execution(self, replaying: bool, outcomes: List[history.HistoryEvent]
         self._replaying = replaying
         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)
 
     def _end_execution(self) -> None:
         if self._replaying:

cadence/_internal/workflow/statemachine/marker_state_machine.py

@@ -0,0 +1,109 @@
+from msgspec import DecodeError, Struct, json
+
+from cadence._internal.workflow.statemachine.decision_state_machine import (
+    BaseDecisionStateMachine,
+    DecisionId,
+    DecisionState,
+    DecisionType,
+)
+from cadence._internal.workflow.statemachine.event_dispatcher import EventDispatcher
+from cadence.api.v1 import decision, history
+from cadence.api.v1.common_pb2 import Payload
+
+# Marker type names match the Go SDK constants:
+# https://github.com/cadence-workflow/cadence-go-client/blob/727b555be0fd0f65ad201832ba078b661919034e/internal/internal_decision_state_machine.go#L160-L163
+SIDE_EFFECT_MARKER_NAME = "SideEffect"
+VERSION_MARKER_NAME = "Version"
+LOCAL_ACTIVITY_MARKER_NAME = "LocalActivity"
+MUTABLE_SIDE_EFFECT_MARKER_NAME = "MutableSideEffect"
+
+# TODO(local-activities): when we implement LocalActivity markers, keep the split of
+# metadata in the Header and the raw payload in the Details. Storing the DecisionID under a
+# consistent header key too may simplify the code, though each marker type will always need
+# some custom logic. LocalActivity is the hard case: it must also carry a Failure
+# (reason: str, details: bytes), so MarkerHeader will need to grow to represent that.
+
+KNOWN_MARKER_NAMES = frozenset(
+    {
+        SIDE_EFFECT_MARKER_NAME,
+        VERSION_MARKER_NAME,
+        LOCAL_ACTIVITY_MARKER_NAME,
+        MUTABLE_SIDE_EFFECT_MARKER_NAME,
+    }
+)
+
+MARKER_HEADER_KEY = "MarkerHeader"
+
+marker_events = EventDispatcher()
+
+
+class MarkerHeader(Struct):
+    context_id: str
+
+
+def encode_marker_header(context_id: str) -> Payload:
+    """Serialize marker metadata for storage under MARKER_HEADER_KEY."""
+    return Payload(data=json.encode(MarkerHeader(context_id=context_id)))
+
+
+def marker_context_id(
+    attrs: decision.RecordMarkerDecisionAttributes
+    | history.MarkerRecordedEventAttributes,
+) -> str | None:
+    """Read the context_id from a marker's Header.
+
+    record_marker always sets the header and callers filter the immediate-cancellation
+    marker upstream, so None is a defensive fallback for a missing/malformed header, not
+    a case hit in normal replay.
+    """
+    if MARKER_HEADER_KEY not in attrs.header.fields:
+        return None
+    try:
+        return json.decode(
+            attrs.header.fields[MARKER_HEADER_KEY].data, type=MarkerHeader
+        ).context_id
+    except DecodeError:
+        return None
+
+
+def marker_decision_id(marker_name: str, context_id: str) -> DecisionId:
+    """Build the DecisionId that identifies a marker instance.
+
+    Format matches the Go SDK's fmt.Sprintf("%v_%v", markerName, contextID):
+    https://github.com/cadence-workflow/cadence-go-client/blob/727b555be0fd0f65ad201832ba078b661919034e/internal/internal_decision_state_machine.go#L794
+    """
+    return DecisionId(DecisionType.MARKER, f"{marker_name}_{context_id}")
+
+
+class MarkerStateMachine(BaseDecisionStateMachine):
+    """State machine for RecordMarker decisions."""
+
+    request: decision.RecordMarkerDecisionAttributes
+    _marker_name: str
+    _context_id: str
+
+    def __init__(
+        self,
+        request: decision.RecordMarkerDecisionAttributes,
+        marker_name: str,
+        context_id: str,
+    ) -> None:
+        super().__init__()
+        self.request = request
+        self._marker_name = marker_name
+        self._context_id = context_id
+
+    def get_id(self) -> DecisionId:
+        return marker_decision_id(self._marker_name, self._context_id)
+
+    def get_decision(self) -> decision.Decision | None:
+        if self.state is DecisionState.REQUESTED:
+            return decision.Decision(record_marker_decision_attributes=self.request)
+        return None
+
+    def request_cancel(self, message: str | None = None) -> bool:
+        return False
+
+    @marker_events.event()
+    def handle_recorded(self, _: history.MarkerRecordedEventAttributes) -> None:
+        self._transition(DecisionState.COMPLETED)