feat: Add first_run_at support to StartWorkflowOptions (cadence-workflow#69)

<!-- Describe what has changed in this PR -->
**What changed?**
- Add first_run_at datetime field to StartWorkflowOptions for precise workflow start time control

<!-- Tell your future self why have you made these changes -->
**Why?**
This PR gives python client feature parity on precise workflow start
time control

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
unit test

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

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
@@ -40,6 +41,7 @@ class StartWorkflowOptions(TypedDict, total=False):
     delay_start: timedelta
     jitter_start: timedelta
     cron_overlap_policy: workflow_pb2.CronOverlapPolicy
+    first_run_at: datetime
 
 
 def _validate_and_apply_defaults(options: StartWorkflowOptions) -> StartWorkflowOptions:
@@ -70,6 +72,21 @@ def _validate_and_apply_defaults(options: StartWorkflowOptions) -> StartWorkflow
     if jitter_start is not None and jitter_start < timedelta(0):
         raise ValueError("jitter_start cannot be negative")
 
+    # 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
+        if first_run_at.timestamp() < 0:
+            raise ValueError(
+                "first_run_at cannot be before Unix epoch (January 1, 1970 UTC)"
+            )
+
     return options
 
 
@@ -216,6 +233,13 @@ def _build_start_workflow_request(
             jitter_duration.FromTimedelta(jitter_start)
             request.jitter_start.CopyFrom(jitter_duration)
 
+        # 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
@@ -345,6 +345,76 @@ async def test_build_request_with_cron_overlap_policy_enum(self, mock_client):
 
         assert request.cron_overlap_policy == workflow_pb2.CRON_OVERLAP_POLICY_SKIPPED
 
+    @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."""