code: def test_task_events_gc_default_policy(shutdown_only):
      @ray.remote
      def finish_task():
          pass
  
      @ray.remote
      def running_task():
          time.sleep(999)
  
      @ray.remote
      class Actor:
          def actor_finish_task(self):
              pass
  
          def actor_running_task(self):
              time.sleep(999)
  
          def ready(self):
              pass
  
      @ray.remote(max_retries=0)
      def error_task():
          raise ValueError("Expected to fail")
  
      ray.init(
          num_cpus=8,
          _system_config={
              "task_events_max_num_task_in_gcs": 5,
              "task_events_skip_driver_for_test": True,
              "task_events_report_interval_ms": 100,
          },
      )
      a = Actor.remote()
      ray.get(a.ready.remote())
      # Run 10 and 5 should be evicted
      ray.get([finish_task.remote() for _ in range(10)])
  
      def verify_tasks(expected_tasks_cnt: Dict[str, int]):
          tasks = list_tasks(raise_on_missing_output=False)
          total_cnt = reduce(
              lambda x, y: x + y, expected_tasks_cnt.values()
          )
          assert len(tasks) == total_cnt
          actual_cnt = defaultdict(int)
          for task in tasks:
              actual_cnt[task.name] += 1
          assert actual_cnt == expected_tasks_cnt
          return True
  
      wait_for_condition(
          verify_tasks, expected_tasks_cnt={"finish_task": 5}
      )
  
      # Run a few other tasks to occupy the buffer
      running_task.remote()
      error_task.remote()
  
      wait_for_condition(
          verify_tasks,
          expected_tasks_cnt={
              "finish_task": 3,
              "running_task": 1,
              "error_task": 1,
          },
      )
  
      # Run more finished tasks should not evict those running/error tasks
      for _ in range(3):
          ray.get(a.actor_finish_task.remote())
  
      wait_for_condition(
          verify_tasks,
          expected_tasks_cnt={
              "Actor.actor_finish_task": 3,
              "running_task": 1,
              "error_task": 1,
          },
      )
  
      # Run actor non-finished tasks should not evict those running/error tasks
      [a.actor_running_task.remote() for _ in range(3)]
  
      wait_for_condition(
          verify_tasks,
          expected_tasks_cnt={
              "Actor.actor_running_task": 3,
              "running_task": 1,
              "error_task": 1,
          },
      )
  
      # Run more error tasks now should evict the older "running_task"
      [error_task.remote() for _ in range(5)]
  
      wait_for_condition(
          verify_tasks, expected_tasks_cnt={"error_task": 5}
      )
  
imports: import asyncio
  from collections import defaultdict
  import os
  from typing import Dict
  import pytest
  import sys
  import time
  from ray._private import ray_constants
  from functools import reduce
  import ray
  from ray._private.state_api_test_utils import PidActor, get_state_api_manager, verify_tasks_running_or_terminated, verify_failed_task
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import async_wait_for_condition_async_predicate, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_actors, list_jobs, list_tasks
label_desc: Use image node_id 19 for calling a global function with example usage: test_task_events_gc_default_policy(shutdown_only) and returns: True