code: def test_fault_tolerance_detached_actor(shutdown_only):
      """
      Tests that tasks from a detached actor **shouldn't** be marked as failed
      """
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      pid_actor = PidActor.remote()
  
      # Check a detached actor's parent task's failure do not
      # affect the actor's task subtree.
      @ray.remote(max_retries=0)
      def parent_starts_detached_actor(pid_actor):
          @ray.remote
          class DetachedActor:
              def __init__(self):
                  pass
  
              async def running(self):
                  while not self.running:
                      await asyncio.sleep(0.1)
                  pass
  
              async def run(self, pid_actor):
                  ray.get(
                      pid_actor.report_pid.remote(
                          "detached-actor-run", os.getpid(), "RUNNING"
                      )
                  )
                  self.running = True
                  await asyncio.sleep(999)
  
          # start a detached actor
          a = DetachedActor.options(
              name="detached-actor",
              lifetime="detached",
              namespace="test",
          ).remote()
          a.run.options(name="detached-actor-run").remote(pid_actor)
          ray.get(a.running.remote())
  
          # Enough time for events to be reported to GCS.
          time.sleep(1)
          # fail this parent task
          os._exit(1)
  
      with pytest.raises(ray.exceptions.WorkerCrashedError):
          ray.get(parent_starts_detached_actor.remote(pid_actor))
  
      a = ray.get_actor("detached-actor", namespace="test")
      task_pids = ray.get(pid_actor.get_pids.remote())
      wait_for_condition(
          verify_tasks_running_or_terminated,
          task_pids=task_pids,
          expect_num_tasks=1,
      )
  
      a = ray.get_actor("detached-actor", namespace="test")
      ray.kill(a)
  
      # Verify the actual process no longer running.
      task_pids["detached-actor-run"] = (
          task_pids["detached-actor-run"][0],
          "FAILED",
      )
      wait_for_condition(
          verify_tasks_running_or_terminated,
          task_pids=task_pids,
          expect_num_tasks=1,
      )
  
      # Verify failed task marked with expected info.
      wait_for_condition(
          verify_failed_task,
          name="detached-actor-run",
          error_type="WORKER_DIED",
          error_message="The actor is dead because it was killed by `ray.kill`",
      )
  
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 6 for calling a global function with example usage: test_fault_tolerance_detached_actor(shutdown_only) without return types