code: def test_worker_exit_intended_system_exit_and_user_error(
      ray_start_cluster,
  ):
      """
      INTENDED_SYSTEM_EXIT
      - (not tested, hard to test) Unused resource removed
      - (tested) Pg removed
      - (tested) Idle
      USER_ERROR
      - (tested) Actor init failed
      """
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=1)
      ray.init(address=cluster.address)
  
      @ray.remote
      def f():
          return ray.get(g.remote())
  
      @ray.remote
      def g():
          return os.getpid()
  
      # Start a task that has a blocking call ray.get with g.remote.
      # g.remote will borrow the CPU and start a new worker.
      # The worker started for g.remote will exit by IDLE timeout.
      pid = ray.get(f.remote())
  
      def verify_exit_by_idle_timeout():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          return (
              type == "INTENDED_SYSTEM_EXIT" and "it was idle" in detail
          )
  
      wait_for_condition(verify_exit_by_idle_timeout)
  
      @ray.remote(num_cpus=1)
      class A:
          def __init__(self):
              self.sleeping = False
  
          async def getpid(self):
              while not self.sleeping:
                  await asyncio.sleep(0.1)
              return os.getpid()
  
          async def sleep(self):
              self.sleeping = True
              await asyncio.sleep(9999)
  
      pg = ray.util.placement_group(bundles=[{"CPU": 1}])
      a = A.options(
          scheduling_strategy=PlacementGroupSchedulingStrategy(
              placement_group=pg
          )
      ).remote()
      a.sleep.options(name="sleep").remote()
      pid = ray.get(a.getpid.remote())
      ray.util.remove_placement_group(pg)
  
      def verify_exit_by_pg_removed():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          assert verify_failed_task(
              name="sleep",
              error_type="ACTOR_DIED",
              error_message=[
                  "INTENDED_SYSTEM_EXIT",
                  "placement group was removed",
              ],
          )
          return (
              type == "INTENDED_SYSTEM_EXIT"
              and "placement group was removed" in detail
          )
  
      wait_for_condition(verify_exit_by_pg_removed)
  
      @ray.remote
      class PidDB:
          def __init__(self):
              self.pid = None
  
          def record_pid(self, pid):
              self.pid = pid
  
          def get_pid(self):
              return self.pid
  
      p = PidDB.remote()
  
      @ray.remote
      class FaultyActor:
          def __init__(self):
              p.record_pid.remote(os.getpid())
              raise Exception("exception in the initialization method")
  
          def ready(self):
              pass
  
      a = FaultyActor.remote()
      wait_for_condition(
          lambda: ray.get(p.get_pid.remote()) is not None
      )
      pid = ray.get(p.get_pid.remote())
  
      def verify_exit_by_actor_init_failure():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          assert (
              type == "USER_ERROR"
              and "exception in the initialization method" in detail
          )
          return verify_failed_task(
              name="FaultyActor.__init__",
              error_type="TASK_EXECUTION_EXCEPTION",
              error_message="exception in the initialization method",
          )
  
      wait_for_condition(verify_exit_by_actor_init_failure)
  
file_path: ray/python/ray/tests/test_exit_observability.py
imports: import asyncio
  import os
  import signal
  import sys
  import pytest
  from ray._private.state_api_test_utils import verify_failed_task
  import ray
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.util.state import list_workers, list_nodes, list_tasks
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 4 for calling a global function with example usage: test_worker_exit_intended_system_exit_and_user_error(ray_start_cluster) and returns: ray, os, unknown, unknown, verify_failed_task, os, self