[core][event] rename TaskExecutionEvent to TaskLifecycleEvent (#56853)

This PR refactors the `TaskExecutionEvent` proto in two ways:
- Rename the file to `events_task_lifecycle_event.proto`
- Refactor the task_state from a map to an array of TaskState and
timestamp. Also rename the field to `state_transitions` for consistency.

This PR depends on the upstream to update their logic to consume this
new schema.

Test:
- CI

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Renames task execution event to task lifecycle event and changes its
schema from a state map to an ordered state_transitions list, updating
core, GCS, dashboard, builds, tests, and docs.
> 
> - **Proto/API changes (breaking)**
> - Rename `TaskExecutionEvent` → `TaskLifecycleEvent` and update
`RayEvent.EventType` (`TASK_EXECUTION_EVENT` → `TASK_LIFECYCLE_EVENT`).
> - Replace `task_state` map with `state_transitions` (list of `{state,
timestamp}`) in `events_task_lifecycle_event.proto`.
> - Update `events_base_event.proto` field from `task_execution_event` →
`task_lifecycle_event` and imports/BUILD deps accordingly.
> - **Core worker**
> - Update buffer/conversion logic in
`src/ray/core_worker/task_event_buffer.{h,cc}` to populate/emit
`TaskLifecycleEvent` with `state_transitions`.
> - **GCS**
> - Update `GcsRayEventConverter` to consume `TASK_LIFECYCLE_EVENT` and
convert `state_transitions` to `state_ts_ns`.
> - **Dashboard/Aggregator**
> - Switch exposable type defaults/env to `TASK_LIFECYCLE_EVENT` in
`python/.../aggregator_agent.py`.
> - **Tests**
> - Adjust unit tests to new event/type and schema across core worker,
GCS, and dashboard.
> - **Docs**
>   - Update event export guide references to new lifecycle event proto.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
61507e8. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Signed-off-by: Cuong Nguyen <can@anyscale.com>

Author: can-anyscale

doc/source/ray-observability/user-guides/ray-event-export.rst

@@ -45,7 +45,7 @@ For each task, Ray exports two types of events: Task Definition Event and Task E
   and actor tasks respectively.
 * Task Execution Events contain task state transition information and metadata 
   generated during task execution. 
-  See `src/ray/protobuf/public/events_task_execution_event.proto <https://github.com/ray-project/ray/blob/master/src/ray/protobuf/public/events_task_execution_event.proto>`_ for the event format.
+  See `src/ray/protobuf/public/events_task_lifecycle_event.proto <https://github.com/ray-project/ray/blob/master/src/ray/protobuf/public/events_task_lifecycle_event.proto>`_ for the event format.
 
 An example of a Task Definition Event and a Task Execution Event:
 
@@ -73,16 +73,7 @@ An example of a Task Definition Event and a Task Execution Event:
             "requiredResources":{
                 "CPU":1.0
             },
-            "runtimeEnvInfo":{
-                "serializedRuntimeEnv":"{}",
-                "runtimeEnvConfig":{
-                    "setupTimeoutSeconds":600,
-                    "eagerInstall":true,
-                    "logFiles":[
-                    
-                    ]
-                }
-            },
+            "serialized_runtime_env": "{}",
             "jobId":"AQAAAA==",
             "parentTaskId":"//////////////////////////8BAAAA",
             "placementGroupId":"////////////////////////",
@@ -96,23 +87,30 @@ An example of a Task Definition Event and a Task Execution Event:
         "message":""
     }
 
-    // task execution event
+    // task lifecycle event
     {
         "eventId":"vkIaAHlQC5KoppGosqs2kBq5k2WzsAAbawDDbQ==",
         "sourceType":"CORE_WORKER",
-        "eventType":"TASK_EXECUTION_EVENT",
+        "eventType":"TASK_LIFECYCLE_EVENT",
         "timestamp":"2025-09-03T18:52:14.469074Z",
         "severity":"INFO",
         "sessionName":"session_2025-09-03_11-52-12_635210_85618",
-        "taskExecutionEvent":{
+        "taskLifecycleEvent":{
             "taskId":"yO9FzNARJXH///////////////8BAAAA",
-            "taskState":{
-                // key is the integer value of TaskStatus enum in common.proto at
-                // https://github.com/ray-project/ray/blob/master/src/ray/protobuf/common.proto
-                "2":"2025-09-03T18:52:14.467402Z", // PENDING_NODE_ASSIGNMENT
-                "1":"2025-09-03T18:52:14.467290Z", // PENDING_ARGS_AVAIL
-                "5":"2025-09-03T18:52:14.469074Z" // SUBMITTED_TO_WORKER
-            },
+            "stateTransitions": [
+                {
+                    "state":"PENDING_NODE_ASSIGNMENT",
+                    "timestamp":"2025-09-03T18:52:14.467402Z"
+                },
+                {
+                    "state":"PENDING_ARGS_AVAIL",
+                    "timestamp":"2025-09-03T18:52:14.467290Z"
+                },
+                {
+                    "state":"SUBMITTED_TO_WORKER",
+                    "timestamp":"2025-09-03T18:52:14.469074Z"
+                }
+            ],
             "nodeId":"ZvxTI6x9dlMFqMlIHErJpg5UEGK1INsKhW2zyg==",
             "workerId":"hMybCNYIFi+/yInYYhdc+qH8yMF65j/8+uCTmw==",
             "jobId":"AQAAAA==",

python/ray/dashboard/modules/aggregator/aggregator_agent.py

@@ -54,8 +54,7 @@
 # The list of all supported event types can be found in src/ray/protobuf/public/events_base_event.proto (EventType enum)
 # By default TASK_PROFILE_EVENT is not exposed to external services
 DEFAULT_EXPOSABLE_EVENT_TYPES = (
-    "TASK_DEFINITION_EVENT,TASK_EXECUTION_EVENT,"
-    "ACTOR_TASK_DEFINITION_EVENT,ACTOR_TASK_EXECUTION_EVENT,"
+    "TASK_DEFINITION_EVENT,TASK_LIFECYCLE_EVENT,ACTOR_TASK_DEFINITION_EVENT,"
     "DRIVER_JOB_DEFINITION_EVENT,DRIVER_JOB_LIFECYCLE_EVENT,"
     "ACTOR_DEFINITION_EVENT,ACTOR_LIFECYCLE_EVENT,"
     "NODE_DEFINITION_EVENT,NODE_LIFECYCLE_EVENT,"

python/ray/dashboard/modules/aggregator/tests/test_aggregator_agent.py

@@ -41,8 +41,8 @@
 from ray.core.generated.events_task_definition_event_pb2 import (
     TaskDefinitionEvent,
 )
-from ray.core.generated.events_task_execution_event_pb2 import (
-    TaskExecutionEvent,
+from ray.core.generated.events_task_lifecycle_event_pb2 import (
+    TaskLifecycleEvent,
 )
 from ray.core.generated.events_task_profile_events_pb2 import TaskProfileEvents
 from ray.core.generated.profile_events_pb2 import ProfileEventEntry, ProfileEvents
@@ -551,20 +551,23 @@ def _verify_task_definition_event_json(req_json, expected_timestamp):
     }
 
 
-def _create_task_execution_event_proto(timestamp):
+def _create_task_lifecycle_event_proto(timestamp):
     return RayEvent(
         event_id=b"1",
         source_type=RayEvent.SourceType.CORE_WORKER,
-        event_type=RayEvent.EventType.TASK_EXECUTION_EVENT,
+        event_type=RayEvent.EventType.TASK_LIFECYCLE_EVENT,
         timestamp=timestamp,
         severity=RayEvent.Severity.INFO,
         session_name="test_session",
-        task_execution_event=TaskExecutionEvent(
+        task_lifecycle_event=TaskLifecycleEvent(
             task_id=b"1",
             task_attempt=1,
-            task_state={
-                TaskStatus.RUNNING: timestamp,
-            },
+            state_transitions=[
+                TaskLifecycleEvent.StateTransition(
+                    state=TaskStatus.RUNNING,
+                    timestamp=timestamp,
+                ),
+            ],
             ray_error_info=RayErrorInfo(
                 error_type=ErrorType.TASK_EXECUTION_EXCEPTION,
             ),
@@ -575,13 +578,13 @@ def _create_task_execution_event_proto(timestamp):
     )
 
 
-def _verify_task_execution_event_json(req_json, expected_timestamp):
+def _verify_task_lifecycle_event_json(req_json, expected_timestamp):
     assert len(req_json) == 1
 
     # Verify the base event fields
     assert req_json[0]["eventId"] == base64.b64encode(b"1").decode()
     assert req_json[0]["sourceType"] == "CORE_WORKER"
-    assert req_json[0]["eventType"] == "TASK_EXECUTION_EVENT"
+    assert req_json[0]["eventType"] == "TASK_LIFECYCLE_EVENT"
     assert req_json[0]["timestamp"] == expected_timestamp
     assert req_json[0]["severity"] == "INFO"
     assert (
@@ -591,23 +594,26 @@ def _verify_task_execution_event_json(req_json, expected_timestamp):
 
     # Verify the task execution event specific fields
     assert (
-        req_json[0]["taskExecutionEvent"]["taskId"] == base64.b64encode(b"1").decode()
+        req_json[0]["taskLifecycleEvent"]["taskId"] == base64.b64encode(b"1").decode()
     )
-    assert req_json[0]["taskExecutionEvent"]["taskAttempt"] == 1
-    assert req_json[0]["taskExecutionEvent"]["taskState"] == {
-        "8": expected_timestamp,
-    }
+    assert req_json[0]["taskLifecycleEvent"]["taskAttempt"] == 1
+    assert req_json[0]["taskLifecycleEvent"]["stateTransitions"] == [
+        {
+            "state": "RUNNING",
+            "timestamp": expected_timestamp,
+        }
+    ]
     assert (
-        req_json[0]["taskExecutionEvent"]["rayErrorInfo"]["errorType"]
+        req_json[0]["taskLifecycleEvent"]["rayErrorInfo"]["errorType"]
         == "TASK_EXECUTION_EXCEPTION"
     )
     assert (
-        req_json[0]["taskExecutionEvent"]["nodeId"] == base64.b64encode(b"1").decode()
+        req_json[0]["taskLifecycleEvent"]["nodeId"] == base64.b64encode(b"1").decode()
     )
     assert (
-        req_json[0]["taskExecutionEvent"]["workerId"] == base64.b64encode(b"1").decode()
+        req_json[0]["taskLifecycleEvent"]["workerId"] == base64.b64encode(b"1").decode()
     )
-    assert req_json[0]["taskExecutionEvent"]["workerPid"] == 1
+    assert req_json[0]["taskLifecycleEvent"]["workerPid"] == 1
 
 
 def _create_profile_event_request(timestamp):
@@ -680,9 +686,9 @@ def _verify_profile_event_json(req_json, expected_timestamp):
         id="task_definition_event",
     ),
     pytest.param(
-        _create_task_execution_event_proto,
-        _verify_task_execution_event_json,
-        id="task_execution_event",
+        _create_task_lifecycle_event_proto,
+        _verify_task_lifecycle_event_json,
+        id="task_lifecycle_event",
     ),
     pytest.param(
         _create_profile_event_request, _verify_profile_event_json, id="profile_event"
@@ -697,7 +703,7 @@ def _verify_profile_event_json(req_json, expected_timestamp):
         {
             "env_vars": {
                 "RAY_DASHBOARD_AGGREGATOR_AGENT_EVENTS_EXPORT_ADDR": _EVENT_AGGREGATOR_AGENT_TARGET_ADDR,
-                "RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES": "TASK_DEFINITION_EVENT,TASK_EXECUTION_EVENT,ACTOR_TASK_DEFINITION_EVENT,ACTOR_TASK_EXECUTION_EVENT,TASK_PROFILE_EVENT",
+                "RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES": "TASK_DEFINITION_EVENT,TASK_LIFECYCLE_EVENT,ACTOR_TASK_DEFINITION_EVENT,TASK_PROFILE_EVENT",
             },
         },
     ],

python/ray/dashboard/modules/aggregator/tests/test_multi_consumer_event_buffer.py

@@ -62,7 +62,7 @@ async def test_add_event_buffer_overflow(self):
         events = []
         event_types = [
             RayEvent.EventType.TASK_DEFINITION_EVENT,
-            RayEvent.EventType.TASK_EXECUTION_EVENT,
+            RayEvent.EventType.TASK_LIFECYCLE_EVENT,
             RayEvent.EventType.ACTOR_TASK_DEFINITION_EVENT,
         ]
         for i in range(3):

src/ray/core_worker/task_event_buffer.cc

@@ -216,17 +216,20 @@ void TaskStatusEvent::PopulateRpcRayTaskDefinitionEvent(T &definition_event_data
   }
 }
 
-void TaskStatusEvent::PopulateRpcRayTaskExecutionEvent(
-    rpc::events::TaskExecutionEvent &execution_event_data,
+void TaskStatusEvent::PopulateRpcRayTaskLifecycleEvent(
+    rpc::events::TaskLifecycleEvent &lifecycle_event_data,
     google::protobuf::Timestamp timestamp) {
   // Task identifier
-  execution_event_data.set_task_id(task_id_.Binary());
-  execution_event_data.set_task_attempt(attempt_number_);
+  lifecycle_event_data.set_task_id(task_id_.Binary());
+  lifecycle_event_data.set_task_attempt(attempt_number_);
 
   // Task state
-  auto &task_state = *execution_event_data.mutable_task_state();
   if (task_status_ != rpc::TaskStatus::NIL) {
-    task_state[task_status_] = timestamp;
+    rpc::events::TaskLifecycleEvent::StateTransition state_transition;
+    state_transition.set_state(task_status_);
+    state_transition.mutable_timestamp()->CopyFrom(timestamp);
+    *lifecycle_event_data.mutable_state_transitions()->Add() =
+        std::move(state_transition);
   }
 
   // Task property updates
@@ -235,30 +238,30 @@ void TaskStatusEvent::PopulateRpcRayTaskExecutionEvent(
   }
 
   if (state_update_->error_info_.has_value()) {
-    execution_event_data.mutable_ray_error_info()->CopyFrom(*state_update_->error_info_);
+    lifecycle_event_data.mutable_ray_error_info()->CopyFrom(*state_update_->error_info_);
   }
 
   if (state_update_->node_id_.has_value()) {
     RAY_CHECK(task_status_ == rpc::TaskStatus::SUBMITTED_TO_WORKER)
             .WithField("TaskStatus", task_status_)
         << "Node ID should be included when task status changes to "
            "SUBMITTED_TO_WORKER.";
-    execution_event_data.set_node_id(state_update_->node_id_->Binary());
+    lifecycle_event_data.set_node_id(state_update_->node_id_->Binary());
   }
 
   if (state_update_->worker_id_.has_value()) {
     RAY_CHECK(task_status_ == rpc::TaskStatus::SUBMITTED_TO_WORKER)
             .WithField("TaskStatus", task_status_)
         << "Worker ID should be included when task status changes to "
            "SUBMITTED_TO_WORKER.";
-    execution_event_data.set_worker_id(state_update_->worker_id_->Binary());
+    lifecycle_event_data.set_worker_id(state_update_->worker_id_->Binary());
   }
 
   if (state_update_->pid_.has_value()) {
-    execution_event_data.set_worker_pid(state_update_->pid_.value());
+    lifecycle_event_data.set_worker_pid(state_update_->pid_.value());
   }
 
-  execution_event_data.set_job_id(job_id_.Binary());
+  lifecycle_event_data.set_job_id(job_id_.Binary());
 }
 
 void TaskStatusEvent::PopulateRpcRayEventBaseFields(
@@ -278,7 +281,7 @@ void TaskStatusEvent::PopulateRpcRayEventBaseFields(
       ray_event.set_event_type(rpc::events::RayEvent::TASK_DEFINITION_EVENT);
     }
   } else {
-    ray_event.set_event_type(rpc::events::RayEvent::TASK_EXECUTION_EVENT);
+    ray_event.set_event_type(rpc::events::RayEvent::TASK_LIFECYCLE_EVENT);
   }
 }
 
@@ -301,14 +304,14 @@ void TaskStatusEvent::ToRpcRayEvents(RayEventsTuple &ray_events_tuple) {
   }
 
   // Populate the task execution event
-  PopulateRpcRayEventBaseFields(ray_events_tuple.task_execution_event.has_value()
-                                    ? ray_events_tuple.task_execution_event.value()
-                                    : ray_events_tuple.task_execution_event.emplace(),
+  PopulateRpcRayEventBaseFields(ray_events_tuple.task_lifecycle_event.has_value()
+                                    ? ray_events_tuple.task_lifecycle_event.value()
+                                    : ray_events_tuple.task_lifecycle_event.emplace(),
                                 false,
                                 timestamp);
-  auto task_execution_event =
-      ray_events_tuple.task_execution_event.value().mutable_task_execution_event();
-  PopulateRpcRayTaskExecutionEvent(*task_execution_event, timestamp);
+  auto task_lifecycle_event =
+      ray_events_tuple.task_lifecycle_event.value().mutable_task_lifecycle_event();
+  PopulateRpcRayTaskLifecycleEvent(*task_lifecycle_event, timestamp);
 }
 
 void TaskProfileEvent::ToRpcTaskEvents(rpc::TaskEvents *rpc_task_events) {
@@ -629,9 +632,9 @@ TaskEventBufferImpl::CreateRayEventsDataToSend(
       auto events = data->add_events();
       *events = std::move(ray_events_tuple.task_definition_event.value());
     }
-    if (ray_events_tuple.task_execution_event) {
+    if (ray_events_tuple.task_lifecycle_event) {
       auto events = data->add_events();
-      *events = std::move(ray_events_tuple.task_execution_event.value());
+      *events = std::move(ray_events_tuple.task_lifecycle_event.value());
     }
     if (ray_events_tuple.task_profile_event) {
       auto events = data->add_events();
@@ -656,7 +659,7 @@ TaskEventBuffer::TaskEventDataToSend TaskEventBufferImpl::CreateDataToSend(
     const absl::flat_hash_set<TaskAttempt> &dropped_task_attempts_to_send) {
   // Aggregate the task events by TaskAttempt.
   absl::flat_hash_map<TaskAttempt, rpc::TaskEvents> agg_task_events;
-  // (task_attempt, (task_definition_event, task_execution_event, task_profile_event))
+  // (task_attempt, (task_definition_event, task_lifecycle_event, task_profile_event))
   absl::flat_hash_map<TaskAttempt, RayEventsTuple> agg_ray_events;
 
   auto to_rpc_event_fn =

src/ray/core_worker/task_event_buffer.h

@@ -44,15 +44,15 @@ namespace worker {
 using TaskAttempt = std::pair<TaskID, int32_t>;
 
 /// A struct containing a tuple of rpc::events::RayEvent.
-/// When converting the TaskStatusEvent, task_definition_event and task_execution_event
+/// When converting the TaskStatusEvent, task_definition_event and task_lifecycle_event
 /// will be populated with rpc::events::TaskDefinitionEvent and
-/// rpc::events::TaskExecutionEvent respectively. When converting the TaskProfileEvent,
+/// rpc::events::TaskLifecycleEvent respectively. When converting the TaskProfileEvent,
 /// task_profile_event will be populated with rpc::events::TaskProfileEvent. A struct is
-/// needed because the TaskProfileEvent, TaskDefinitionEvent and TaskExecutionEvent all
+/// needed because the TaskProfileEvent, TaskDefinitionEvent and TaskLifecycleEvent all
 /// can share the same task_id and attempt_number.
 struct RayEventsTuple {
   std::optional<rpc::events::RayEvent> task_definition_event;
-  std::optional<rpc::events::RayEvent> task_execution_event;
+  std::optional<rpc::events::RayEvent> task_lifecycle_event;
   std::optional<rpc::events::RayEvent> task_profile_event;
 };
 
@@ -171,7 +171,7 @@ class TaskStatusEvent : public TaskEvent {
 
   /// The function to convert the TaskStatusEvent class to a pair of
   /// rpc::events::RayEvent with rpc::events::TaskDefinitionEvent and
-  /// rpc::events::TaskExecutionEvent respectively. The TaskExecutionEvent will always
+  /// rpc::events::TaskLifecycleEvent respectively. The TaskLifecycleEvent will always
   /// be populated. The TaskDefinitionEvent will be populated only when the task_spec_
   /// is not null.
   /// NOTE: this method will modify internal states by moving fields of task_spec_ to
@@ -189,9 +189,9 @@ class TaskStatusEvent : public TaskEvent {
   template <typename T>
   void PopulateRpcRayTaskDefinitionEvent(T &definition_event_data);
 
-  // Helper functions to populate the task execution event of rpc::events::RayEvent
-  void PopulateRpcRayTaskExecutionEvent(
-      rpc::events::TaskExecutionEvent &execution_event_data,
+  // Helper functions to populate the task lifecycle event of rpc::events::RayEvent
+  void PopulateRpcRayTaskLifecycleEvent(
+      rpc::events::TaskLifecycleEvent &lifecycle_event_data,
       google::protobuf::Timestamp timestamp);
 
   // Helper functions to populate the base fields of rpc::events::RayEvent