Add new workflow info field for workflow_start_time

temporalio/worker/_workflow.py

@@ -532,6 +532,9 @@ def _create_workflow_instance(
                 init.search_attributes
             ),
             start_time=act.timestamp.ToDatetime().replace(tzinfo=timezone.utc),
+            workflow_start_time=init.start_time.ToDatetime().replace(
+                tzinfo=timezone.utc
+            ),
             task_queue=self._task_queue,
             task_timeout=init.workflow_task_timeout.ToTimedelta(),
             typed_search_attributes=temporalio.converter.decode_typed_search_attributes(

temporalio/worker/workflow_sandbox/_runner.py

@@ -42,6 +42,7 @@
     run_timeout=None,
     search_attributes={},
     start_time=datetime.fromtimestamp(0, timezone.utc),
+    workflow_start_time=datetime.fromtimestamp(0, timezone.utc),
     task_queue="sandbox-validate-task_queue",
     task_timeout=timedelta(),
     typed_search_attributes=temporalio.common.TypedSearchAttributes.empty,

temporalio/workflow.py

@@ -508,6 +508,8 @@ class Info:
     """
 
     start_time: datetime
+    """The start time of the first task executed by the workflow."""
+
     task_queue: str
     task_timeout: timedelta
 
@@ -520,6 +522,10 @@ class Info:
     """
 
     workflow_id: str
+
+    workflow_start_time: datetime
+    """The start time of the workflow based on the workflow initialization."""
+
     workflow_type: str
 
     def _logger_details(self) -> Mapping[str, Any]:

tests/worker/test_workflow.py

@@ -220,12 +220,13 @@ async def test_workflow_info(client: Client, env: WorkflowEnvironment):
             maximum_interval=timedelta(seconds=5),
             maximum_attempts=6,
         )
-        info = await client.execute_workflow(
+        handle = await client.start_workflow(
             InfoWorkflow.run,
             id=workflow_id,
             task_queue=worker.task_queue,
             retry_policy=retry_policy,
         )
+        info = await handle.result()
         assert info["attempt"] == 1
         assert info["cron_schedule"] is None
         assert info["execution_timeout"] is None
@@ -235,12 +236,27 @@ async def test_workflow_info(client: Client, env: WorkflowEnvironment):
         )
         assert uuid.UUID(info["run_id"]).version == 7
         assert info["run_timeout"] is None
-        datetime.fromisoformat(info["start_time"])
         assert info["task_queue"] == worker.task_queue
         assert info["task_timeout"] == "0:00:10"
         assert info["workflow_id"] == workflow_id
         assert info["workflow_type"] == "InfoWorkflow"
 
+        async for e in handle.fetch_history_events():
+            if e.HasField("workflow_execution_started_event_attributes"):
+                assert info["workflow_start_time"] == json.loads(
+                    json.dumps(
+                        e.event_time.ToDatetime().replace(tzinfo=timezone.utc),
+                        default=str,
+                    )
+                )
+            elif e.HasField("workflow_task_started_event_attributes"):
+                assert info["start_time"] == json.loads(
+                    json.dumps(
+                        e.event_time.ToDatetime().replace(tzinfo=timezone.utc),
+                        default=str,
+                    )
+                )
+
 
 @dataclass
 class HistoryInfo: