diff --git a/cadence/_internal/workflow/context.py b/cadence/_internal/workflow/context.py index adb0b05d..c3f72644 100644 --- a/cadence/_internal/workflow/context.py +++ b/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 diff --git a/cadence/_internal/workflow/statemachine/cancellation.py b/cadence/_internal/workflow/statemachine/cancellation.py index f98ee9a0..8de3ade1 100644 --- a/cadence/_internal/workflow/statemachine/cancellation.py +++ b/cadence/_internal/workflow/statemachine/cancellation.py @@ -10,19 +10,19 @@ 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)), ) @@ -30,7 +30,6 @@ def to_marker( 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 diff --git a/cadence/_internal/workflow/statemachine/decision_manager.py b/cadence/_internal/workflow/statemachine/decision_manager.py index edc0a0d6..97d57866 100644 --- a/cadence/_internal/workflow/statemachine/decision_manager.py +++ b/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: diff --git a/cadence/_internal/workflow/statemachine/marker_state_machine.py b/cadence/_internal/workflow/statemachine/marker_state_machine.py new file mode 100644 index 00000000..96ca226a --- /dev/null +++ b/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) diff --git a/cadence/_internal/workflow/statemachine/nondeterminism.py b/cadence/_internal/workflow/statemachine/nondeterminism.py index b5fde434..3773c68a 100644 --- a/cadence/_internal/workflow/statemachine/nondeterminism.py +++ b/cadence/_internal/workflow/statemachine/nondeterminism.py @@ -15,6 +15,11 @@ DecisionId, DecisionType, ) +from cadence._internal.workflow.statemachine.marker_state_machine import ( + marker_context_id, + marker_decision_id, + VERSION_MARKER_NAME, +) from cadence.api.v1 import decision, history @@ -113,12 +118,12 @@ def _add_expectations( self._expectations[decision_id] = to_expect - def validate_action(self, attributes: Any) -> None: + def validate_action(self, attributes: Any) -> Expectation | None: props = to_expectation(attributes) if props is None: - return + return None - self._validate_expectation(props) + return self._validate_expectation(props) def validate_cancel(self, decision_id: DecisionId) -> None: # Cancellation may happen automatically, ignore it @@ -128,7 +133,8 @@ def validate_cancel(self, decision_id: DecisionId) -> None: Expectation(decision_id=decision_id, properties=CANCEL) ) - def _validate_expectation(self, actual: Expectation) -> None: + def _validate_expectation(self, actual: Expectation) -> Expectation: + """Returns the matched Expectation on success, or raises NonDeterminismError.""" if not self._expectations: self._fail(None, actual) @@ -157,7 +163,7 @@ def _validate_expectation(self, actual: Expectation) -> None: # A: [Schedule, Cancel] if len(upcoming) == 1: # Using the above example, If they rewrote it to be: - # Schedula A + # Schedule A # Cancel A # Schedule B # All expectations would be met, but we still need to report that it's out of order. @@ -166,13 +172,15 @@ def _validate_expectation(self, actual: Expectation) -> None: self._fail(next_expectation, actual) if actual == upcoming[0]: + matched = upcoming[0] del self._expectations[actual.decision_id] + return matched else: self._fail(next_expectation, actual) else: next_for_decision = upcoming[0] if next_for_decision == actual: - upcoming.pop(0) + return upcoming.pop(0) else: self._fail(next_expectation, actual) @@ -355,6 +363,35 @@ def _( ) +# Markers - Enforce marker type (marker_name) and instance id (context_id from the Header) +# via the DecisionId alone; the Expectation body is empty since the DecisionId already +# captures both the type and the identity. +# +# Version markers are exempt: adding/removing a version check is always safe, so both +# handlers return None (no expectation on either side). This matches Go SDK behaviour. +# +# Details are intentionally excluded from Expectation; DecisionManager stores recorded +# values in _recorded_marker_details and returns the historical value on replay directly. +@to_expectation.register +def _(attrs: decision.RecordMarkerDecisionAttributes) -> Expectation | None: + context_id = marker_context_id(attrs) + if context_id is None: + return None + if attrs.marker_name == VERSION_MARKER_NAME: + return None + return Expectation(marker_decision_id(attrs.marker_name, context_id), {}) + + +@to_expectation.register +def _(attrs: history.MarkerRecordedEventAttributes) -> Expectation | None: + context_id = marker_context_id(attrs) + if context_id is None: + return None + if attrs.marker_name == VERSION_MARKER_NAME: + return None + return Expectation(marker_decision_id(attrs.marker_name, context_id), {}) + + # Workflow Completion - Enforce complete vs failure. Maybe we should enforce the output data? @to_expectation.register def _(_: decision.CompleteWorkflowExecutionDecisionAttributes) -> Expectation: diff --git a/cadence/testing/_workflow_environment.py b/cadence/testing/_workflow_environment.py index 3eb6b724..3a11f7a9 100644 --- a/cadence/testing/_workflow_environment.py +++ b/cadence/testing/_workflow_environment.py @@ -217,6 +217,13 @@ async def wait_condition(self, predicate: Callable[[], bool]) -> None: loop = cast(DeterministicEventLoop, get_running_loop()) await loop.create_waiter(predicate) + def side_effect( + self, + fn: Callable[[], ResultType], + result_type: Type[ResultType], + ) -> ResultType: + return fn() + async def signal_child_workflow( self, child_workflow_id: str, diff --git a/cadence/workflow.py b/cadence/workflow.py index 631b0f90..0581c59c 100644 --- a/cadence/workflow.py +++ b/cadence/workflow.py @@ -188,6 +188,17 @@ async def wait_condition(predicate: Callable[[], bool]) -> None: await WorkflowContext.get().wait_condition(predicate) +def side_effect( + fn: Callable[[], ResultType], + result_type: Type[ResultType], +) -> ResultType: + """Execute non-deterministic code and record the result as a SideEffect marker. + + On replay the function is not called; the value from workflow history is returned. + """ + return WorkflowContext.get().side_effect(fn, result_type) + + def is_cancel_requested() -> bool: return WorkflowContext.get().is_cancel_requested() @@ -607,6 +618,13 @@ async def start_timer(self, duration: timedelta) -> None: ... @abstractmethod async def wait_condition(self, predicate: Callable[[], bool]) -> None: ... + @abstractmethod + def side_effect( + self, + fn: Callable[[], ResultType], + result_type: Type[ResultType], + ) -> ResultType: ... + @abstractmethod def is_cancel_requested(self) -> bool: ... diff --git a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py index 6f21b5f3..14d395a7 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py +++ b/tests/cadence/_internal/workflow/statemachine/test_decision_manager.py @@ -4,11 +4,17 @@ import pytest from cadence.error import ChildWorkflowError, StartChildWorkflowExecutionFailed +from cadence._internal.workflow.statemachine.cancellation import CANCEL_MARKER_NAME from cadence._internal.workflow.statemachine.decision_manager import DecisionManager from cadence._internal.workflow.statemachine.event_dispatcher import ( EventDispatcher, resolve_id_attr, ) +from cadence._internal.workflow.statemachine.marker_state_machine import ( + encode_marker_header, + MARKER_HEADER_KEY, +) +from cadence._internal.workflow.statemachine.nondeterminism import NonDeterminismError from cadence._internal.workflow.statemachine.signal_external_workflow_state_machine import ( SignalExternalWorkflowFailed, ) @@ -215,6 +221,203 @@ async def test_collection_decisions_reordering(): assert activity2.done() is False +async def test_record_marker_collects_in_workflow_order(): + decisions = DecisionManager(asyncio.get_event_loop()) + marker_attrs = decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", details=Payload(data=b"marker") + ) + + decisions.start_timer(decision.StartTimerDecisionAttributes()) + recorded = decisions.record_marker(marker_attrs) + decisions.schedule_activity(decision.ScheduleActivityTaskDecisionAttributes()) + + assert recorded == Payload(data=b"marker") + assert decisions.collect_pending_decisions() == [ + decision.Decision( + start_timer_decision_attributes=decision.StartTimerDecisionAttributes( + timer_id="0" + ) + ), + decision.Decision(record_marker_decision_attributes=marker_attrs), + decision.Decision( + schedule_activity_task_decision_attributes=decision.ScheduleActivityTaskDecisionAttributes( + activity_id="2" + ) + ), + ] + + +async def test_record_marker_history_clears_pending_decision(): + decisions = DecisionManager(asyncio.get_event_loop()) + marker_attrs = decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", details=Payload(data=b"marker") + ) + + decisions.record_marker(marker_attrs) + + decisions.handle_history_event( + marker_recorded(1, "SideEffect", Payload(data=b"marker"), context_id="0") + ) + + assert decisions.collect_pending_decisions() == [] + + +async def test_replayed_marker_returns_recorded_details(): + decisions = DecisionManager(asyncio.get_event_loop()) + recorded = marker_recorded( + 1, "SideEffect", Payload(data=b"history-value"), context_id="0" + ) + + with decisions.track_nondeterminism(True, [recorded]): + result = decisions.record_marker( + decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", details=Payload(data=b"current-value") + ) + ) + + assert result == Payload(data=b"history-value") + + +async def test_replayed_marker_name_mismatch_is_nondeterministic(): + decisions = DecisionManager(asyncio.get_event_loop()) + recorded = marker_recorded( + 1, "SideEffect", Payload(data=b"history-value"), context_id="0" + ) + + with pytest.raises(NonDeterminismError): + with decisions.track_nondeterminism(True, [recorded]): + decisions.record_marker( + decision.RecordMarkerDecisionAttributes(marker_name="LocalActivity") + ) + + +async def test_replayed_marker_order_mismatch_is_nondeterministic(): + decisions = DecisionManager(asyncio.get_event_loop()) + first = marker_recorded(1, "SideEffect", Payload(data=b"first"), context_id="0") + second = marker_recorded( + 2, "LocalActivity", Payload(data=b"second"), context_id="1" + ) + + with pytest.raises(NonDeterminismError): + with decisions.track_nondeterminism(True, [first, second]): + # Auto-assigned ID is "0"; history expects "SideEffect" at that slot, + # so recording "LocalActivity" first is a non-determinism violation. + decisions.record_marker( + decision.RecordMarkerDecisionAttributes(marker_name="LocalActivity") + ) + + +async def test_replayed_marker_missing_from_workflow_is_nondeterministic(): + decisions = DecisionManager(asyncio.get_event_loop()) + recorded = marker_recorded( + 1, "SideEffect", Payload(data=b"history-value"), context_id="0" + ) + + with pytest.raises(NonDeterminismError): + with decisions.track_nondeterminism(True, [recorded]): + pass + + +async def test_replayed_marker_without_context_id_is_nondeterministic(): + decisions = DecisionManager(asyncio.get_event_loop()) + # Raw details with no length-prefix encoding — treated as produced by another SDK. + recorded = marker_recorded(1, "SideEffect", Payload(data=b"history-value")) + + with pytest.raises(NonDeterminismError): + with decisions.track_nondeterminism(True, [recorded]): + decisions.record_marker( + decision.RecordMarkerDecisionAttributes(marker_name="SideEffect") + ) + + +async def test_record_marker_ids_are_sequential(): + decisions = DecisionManager(asyncio.get_event_loop()) + + decisions.record_marker( + decision.RecordMarkerDecisionAttributes(marker_name="SideEffect") + ) + decisions.record_marker( + decision.RecordMarkerDecisionAttributes(marker_name="LocalActivity") + ) + + keys = list(decisions.state_machines.keys()) + assert keys[0].id == "SideEffect_0" + assert keys[1].id == "LocalActivity_1" + + +async def test_record_marker_requires_marker_name(): + decisions = DecisionManager(asyncio.get_event_loop()) + + with pytest.raises(ValueError, match="marker_name is required"): + decisions.record_marker(decision.RecordMarkerDecisionAttributes()) + + assert decisions.collect_pending_decisions() == [] + + +async def test_cancel_marker_is_not_logged_as_unknown_marker_name(caplog): + # The immediate-cancellation marker (cancellation.py) is Python-specific and encodes + # a JSON object in Details, not a [context_id, user_data] pair. It must not be mistaken + # for an unrecognized marker_name and warned about. + decisions = DecisionManager(asyncio.get_event_loop()) + cancel_marker = history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( + marker_name=CANCEL_MARKER_NAME, + details=Payload(data=b'{"canceled": true, "id": "0", "type": "ACTIVITY"}'), + ), + ) + + with caplog.at_level("WARNING"): + decisions.handle_history_event(cancel_marker) + + assert "unknown marker_name" not in caplog.text + + +async def test_version_marker_is_not_flagged_as_nondeterministic(): + # Version markers are exempt from non-determinism tracking. + # History that includes a Version marker must replay without error, and the + # Version expectation is never added so complete_replay doesn't fail either. + decisions = DecisionManager(asyncio.get_event_loop()) + # History: Version_0 (exempt) then SideEffect_1 (tracked). + version_recorded = marker_recorded( + 1, "Version", Payload(data=b"v1"), context_id="0" + ) + side_effect_recorded = marker_recorded( + 2, "SideEffect", Payload(data=b"side"), context_id="1" + ) + + with decisions.track_nondeterminism(True, [version_recorded, side_effect_recorded]): + # counter="0" → key "Version_0"; no expectation created or consumed. + decisions.record_marker( + decision.RecordMarkerDecisionAttributes( + marker_name="Version", details=Payload(data=b"v1") + ) + ) + # counter="1" → key "SideEffect_1"; expectation from history is consumed. + decisions.record_marker( + decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", details=Payload(data=b"new") + ) + ) + + +async def test_replayed_marker_details_come_from_history_not_current_value(): + # Ensure the historical value is returned even when the current call passes different data. + decisions = DecisionManager(asyncio.get_event_loop()) + recorded = marker_recorded( + 1, "SideEffect", Payload(data=b"history-value"), context_id="0" + ) + + with decisions.track_nondeterminism(True, [recorded]): + result = decisions.record_marker( + decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", details=Payload(data=b"current-value") + ) + ) + + assert result == Payload(data=b"history-value") + + async def test_child_workflow_dispatch(): decisions = DecisionManager(asyncio.get_event_loop()) @@ -381,6 +584,8 @@ class Handler: def handle(self, _: history.TimerStartedEventAttributes) -> None: pass + _ = Handler # the decorator's side effect (registering the handler) is what's under test + action = dispatcher.handlers[history.TimerStartedEventAttributes] assert action.id_attr == "" @@ -448,6 +653,23 @@ def activity_completed( ) +def marker_recorded( + event_id: int, marker_name: str, details: Payload, context_id: str | None = None +) -> history.HistoryEvent: + attrs = history.MarkerRecordedEventAttributes( + marker_name=marker_name, + details=details, + ) + if context_id is not None: + attrs.header.fields[MARKER_HEADER_KEY].CopyFrom( + encode_marker_header(context_id) + ) + return history.HistoryEvent( + event_id=event_id, + marker_recorded_event_attributes=attrs, + ) + + async def test_signal_external_workflow_creates_machine(): decisions = DecisionManager(asyncio.get_event_loop()) diff --git a/tests/cadence/_internal/workflow/statemachine/test_marker_state_machine.py b/tests/cadence/_internal/workflow/statemachine/test_marker_state_machine.py new file mode 100644 index 00000000..1119f4f8 --- /dev/null +++ b/tests/cadence/_internal/workflow/statemachine/test_marker_state_machine.py @@ -0,0 +1,122 @@ +from cadence._internal.workflow.statemachine.decision_state_machine import ( + DecisionId, + DecisionType, +) +from cadence._internal.workflow.statemachine.marker_state_machine import ( + MarkerStateMachine, + encode_marker_header, + marker_context_id, + marker_decision_id, + MARKER_HEADER_KEY, + SIDE_EFFECT_MARKER_NAME, +) +from cadence.api.v1 import decision, history +from cadence.api.v1.common_pb2 import Header, Payload + + +async def test_marker_state_machine_requested(): + attrs = decision.RecordMarkerDecisionAttributes(marker_name=SIDE_EFFECT_MARKER_NAME) + machine = MarkerStateMachine(attrs, SIDE_EFFECT_MARKER_NAME, "0") + + assert machine.get_decision() == decision.Decision( + record_marker_decision_attributes=attrs + ) + + +async def test_marker_state_machine_recorded(): + attrs = decision.RecordMarkerDecisionAttributes(marker_name=SIDE_EFFECT_MARKER_NAME) + machine = MarkerStateMachine(attrs, SIDE_EFFECT_MARKER_NAME, "0") + + machine.handle_recorded( + history.MarkerRecordedEventAttributes( + marker_name=SIDE_EFFECT_MARKER_NAME, + details=Payload(data=b"recorded"), + ) + ) + + assert machine.get_decision() is None + + +async def test_marker_state_machine_not_cancellable(): + attrs = decision.RecordMarkerDecisionAttributes(marker_name=SIDE_EFFECT_MARKER_NAME) + machine = MarkerStateMachine(attrs, SIDE_EFFECT_MARKER_NAME, "0") + + assert machine.request_cancel() is False + assert machine.get_decision() == decision.Decision( + record_marker_decision_attributes=attrs + ) + + +async def test_marker_state_machine_preserves_details_and_header(): + attrs = decision.RecordMarkerDecisionAttributes( + marker_name=SIDE_EFFECT_MARKER_NAME, + details=Payload(data=b"payload"), + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("0")}), + ) + machine = MarkerStateMachine(attrs, SIDE_EFFECT_MARKER_NAME, "0") + + emitted = machine.get_decision() + + assert emitted == decision.Decision(record_marker_decision_attributes=attrs) + # Details stay the raw user payload; the context_id lives in the header. + assert emitted.record_marker_decision_attributes.details == Payload(data=b"payload") + assert marker_context_id(emitted.record_marker_decision_attributes) == "0" + + +async def test_marker_state_machine_id_uses_type_context_format(): + attrs = decision.RecordMarkerDecisionAttributes(marker_name=SIDE_EFFECT_MARKER_NAME) + machine = MarkerStateMachine(attrs, SIDE_EFFECT_MARKER_NAME, "42") + + assert machine.get_id() == marker_decision_id(SIDE_EFFECT_MARKER_NAME, "42") + assert machine.get_id().id == "SideEffect_42" + + +async def test_marker_decision_id_format(): + # This format is the single source of truth for marker DecisionIds — every lookup + # (routing, the replay details cache, non-determinism expectations) must derive its + # key from marker_decision_id rather than reconstructing the string inline. + assert marker_decision_id(SIDE_EFFECT_MARKER_NAME, "0") == DecisionId( + DecisionType.MARKER, "SideEffect_0" + ) + + +async def test_marker_context_id_roundtrip(): + attrs = history.MarkerRecordedEventAttributes( + marker_name=SIDE_EFFECT_MARKER_NAME, + details=Payload(data=b"hello world"), + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header("my-id")}), + ) + + assert marker_context_id(attrs) == "my-id" + # The payload is untouched by the metadata encoding. + assert attrs.details == Payload(data=b"hello world") + + +async def test_marker_context_id_absent_header_returns_none(): + # A marker with no marker header — e.g. from another SDK or pre-header history. + attrs = history.MarkerRecordedEventAttributes( + marker_name=SIDE_EFFECT_MARKER_NAME, + details=Payload(data=b"raw-history-value"), + ) + + assert marker_context_id(attrs) is None + + +async def test_marker_context_id_ignores_unrelated_header_keys(): + # A header that carries other keys but not ours is treated as having no context_id. + attrs = history.MarkerRecordedEventAttributes( + marker_name=SIDE_EFFECT_MARKER_NAME, + header=Header(fields={"SomethingElse": Payload(data=b"value")}), + ) + + assert marker_context_id(attrs) is None + + +async def test_marker_context_id_malformed_header_returns_none(): + # Garbage under our header key must be treated as absent rather than raising. + attrs = history.MarkerRecordedEventAttributes( + marker_name=SIDE_EFFECT_MARKER_NAME, + header=Header(fields={MARKER_HEADER_KEY: Payload(data=b"not-json")}), + ) + + assert marker_context_id(attrs) is None diff --git a/tests/cadence/_internal/workflow/statemachine/test_nondeterminism.py b/tests/cadence/_internal/workflow/statemachine/test_nondeterminism.py index ed04685b..970cfef6 100644 --- a/tests/cadence/_internal/workflow/statemachine/test_nondeterminism.py +++ b/tests/cadence/_internal/workflow/statemachine/test_nondeterminism.py @@ -2,7 +2,10 @@ from typing import Any -from cadence._internal.workflow.statemachine.cancellation import to_marker +from cadence._internal.workflow.statemachine.cancellation import ( + CANCEL_MARKER_NAME, + to_marker, +) from cadence._internal.workflow.statemachine.completion_state_machine import ( COMPLETION_ID, ) @@ -10,6 +13,10 @@ DecisionId, DecisionType, ) +from cadence._internal.workflow.statemachine.marker_state_machine import ( + encode_marker_header, + MARKER_HEADER_KEY, +) from cadence._internal.workflow.statemachine.nondeterminism import ( to_expectation, Expectation, @@ -20,6 +27,10 @@ from cadence.api.v1 import common, decision, history +def _marker_header(context_id: str) -> common.Header: + return common.Header(fields={MARKER_HEADER_KEY: encode_marker_header(context_id)}) + + class TestDeterminismTracker: def test_single_expectation_met(self): tracker = DeterminismTracker() @@ -73,7 +84,7 @@ def test_cancel_marker_met(self): tracker.add_expectation( history.HistoryEvent( marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( - marker_name="Cancel_0", + marker_name=CANCEL_MARKER_NAME, details=to_marker( DecisionId(DecisionType.ACTIVITY, "0"), {"activity_type": "act"} ).details, @@ -102,7 +113,7 @@ def test_cancel_schedule_reorder_allowed(self): history.HistoryEvent( event_id=2, marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( - marker_name="Cancel_0", + marker_name=CANCEL_MARKER_NAME, details=to_marker( DecisionId(DecisionType.ACTIVITY, "0"), {"activity_type": "act"} ).details, @@ -129,7 +140,7 @@ def test_cancel_expected_nondeterminism(self): tracker.add_expectation( history.HistoryEvent( marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( - marker_name="Cancel_0", + marker_name=CANCEL_MARKER_NAME, details=to_marker( DecisionId(DecisionType.ACTIVITY, "0"), {"activity_type": "act"} ).details, @@ -163,7 +174,7 @@ def test_cancel_early_nondeterminism(self): history.HistoryEvent( event_id=2, marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( - marker_name="Cancel_0", + marker_name=CANCEL_MARKER_NAME, details=to_marker( DecisionId(DecisionType.ACTIVITY, "0"), {"activity_type": "act"} ).details, @@ -191,7 +202,7 @@ def test_cancel_props_changed_nondeterminism(self): tracker.add_expectation( history.HistoryEvent( marker_recorded_event_attributes=history.MarkerRecordedEventAttributes( - marker_name="Cancel_0", + marker_name=CANCEL_MARKER_NAME, details=to_marker( DecisionId(DecisionType.ACTIVITY, "0"), {"activity_type": "act"} ).details, @@ -339,6 +350,35 @@ def test_expectations_out_of_order_nondeterminism(self): DecisionId(DecisionType.ACTIVITY, "1"), {"activity_type": "act"} ) + def test_marker_expectation_does_not_carry_details(self): + # Details are handled by DecisionManager._recorded_marker_details, not Expectation. + tracker = DeterminismTracker() + recorded = history.MarkerRecordedEventAttributes( + marker_name="SideEffect", + details=common.Payload(data=b"history-value"), + header=_marker_header("0"), + ) + requested = decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", + details=common.Payload(data=b"history-value"), + header=_marker_header("0"), + ) + tracker.add_expectation( + history.HistoryEvent( + event_id=1, + marker_recorded_event_attributes=recorded, + ) + ) + + expectation = tracker.validate_action(requested) + + assert expectation == Expectation( + DecisionId(DecisionType.MARKER, "SideEffect_0"), + {}, + ) + assert "details" not in expectation.properties + tracker.complete_replay() + @pytest.mark.parametrize( "attrs,expected", @@ -481,6 +521,53 @@ def test_expectations_out_of_order_nondeterminism(self): ), # Unknown type returns None ("not_a_supported_type", None), + # Marker: SideEffect decision-side + ( + decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", + header=_marker_header("0"), + ), + Expectation( + DecisionId(DecisionType.MARKER, "SideEffect_0"), + {}, + ), + ), + # Marker: SideEffect history-side (no details in properties) + ( + history.MarkerRecordedEventAttributes( + marker_name="SideEffect", + details=common.Payload(data=b"value"), + header=_marker_header("0"), + ), + Expectation( + DecisionId(DecisionType.MARKER, "SideEffect_0"), + {}, + ), + ), + # Marker: Version decision-side → None (exempt) + ( + decision.RecordMarkerDecisionAttributes( + marker_name="Version", + header=_marker_header("0"), + ), + None, + ), + # Marker: Version history-side → None (exempt) + ( + history.MarkerRecordedEventAttributes( + marker_name="Version", + header=_marker_header("0"), + ), + None, + ), + # Marker: no marker header → None + ( + decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", + details=common.Payload(data=b"raw-no-encoding"), + ), + None, + ), ], ) def test_to_expectation(attrs: Any, expected: Expectation): diff --git a/tests/cadence/_internal/workflow/test_context_side_effect.py b/tests/cadence/_internal/workflow/test_context_side_effect.py new file mode 100644 index 00000000..b2d7920d --- /dev/null +++ b/tests/cadence/_internal/workflow/test_context_side_effect.py @@ -0,0 +1,131 @@ +from unittest.mock import MagicMock + +import pytest + +from cadence._internal.workflow.context import Context +from cadence.api.v1.common_pb2 import Payload +from cadence.data_converter import DefaultDataConverter +from cadence.workflow import WorkflowInfo +import cadence.workflow as workflow_module + + +def _make_ctx(*, replay: bool = False) -> tuple[Context, MagicMock]: + dm = MagicMock() + dc = DefaultDataConverter() + info = WorkflowInfo( + workflow_type="Wf", + workflow_domain="domain", + workflow_id="wid", + workflow_run_id="rid", + workflow_task_list="tl", + data_converter=dc, + ) + ctx = Context(info, dm) + ctx.set_replay_mode(replay) + return ctx, dm + + +def _dc() -> DefaultDataConverter: + return DefaultDataConverter() + + +def test_side_effect_calls_fn_and_returns_result(): + ctx, dm = _make_ctx() + dc = _dc() + dm.record_marker.return_value = dc.to_data([42]) + + result = ctx.side_effect(lambda: 42, int) + + assert result == 42 + dm.record_marker.assert_called_once() + attrs = dm.record_marker.call_args[0][0] + assert attrs.marker_name == "SideEffect" + assert attrs.details == dc.to_data([42]) + + +def test_side_effect_fn_is_called_on_first_run(): + ctx, dm = _make_ctx() + dc = _dc() + dm.record_marker.return_value = dc.to_data([0]) + + calls: list[int] = [] + + def record_call() -> int: + calls.append(1) + return 1 + + ctx.side_effect(record_call, int) + + assert len(calls) == 1 + + +def test_side_effect_fn_is_skipped_on_replay(): + ctx, dm = _make_ctx(replay=True) + dc = _dc() + dm.record_marker.return_value = dc.to_data(["history-value"]) + + calls: list[int] = [] + + def record_call() -> str: + calls.append(1) + return "current-value" + + result = ctx.side_effect(record_call, str) + + assert len(calls) == 0 + assert result == "history-value" + + +def test_side_effect_on_replay_returns_history_value(): + ctx, dm = _make_ctx(replay=True) + dc = _dc() + dm.record_marker.return_value = dc.to_data(["history-value"]) + + result = ctx.side_effect(lambda: "current-value", str) + + assert result == "history-value" + + +def test_side_effect_on_first_run_passes_fn_result_as_details(): + ctx, dm = _make_ctx() + dc = _dc() + dm.record_marker.return_value = dc.to_data(["x"]) + + ctx.side_effect(lambda: "x", str) + + attrs = dm.record_marker.call_args[0][0] + assert attrs.details == dc.to_data(["x"]) + + +def test_side_effect_on_replay_passes_empty_details(): + ctx, dm = _make_ctx(replay=True) + dc = _dc() + dm.record_marker.return_value = dc.to_data(["history"]) + + ctx.side_effect(lambda: "current", str) + + attrs = dm.record_marker.call_args[0][0] + assert attrs.details == Payload() + + +def _raises_value_error() -> int: + raise ValueError("boom") + + +def test_side_effect_raises_if_fn_raises(): + ctx, _ = _make_ctx() + + with pytest.raises(ValueError, match="boom"): + ctx.side_effect(_raises_value_error, int) + + +def test_side_effect_module_level_dispatches_through_context(): + ctx, dm = _make_ctx() + dc = _dc() + dm.record_marker.return_value = dc.to_data([7]) + + with ctx._activate(): + result = workflow_module.side_effect(lambda: 7, int) + + assert result == 7 + dm.record_marker.assert_called_once() diff --git a/tests/cadence/_internal/workflow/test_marker_integration.py b/tests/cadence/_internal/workflow/test_marker_integration.py new file mode 100644 index 00000000..a8c70ea4 --- /dev/null +++ b/tests/cadence/_internal/workflow/test_marker_integration.py @@ -0,0 +1,220 @@ +import asyncio + +from cadence._internal.workflow.decision_events_iterator import DecisionEventsIterator +from cadence._internal.workflow.statemachine.decision_manager import DecisionManager +from cadence._internal.workflow.statemachine.marker_state_machine import ( + encode_marker_header, + MARKER_HEADER_KEY, +) +from cadence.api.v1 import decision, history +from cadence.api.v1.common_pb2 import Header, Payload + + +async def test_replay_marker_event_is_preloaded_before_marker_decision_exists(): + decisions = DecisionManager(asyncio.get_event_loop()) + marker_attrs = decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", + details=Payload(data=b"recorded-value"), + ) + decision_events = next(iter(DecisionEventsIterator(_history_with_marker_output()))) + + assert decision_events.replay is True + assert len(decision_events.markers) == 1 + + for marker_event in decision_events.markers: + decisions.handle_history_event(marker_event) + + decisions.record_marker(marker_attrs) + assert decisions.collect_pending_decisions() == [ + decision.Decision(record_marker_decision_attributes=marker_attrs) + ] + + for event in decision_events.output: + decisions.handle_history_event(event) + + assert decisions.collect_pending_decisions() == [] + + +async def test_current_decision_task_emits_record_marker_decision(): + decisions = DecisionManager(asyncio.get_event_loop()) + decision_events = next(iter(DecisionEventsIterator(_current_decision_history()))) + marker_attrs = decision.RecordMarkerDecisionAttributes( + marker_name="MutableSideEffect", + details=Payload(data=b"new-value"), + ) + + assert decision_events.replay is False + assert decision_events.output == [] + assert decision_events.markers == [] + + decisions.record_marker(marker_attrs) + + assert decisions.collect_pending_decisions() == [ + decision.Decision(record_marker_decision_attributes=marker_attrs) + ] + + +async def test_multiple_replayed_marker_outputs_complete_in_decision_order(): + decisions = DecisionManager(asyncio.get_event_loop()) + first_attrs = decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", + details=Payload(data=b"first"), + ) + second_attrs = decision.RecordMarkerDecisionAttributes( + marker_name="LocalActivity", + details=Payload(data=b"second"), + ) + decision_events = next( + iter(DecisionEventsIterator(_history_with_multiple_marker_outputs())) + ) + + assert decision_events.replay is True + assert len(decision_events.markers) == 2 + + for marker_event in decision_events.markers: + decisions.handle_history_event(marker_event) + + decisions.record_marker(first_attrs) + decisions.record_marker(second_attrs) + assert decisions.collect_pending_decisions() == [ + decision.Decision(record_marker_decision_attributes=first_attrs), + decision.Decision(record_marker_decision_attributes=second_attrs), + ] + + for event in decision_events.output: + decisions.handle_history_event(event) + + assert decisions.collect_pending_decisions() == [] + + +async def test_version_marker_added_on_replay_is_not_nondeterministic(): + # Version markers are exempt from non-determinism tracking (same as Go SDK). + # A history containing a Version marker followed by a SideEffect must replay without error. + decisions = DecisionManager(asyncio.get_event_loop()) + decision_events = next( + iter(DecisionEventsIterator(_history_with_version_and_side_effect())) + ) + + assert decision_events.replay is True + + for marker_event in decision_events.markers: + decisions.handle_history_event(marker_event) + + with decisions.track_nondeterminism(decision_events.replay, decision_events.output): + # counter="0" → "Version_0": no expectation created or consumed. + decisions.record_marker( + decision.RecordMarkerDecisionAttributes( + marker_name="Version", details=Payload(data=b"v1") + ) + ) + # counter="1" → "SideEffect_1": expectation from history is consumed. + decisions.record_marker( + decision.RecordMarkerDecisionAttributes( + marker_name="SideEffect", details=Payload(data=b"new") + ) + ) + + +def _history_with_version_and_side_effect() -> list[history.HistoryEvent]: + return [ + _workflow_started(1), + _decision_task_scheduled(2), + _decision_task_started(3, scheduled_event_id=2), + _decision_task_completed(4, scheduled_event_id=2, started_event_id=3), + _marker_recorded(5, "Version", Payload(data=b"v1"), "0"), + _marker_recorded(6, "SideEffect", Payload(data=b"recorded"), "1"), + _decision_task_scheduled(7), + _decision_task_started(8, scheduled_event_id=7), + ] + + +def _current_decision_history() -> list[history.HistoryEvent]: + return [ + _workflow_started(1), + _decision_task_scheduled(2), + _decision_task_started(3, scheduled_event_id=2), + ] + + +def _history_with_marker_output() -> list[history.HistoryEvent]: + return [ + _workflow_started(1), + _decision_task_scheduled(2), + _decision_task_started(3, scheduled_event_id=2), + _decision_task_completed(4, scheduled_event_id=2, started_event_id=3), + _marker_recorded(5, "SideEffect", Payload(data=b"recorded-value"), "0"), + _decision_task_scheduled(6), + _decision_task_started(7, scheduled_event_id=6), + ] + + +def _history_with_multiple_marker_outputs() -> list[history.HistoryEvent]: + return [ + _workflow_started(1), + _decision_task_scheduled(2), + _decision_task_started(3, scheduled_event_id=2), + _decision_task_completed(4, scheduled_event_id=2, started_event_id=3), + _marker_recorded(5, "SideEffect", Payload(data=b"first"), "0"), + _marker_recorded(6, "LocalActivity", Payload(data=b"second"), "1"), + _decision_task_scheduled(7), + _decision_task_started(8, scheduled_event_id=7), + ] + + +def _workflow_started(event_id: int) -> history.HistoryEvent: + return history.HistoryEvent( + event_id=event_id, + workflow_execution_started_event_attributes=history.WorkflowExecutionStartedEventAttributes(), + ) + + +def _decision_task_scheduled(event_id: int) -> history.HistoryEvent: + return history.HistoryEvent( + event_id=event_id, + decision_task_scheduled_event_attributes=history.DecisionTaskScheduledEventAttributes(), + ) + + +def _decision_task_started( + event_id: int, + *, + scheduled_event_id: int, +) -> history.HistoryEvent: + return history.HistoryEvent( + event_id=event_id, + decision_task_started_event_attributes=history.DecisionTaskStartedEventAttributes( + scheduled_event_id=scheduled_event_id, + ), + ) + + +def _decision_task_completed( + event_id: int, + *, + scheduled_event_id: int, + started_event_id: int, +) -> history.HistoryEvent: + return history.HistoryEvent( + event_id=event_id, + decision_task_completed_event_attributes=history.DecisionTaskCompletedEventAttributes( + scheduled_event_id=scheduled_event_id, + started_event_id=started_event_id, + ), + ) + + +def _marker_recorded( + event_id: int, + marker_name: str, + details: Payload, + context_id: str, +) -> history.HistoryEvent: + attrs = history.MarkerRecordedEventAttributes( + marker_name=marker_name, + details=details, + header=Header(fields={MARKER_HEADER_KEY: encode_marker_header(context_id)}), + ) + return history.HistoryEvent( + event_id=event_id, + marker_recorded_event_attributes=attrs, + )