Add first_run_at support to StartWorkflowOptions

- Add first_run_at datetime field for explicit cron workflow scheduling
- Require timezone-aware datetime to prevent local/UTC ambiguity bugs
- Validate timestamp is not before Unix epoch (matching Go client)
- Add comprehensive tests for timezone handling and validation

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

Author: timl3136

cadence/client.py

@@ -1,11 +1,12 @@
 import os
 import socket
 import uuid
-from datetime import timedelta
+from datetime import datetime, timedelta
 from typing import TypedDict, Unpack, Any, cast, Union
 
 from grpc import ChannelCredentials, Compression
 from google.protobuf.duration_pb2 import Duration
+from google.protobuf.timestamp_pb2 import Timestamp
 
 from cadence._internal.rpc.error import CadenceErrorInterceptor
 from cadence._internal.rpc.retry import RetryInterceptor
@@ -36,6 +37,7 @@ class StartWorkflowOptions(TypedDict, total=False):
     workflow_id: str
     task_start_to_close_timeout: timedelta
     cron_schedule: str
+    first_run_at: datetime
 
 
 def _validate_and_apply_defaults(options: StartWorkflowOptions) -> StartWorkflowOptions:
@@ -56,6 +58,21 @@ 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 first_run_at (must be timezone-aware and not before Unix epoch)
+    first_run_at = options.get("first_run_at")
+    if first_run_at is not None:
+        # Require timezone-aware datetime to prevent ambiguity
+        if first_run_at.tzinfo is None:
+            raise ValueError(
+                "first_run_at must be timezone-aware. "
+                "Use datetime.now(timezone.utc) or datetime(..., tzinfo=timezone.utc)"
+            )
+        # Validate timestamp is not before Unix epoch (matching Go client behavior)
+        if first_run_at.timestamp() < 0:
+            raise ValueError(
+                "first_run_at cannot be before Unix epoch (January 1, 1970 UTC)"
+            )
+
     return options
 
 
@@ -186,6 +203,13 @@ def _build_start_workflow_request(
         if options.get("cron_schedule"):
             request.cron_schedule = options["cron_schedule"]
 
+        # Set first_run_at if provided
+        first_run_at = options.get("first_run_at")
+        if first_run_at is not None:
+            first_run_timestamp = Timestamp()
+            first_run_timestamp.FromDatetime(first_run_at)
+            request.first_run_at.CopyFrom(first_run_timestamp)
+
         return request
 
     async def start_workflow(

tests/cadence/test_client_workflow.py

@@ -1,6 +1,6 @@
 import pytest
 import uuid
-from datetime import timedelta
+from datetime import timedelta, datetime, timezone
 from unittest.mock import AsyncMock, Mock, PropertyMock
 
 from cadence.api.v1.common_pb2 import WorkflowExecution
@@ -241,6 +241,76 @@ 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_first_run_at_timezone_aware(self, mock_client):
+        """Test building request with timezone-aware first_run_at."""
+        client = Client(domain="test-domain", target="localhost:7933")
+
+        first_run = datetime(2024, 6, 1, 12, 0, 0, tzinfo=timezone.utc)
+
+        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 * * * *",
+            first_run_at=first_run,
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.HasField("first_run_at")
+        # Convert back and verify
+        result_dt = request.first_run_at.ToDatetime()
+        assert result_dt == first_run.replace(
+            tzinfo=None
+        )  # ToDatetime returns naive UTC
+
+    def test_first_run_at_naive_datetime_raises_error(self):
+        """Test that timezone-naive first_run_at raises ValueError."""
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            first_run_at=datetime(2024, 1, 1, 12, 0, 0),  # Naive
+        )
+        with pytest.raises(ValueError, match="must be timezone-aware"):
+            _validate_and_apply_defaults(options)
+
+    def test_first_run_at_aware_datetime_is_valid(self):
+        """Test that timezone-aware first_run_at is valid."""
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            first_run_at=datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
+        )
+        # Should not raise
+        validated = _validate_and_apply_defaults(options)
+        assert validated["first_run_at"] == datetime(
+            2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc
+        )
+
+    def test_first_run_at_before_epoch_raises_error(self):
+        """Test that first_run_at before Unix epoch raises ValueError."""
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            first_run_at=datetime(1960, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
+        )
+        with pytest.raises(ValueError, match="cannot be before Unix epoch"):
+            _validate_and_apply_defaults(options)
+
+    def test_first_run_at_at_epoch_is_valid(self):
+        """Test that first_run_at exactly at Unix epoch is valid."""
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            first_run_at=datetime(1970, 1, 1, 0, 0, 0, tzinfo=timezone.utc),
+        )
+        # Should not raise
+        validated = _validate_and_apply_defaults(options)
+        assert validated["first_run_at"] == datetime(
+            1970, 1, 1, 0, 0, 0, tzinfo=timezone.utc
+        )
+
 
 class TestClientStartWorkflow:
     """Test Client.start_workflow method."""