code: def test_stop_actor_before_start(
      ray_start_4_cpus, tmp_path, cleanup, resource_manager_cls
  ):
      """Test that actor failures are handled properly.
  
      - Start actor that either fails on init or in a task (RayActorError)
      - Schedule task on actor
      - Assert that the correct callbacks are called
      """
      actor_manager = RayActorManager(
          resource_manager=resource_manager_cls()
      )
  
      hang_marker = tmp_path / "hang.txt"
  
      @ray.remote
      class HangingActor:
          def __init__(self):
              while not hang_marker.exists():
                  time.sleep(0.05)
  
      tracked_actor = actor_manager.add_actor(
          HangingActor,
          kwargs={},
          resource_request=ResourceRequest([{"CPU": 1}]),
          on_start=_raise(RuntimeError, "Should not have started"),
          on_stop=_raise(RuntimeError, "Should not have stopped"),
      )
      while not actor_manager.is_actor_started(tracked_actor):
          actor_manager.next(0.05)
  
      # Actor started but hasn't triggered on_start, yet
      actor_manager.remove_actor(tracked_actor)
      hang_marker.write_text("")
      while actor_manager.is_actor_started(tracked_actor):
          actor_manager.next(0.05)
  
      assert actor_manager.num_live_actors == 0
  
imports: import threading
  import time
  from collections import Counter
  import gc
  from typing import Any, Optional, Type
  import pytest
  import ray
  from ray.air import ResourceRequest
  from ray.air.execution import FixedResourceManager, PlacementGroupResourceManager
  from ray.air.execution._internal import Barrier
  from ray.air.execution._internal.actor_manager import RayActorManager
label_desc: Use image node_id 8 for calling a global function with example usage: test_stop_actor_before_start(ray_start_4_cpus, tmp_path, cleanup, resource_manager_cls) without return types