feat: Add first_run_at support to StartWorkflowOptions

Add explicit first run timestamp for cron workflows:
- Add first_run_at datetime field to StartWorkflowOptions
- Add warning for timezone-naive datetimes (treated as UTC)
- Convert datetime to protobuf Timestamp in request building
- Add comprehensive tests for aware/naive datetime handling

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,13 @@
 import os
 import socket
 import uuid
-from datetime import timedelta
+import warnings
+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 +38,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 +59,15 @@ 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")
 
+    # Warn if first_run_at is timezone-naive
+    first_run_at = options.get("first_run_at")
+    if first_run_at is not None and first_run_at.tzinfo is None:
+        warnings.warn(
+            "first_run_at is timezone-naive; it will be treated as UTC",
+            UserWarning,
+            stacklevel=3,
+        )
+
     return options
 
 
@@ -186,6 +198,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,7 @@
 import pytest
 import uuid
-from datetime import timedelta
+import warnings
+from datetime import timedelta, datetime, timezone
 from unittest.mock import AsyncMock, Mock, PropertyMock
 
 from cadence.api.v1.common_pb2 import WorkflowExecution
@@ -241,6 +242,75 @@ 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
+
+    @pytest.mark.asyncio
+    async def test_build_request_with_first_run_at_naive(self, mock_client):
+        """Test building request with timezone-naive first_run_at."""
+        client = Client(domain="test-domain", target="localhost:7933")
+
+        first_run = datetime(2024, 6, 1, 12, 0, 0)  # Naive datetime
+
+        options = StartWorkflowOptions(
+            task_list="test-task-list",
+            execution_start_to_close_timeout=timedelta(minutes=10),
+            task_start_to_close_timeout=timedelta(seconds=30),
+            first_run_at=first_run,
+        )
+
+        request = client._build_start_workflow_request("TestWorkflow", (), options)
+
+        assert request.HasField("first_run_at")
+
+    def test_first_run_at_naive_datetime_warns(self):
+        """Test that timezone-naive first_run_at produces warning."""
+        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.warns(UserWarning, match="timezone-naive"):
+            _validate_and_apply_defaults(options)
+
+    def test_first_run_at_aware_datetime_no_warning(self):
+        """Test that timezone-aware first_run_at produces no warning."""
+        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 warn - use warnings.catch_warnings to verify
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            _validate_and_apply_defaults(options)
+            # Filter for our specific warning
+            relevant_warnings = [
+                warning for warning in w if "timezone-naive" in str(warning.message)
+            ]
+            assert len(relevant_warnings) == 0
+
 
 class TestClientStartWorkflow:
     """Test Client.start_workflow method."""