[core] Enable aggregator mode in state API and task event tests

python/ray/autoscaler/v2/tests/test_node_provider.py

@@ -193,6 +193,7 @@ def _get_provider(name, **kwargs):
 def test_node_providers_basic(get_provider, provider_name):
     # Test launching.
     provider = get_provider(name=provider_name)
+    timeout_s = 30 if provider_name == "fake_multi" else 10
     provider.launch(
         shape={"worker_nodes": 2},
         request_id="1",
@@ -212,7 +213,7 @@ def verify():
         assert nodes_by_type == {"worker_nodes": 4, "worker_nodes1": 1}
         return True
 
-    wait_for_condition(verify)
+    wait_for_condition(verify, timeout=timeout_s)
 
     nodes = provider.get_non_terminated().keys()
 
@@ -238,7 +239,7 @@ def verify():
             assert node.request_id == "4"
         return True
 
-    wait_for_condition(verify)
+    wait_for_condition(verify, timeout=timeout_s)
 
 
 @pytest.mark.parametrize(

python/ray/tests/BUILD.bazel

@@ -563,7 +563,6 @@ py_test_module_list(
         "test_state_api_summary.py",
         "test_streaming_generator_regression.py",
         "test_system_metrics.py",
-        "test_task_events_3.py",
         "test_task_metrics_reconstruction.py",
         "test_top_level_api.py",
         "test_tpu.py",
@@ -876,6 +875,7 @@ py_test_module_list(
         "test_ray_get.py",
         "test_state_api_2.py",
         "test_task_events.py",
+        "test_task_events_3.py",
         "test_unavailable_actors.py",
     ],
     tags = [

python/ray/tests/conftest.py

@@ -1492,6 +1492,34 @@ def make_httpserver(httpserver_listen_address, httpserver_ssl_context):
         server.stop()
 
 
+@pytest.fixture(scope="function")
+def event_routing_config(request, monkeypatch):
+    """
+    fixture to toggle event routing modes.
+    Modes:
+      - "default": Uses the existing core_worker to gcs code path.
+      - "aggregator": Enable publishing events to GCS through the Aggregator agent.
+    """
+    mode = getattr(request, "param", "default")
+    # clear envs to ensure default behavior
+    monkeypatch.delenv(
+        "RAY_DASHBOARD_AGGREGATOR_AGENT_PUBLISH_EVENTS_TO_GCS", raising=False
+    )
+    monkeypatch.delenv("RAY_enable_core_worker_ray_event_to_aggregator", raising=False)
+
+    if mode == "aggregator":
+        print("using aggregator mode")
+        # Enable aggregator path in core worker
+        monkeypatch.setenv("RAY_enable_core_worker_ray_event_to_aggregator", "1")
+        # Explicitly disable core worker to GCS so that all events are only sent to GCS once (through the aggregator pathway)
+        monkeypatch.setenv("RAY_enable_core_worker_task_event_to_gcs", "0")
+        # Ensure aggregator agent publishes to GCS
+        monkeypatch.setenv(
+            "RAY_DASHBOARD_AGGREGATOR_AGENT_PUBLISH_EVENTS_TO_GCS", "True"
+        )
+    yield
+
+
 @pytest.fixture
 def cleanup_auth_token_env():
     """Reset authentication environment variables, files, and caches."""