code: def test_job_exit_cleanup(ray_start_regular):
      address = ray_start_regular["address"]
  
      driver_script = """
  import time
  
  import ray
  ray.init(address="{}")
  
  @ray.remote
  def f():
      ray.util.rpdb.set_trace()
  
  f.remote()
  # Give the remote function long enough to actually run.
  time.sleep(5)
  """.format(
          address
      )
  
      assert not len(
          ray.experimental.internal_kv._internal_kv_list(
              "RAY_PDB_", namespace=ray_constants.KV_NAMESPACE_PDB
          )
      )
  
      run_string_as_driver(driver_script)
  
      def one_active_session():
          return len(
              ray.experimental.internal_kv._internal_kv_list(
                  "RAY_PDB_", namespace=ray_constants.KV_NAMESPACE_PDB
              )
          )
  
      wait_for_condition(one_active_session)
  
      # Start the debugger. This should clean up any existing sessions that
      # belong to dead jobs.
      p = pexpect.spawn("ray debug")  # noqa:F841
  
      def no_active_sessions():
          return not len(
              ray.experimental.internal_kv._internal_kv_list(
                  "RAY_PDB_", namespace=ray_constants.KV_NAMESPACE_PDB
              )
          )
  
      wait_for_condition(no_active_sessions)
  
imports: import json
  import os
  import platform
  import subprocess
  import sys
  from telnetlib import Telnet
  import unittest
  import pexpect
  import pytest
  import ray
  from ray._private import ray_constants, services
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.cluster_utils import Cluster, cluster_not_supported
label_desc: Use image node_id 5 for calling a global function with example usage: test_job_exit_cleanup(ray_start_regular) and returns: len, unknown