feat: Add delay_start and jitter_start support to StartWorkflowOptions

Add timing control fields for delaying workflow start:
- Add delay_start timedelta field to delay workflow execution
- Add jitter_start timedelta field for random start time jitter
- Add validation to ensure values are non-negative
- Add comprehensive tests for valid values and error cases

Signed-off-by: Tim Li <ltim@uber.com>

Author: timl3136

cadence/client.py

@@ -36,6 +36,8 @@ class StartWorkflowOptions(TypedDict, total=False):
     workflow_id: str
     task_start_to_close_timeout: timedelta
     cron_schedule: str
+    delay_start: timedelta
+    jitter_start: timedelta
 
 
 def _validate_and_apply_defaults(options: StartWorkflowOptions) -> StartWorkflowOptions:
@@ -56,6 +58,16 @@ def _validate_and_apply_defaults(options: StartWorkflowOptions) -> StartWorkflow
     elif task_timeout <= timedelta(0):
         raise ValueError("task_start_to_close_timeout must be greater than 0")
 
+    # Validate delay_start (must be non-negative)
+    delay_start = options.get("delay_start")
+    if delay_start is not None and delay_start < timedelta(0):
+        raise ValueError("delay_start cannot be negative")
+
+    # Validate jitter_start (must be non-negative)
+    jitter_start = options.get("jitter_start")
+    if jitter_start is not None and jitter_start < timedelta(0):
+        raise ValueError("jitter_start cannot be negative")
+
     return options
 
 
@@ -186,6 +198,20 @@ def _build_start_workflow_request(
         if options.get("cron_schedule"):
             request.cron_schedule = options["cron_schedule"]
 
+        # Set delay_start if provided
+        delay_start = options.get("delay_start")
+        if delay_start is not None:
+            delay_duration = Duration()
+            delay_duration.FromTimedelta(delay_start)
+            request.delay_start.CopyFrom(delay_duration)
+
+        # Set jitter_start if provided
+        jitter_start = options.get("jitter_start")
+        if jitter_start is not None:
+            jitter_duration = Duration()
+            jitter_duration.FromTimedelta(jitter_start)
+            request.jitter_start.CopyFrom(jitter_duration)
+
         return request
 
     async def start_workflow(

tests/cadence/test_client_workflow.py

@@ -241,6 +241,88 @@ 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_delay_start(self, mock_client):
+        """Test building request with delay_start."""
+        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),
+            delay_start=timedelta(minutes=5),
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.HasField("delay_start")
+        assert request.delay_start.seconds == 300  # 5 minutes
+
+    @pytest.mark.asyncio
+    async def test_build_request_with_jitter_start(self, mock_client):
+        """Test building request with jitter_start."""
+        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),
+            jitter_start=timedelta(seconds=30),
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.HasField("jitter_start")
+        assert request.jitter_start.seconds == 30
+
+    @pytest.mark.asyncio
+    async def test_build_request_with_delay_and_jitter_start(self, mock_client):
+        """Test building request with both delay_start and jitter_start."""
+        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),
+            delay_start=timedelta(minutes=1),
+            jitter_start=timedelta(seconds=10),
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.delay_start.seconds == 60
+        assert request.jitter_start.seconds == 10
+
+    def test_negative_delay_start_raises_error(self):
+        """Test that negative delay_start raises ValueError."""
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            delay_start=timedelta(seconds=-1),
+        )
+        with pytest.raises(ValueError, match="delay_start cannot be negative"):
+            _validate_and_apply_defaults(options)
+
+    def test_negative_jitter_start_raises_error(self):
+        """Test that negative jitter_start raises ValueError."""
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            jitter_start=timedelta(seconds=-1),
+        )
+        with pytest.raises(ValueError, match="jitter_start cannot be negative"):
+            _validate_and_apply_defaults(options)
+
+    def test_zero_delay_start_is_valid(self):
+        """Test that zero delay_start is valid."""
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            delay_start=timedelta(0),
+        )
+        validated = _validate_and_apply_defaults(options)
+        assert validated["delay_start"] == timedelta(0)
+
 
 class TestClientStartWorkflow:
     """Test Client.start_workflow method."""