code: def test_worker_exit_intended_user_exit(ray_start_cluster):
      """
      INTENDED_USER_EXIT
      - (tested) Shutdown driver
      - (tested) exit_actor
      - (tested) exit(0)
      - (tested) Actor kill request
      - (tested) Task cancel request
      """
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=4)
      ray.init(address=cluster.address)
      cluster.add_node(num_cpus=1, resources={"worker": 1})
  
      driver = """
  import ray
  import os
  ray.init(address="{address}")
  print(os.getpid())
  ray.shutdown()
  """.format(
          address=cluster.address
      )
      a = run_string_as_driver(driver)
      driver_pid = int(a.strip().split("\n")[-1].strip())
  
      def verify_worker_exit_by_shutdown():
          worker = get_worker_by_pid(driver_pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          assert (
              type == "INTENDED_USER_EXIT"
              and "ray.shutdown()" in detail
          )
          return True
  
      wait_for_condition(verify_worker_exit_by_shutdown)
  
      @ray.remote
      class A:
          def pid(self):
              return os.getpid()
  
          def exit(self):
              ray.actor.exit_actor()
  
          def exit_with_exit_code(self):
              sys.exit(0)
  
          def sleep_forever(self):
              import time
  
              # RIP
              time.sleep(999999)
  
      a = A.remote()
      pid = ray.get(a.pid.remote())
      with pytest.raises(
          ray.exceptions.RayActorError, match="exit_actor"
      ):
          ray.get(a.exit.options(name="exit").remote())
  
      def verify_worker_exit_actor():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          assert type == "INTENDED_USER_EXIT" and "exit_actor" in detail
  
          t = list_tasks(filters=[("name", "=", "exit")])[0]
          # exit_actor should be shown as non-task failure.
          assert t["state"] == "FINISHED"
          return True
  
      wait_for_condition(verify_worker_exit_actor)
  
      a = A.remote()
      pid = ray.get(a.pid.remote())
      with pytest.raises(
          ray.exceptions.RayActorError, match="exit code 0"
      ):
          ray.get(
              a.exit_with_exit_code.options(
                  name="exit_with_exit_code"
              ).remote()
          )
  
      def verify_exit_code_0():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          assert (
              type == "INTENDED_USER_EXIT" and "exit code 0" in detail
          )
          t = list_tasks(
              filters=[("name", "=", "exit_with_exit_code")]
          )[0]
          # exit_actor should be shown as non-task failure.
          assert t["state"] == "FINISHED"
          return True
  
      wait_for_condition(verify_exit_code_0)
  
      a = A.remote()
      pid = ray.get(a.pid.remote())
      ray.kill(a)
      with pytest.raises(
          ray.exceptions.RayActorError, match="ray.kill"
      ):
          ray.get(
              a.sleep_forever.options(name="sleep_forever").remote()
          )
  
      def verify_exit_by_ray_kill():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          assert type == "INTENDED_SYSTEM_EXIT" and "ray.kill" in detail
          return verify_failed_task(
              name="sleep_forever",
              error_type="ACTOR_DIED",
              error_message="ray.kill",
          )
  
      wait_for_condition(verify_exit_by_ray_kill)
  
      @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
      def f():
          ray.get(p.record_pid.remote(os.getpid()))
          import time
  
          time.sleep(300)
  
      t = f.options(name="cancel-f").remote()
      wait_for_condition(
          lambda: ray.get(p.get_pid.remote()) is not None, timeout=300
      )
      ray.cancel(t, force=True)
      pid = ray.get(p.get_pid.remote())
  
      def verify_exit_by_ray_cancel():
          worker = get_worker_by_pid(pid)
          type = worker["exit_type"]
          detail = worker["exit_detail"]
          assert type == "INTENDED_USER_EXIT" and "ray.cancel" in detail
          return verify_failed_task(
              name="cancel-f",
              error_type="WORKER_DIED",  # Since it's a force cancel through kill signal.
              error_message="Socket closed",
          )
  
      wait_for_condition(verify_exit_by_ray_cancel)
  
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 3 for calling a global function with example usage: test_worker_exit_intended_user_exit(ray_start_cluster) and returns: True, True, True, verify_failed_task, verify_failed_task, os, self