feat: Add cron_overlap_policy support to StartWorkflowOptions

Add support for configuring cron overlap behavior when starting workflows:
- Add cron_overlap_policy field to StartWorkflowOptions
- Accept both protobuf enum values and string literals ("SKIPPED", "BUFFER_ONE")
- Add helper function _resolve_cron_overlap_policy for string-to-enum conversion
- Add comprehensive tests for enum, string, and error cases

Co-Authored-By: Claude (claude-opus-4-5) <noreply@anthropic.com>
Signed-off-by: Tim Li <ltim@uber.com>

Author: timl3136

cadence/client.py

@@ -2,7 +2,7 @@
 import socket
 import uuid
 from datetime import timedelta
-from typing import TypedDict, Unpack, Any, cast, Union
+from typing import TypedDict, Unpack, Any, cast, Union, Literal
 
 from grpc import ChannelCredentials, Compression
 from google.protobuf.duration_pb2 import Duration
@@ -22,11 +22,36 @@
     SignalWithStartWorkflowExecutionResponse,
 )
 from cadence.api.v1.common_pb2 import WorkflowType, WorkflowExecution
+from cadence.api.v1 import workflow_pb2
 from cadence.api.v1.tasklist_pb2 import TaskList
 from cadence.data_converter import DataConverter, DefaultDataConverter
 from cadence.metrics import MetricsEmitter, NoOpMetricsEmitter
 from cadence.workflow import WorkflowDefinition
 
+# Type alias for cron overlap policy - accepts protobuf enum or string
+CronOverlapPolicyType = (
+    workflow_pb2.CronOverlapPolicy | Literal["SKIPPED", "BUFFER_ONE"]
+)
+
+# Mapping from string to protobuf enum value
+_CRON_OVERLAP_POLICY_MAP: dict[str, int] = {
+    "SKIPPED": workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED,
+    "BUFFER_ONE": workflow_pb2.CRON_OVERLAP_POLICY_BUFFER_ONE,
+}
+
+
+def _resolve_cron_overlap_policy(policy: CronOverlapPolicyType) -> int:
+    """Convert string or enum to protobuf enum value."""
+    if isinstance(policy, str):
+        policy_upper = policy.upper()
+        if policy_upper not in _CRON_OVERLAP_POLICY_MAP:
+            raise ValueError(
+                f"Invalid cron_overlap_policy: '{policy}'. "
+                "Expected 'SKIPPED' or 'BUFFER_ONE'"
+            )
+        return _CRON_OVERLAP_POLICY_MAP[policy_upper]
+    return int(policy)
+
 
 class StartWorkflowOptions(TypedDict, total=False):
     """Options for starting a workflow execution."""
@@ -36,6 +61,7 @@ class StartWorkflowOptions(TypedDict, total=False):
     workflow_id: str
     task_start_to_close_timeout: timedelta
     cron_schedule: str
+    cron_overlap_policy: CronOverlapPolicyType
 
 
 def _validate_and_apply_defaults(options: StartWorkflowOptions) -> StartWorkflowOptions:
@@ -185,6 +211,10 @@ def _build_start_workflow_request(
             request.input.CopyFrom(input_payload)
         if options.get("cron_schedule"):
             request.cron_schedule = options["cron_schedule"]
+        if options.get("cron_overlap_policy") is not None:
+            request.cron_overlap_policy = _resolve_cron_overlap_policy(
+                options["cron_overlap_policy"]
+            )
 
         return request
 

tests/cadence/test_client_workflow.py

@@ -8,7 +8,13 @@
     StartWorkflowExecutionRequest,
     StartWorkflowExecutionResponse,
 )
-from cadence.client import Client, StartWorkflowOptions, _validate_and_apply_defaults
+from cadence.client import (
+    Client,
+    StartWorkflowOptions,
+    _validate_and_apply_defaults,
+    _resolve_cron_overlap_policy,
+)
+from cadence.api.v1 import workflow_pb2
 from cadence.data_converter import DefaultDataConverter
 from cadence.workflow import WorkflowDefinition, WorkflowDefinitionOptions
 
@@ -241,6 +247,111 @@ async def test_build_request_with_cron_schedule(self, mock_client):
 
         assert request.cron_schedule == "0 * * * *"
 
+    @pytest.mark.asyncio
+    async def test_build_request_with_cron_overlap_policy_enum(self, mock_client):
+        """Test building request with cron_overlap_policy as enum."""
+        client = Client(domain="test-domain", target="localhost:7933")
+
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            task_start_to_close_timeout=timedelta(seconds=30),
+            cron_schedule="0 * * * *",
+            cron_overlap_policy=workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED,
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.cron_overlap_policy == workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED
+
+    @pytest.mark.asyncio
+    async def test_build_request_with_cron_overlap_policy_string_skipped(
+        self, mock_client
+    ):
+        """Test building request with cron_overlap_policy as string 'SKIPPED'."""
+        client = Client(domain="test-domain", target="localhost:7933")
+
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            task_start_to_close_timeout=timedelta(seconds=30),
+            cron_schedule="0 * * * *",
+            cron_overlap_policy="SKIPPED",
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.cron_overlap_policy == workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED
+
+    @pytest.mark.asyncio
+    async def test_build_request_with_cron_overlap_policy_string_buffer_one(
+        self, mock_client
+    ):
+        """Test building request with cron_overlap_policy as string 'BUFFER_ONE'."""
+        client = Client(domain="test-domain", target="localhost:7933")
+
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            task_start_to_close_timeout=timedelta(seconds=30),
+            cron_schedule="0 * * * *",
+            cron_overlap_policy="BUFFER_ONE",
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert (
+            request.cron_overlap_policy == workflow_pb2.CRON_OVERLAP_POLICY_BUFFER_ONE
+        )
+
+    @pytest.mark.asyncio
+    async def test_build_request_with_cron_overlap_policy_string_lowercase(
+        self, mock_client
+    ):
+        """Test building request with cron_overlap_policy as lowercase string."""
+        client = Client(domain="test-domain", target="localhost:7933")
+
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            task_start_to_close_timeout=timedelta(seconds=30),
+            cron_schedule="0 * * * *",
+            cron_overlap_policy="skipped",
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.cron_overlap_policy == workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED
+
+
+class TestResolveCronOverlapPolicy:
+    """Test _resolve_cron_overlap_policy helper function."""
+
+    def test_resolve_enum_value(self):
+        """Test resolving protobuf enum value."""
+        result = _resolve_cron_overlap_policy(workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED)
+        assert result == workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED
+
+    def test_resolve_string_skipped(self):
+        """Test resolving string 'SKIPPED'."""
+        result = _resolve_cron_overlap_policy("SKIPPED")
+        assert result == workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED
+
+    def test_resolve_string_buffer_one(self):
+        """Test resolving string 'BUFFER_ONE'."""
+        result = _resolve_cron_overlap_policy("BUFFER_ONE")
+        assert result == workflow_pb2.CRON_OVERLAP_POLICY_BUFFER_ONE
+
+    def test_resolve_string_lowercase(self):
+        """Test resolving lowercase string."""
+        result = _resolve_cron_overlap_policy("buffer_one")
+        assert result == workflow_pb2.CRON_OVERLAP_POLICY_BUFFER_ONE
+
+    def test_resolve_invalid_string_raises_error(self):
+        """Test that invalid string raises ValueError."""
+        with pytest.raises(ValueError, match="Invalid cron_overlap_policy"):
+            _resolve_cron_overlap_policy("INVALID_POLICY")
+
 
 class TestClientStartWorkflow:
     """Test Client.start_workflow method."""