class_docstr: Provide python APIs for job submission and management.
  
  It does not provide persistence, all info will be lost if the cluster
  goes down.
docstr: Configure and return the runtime_env for the supervisor actor.
  
  Args:
      user_runtime_env: The runtime_env specified by the user.
      resources_specified: Whether the user specified resources in the
          submit_job() call. If so, we will skip the workaround introduced
          in #24546 for GPU detection and just use the user's resource
          requests, so that the behavior matches that of the user specifying
          resources for any other actor.
  
  Returns:
      The runtime_env for the supervisor actor.
docstr_tok: ["Configure","and","return","the","runtime_env","for","the","supervisor","actor",".","Args",":","user_runtime_env",":","The","runtime_env","specified","by","the","user",".","resources_specified",":","Whether","the","user","specified","resources","in","the","submit_job","(",")","call",".","If","so",",","we","will","skip","the","workaround","introduced","in","#","24546","for","GPU","detection","and","just","use","the","user","'s","resource","requests",",","so","that","the","behavior","matches","that","of","the","user","specifying","resources","for","any","other","actor",".","Returns",":","The","runtime_env","for","the","supervisor","actor","."]
code: def _get_supervisor_runtime_env(
      self,
      user_runtime_env: Dict[str, Any],
      resources_specified: bool = False,
  ) -> Dict[str, Any]:
      """Configure and return the runtime_env for the supervisor actor.
  
      Args:
          user_runtime_env: The runtime_env specified by the user.
          resources_specified: Whether the user specified resources in the
              submit_job() call. If so, we will skip the workaround introduced
              in #24546 for GPU detection and just use the user's resource
              requests, so that the behavior matches that of the user specifying
              resources for any other actor.
  
      Returns:
          The runtime_env for the supervisor actor.
      """
      # Make a copy to avoid mutating passed runtime_env.
      runtime_env = (
          copy.deepcopy(user_runtime_env)
          if user_runtime_env is not None
          else {}
      )
  
      # NOTE(edoakes): Can't use .get(, {}) here because we need to handle the case
      # where env_vars is explicitly set to `None`.
      env_vars = runtime_env.get("env_vars")
      if env_vars is None:
          env_vars = {}
  
      env_vars[ray_constants.RAY_WORKER_NICENESS] = "0"
  
      if not resources_specified:
          # Don't set CUDA_VISIBLE_DEVICES for the supervisor actor so the
          # driver can use GPUs if it wants to. This will be removed from
          # the driver's runtime_env so it isn't inherited by tasks & actors.
          env_vars[
              ray_constants.NOSET_CUDA_VISIBLE_DEVICES_ENV_VAR
          ] = "1"
      runtime_env["env_vars"] = env_vars
      return runtime_env
  
imports: import asyncio
  import copy
  import json
  import logging
  import os
  import psutil
  import random
  import signal
  import string
  import subprocess
  import sys
  import time
  import traceback
  from asyncio.tasks import FIRST_COMPLETED
  from collections import deque
  from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy, SchedulingStrategyT
  import ray
  from ray._private.gcs_utils import GcsAioClient
  from ray._private.utils import run_background_task
  import ray._private.ray_constants
  from ray._private.runtime_env.constants import RAY_JOB_CONFIG_JSON_ENV_VAR
  from ray.actor import ActorHandle
  from ray.dashboard.consts import RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR, DEFAULT_JOB_START_TIMEOUT_SECONDS, RAY_JOB_START_TIMEOUT_SECONDS_ENV_VAR
  from ray.dashboard.modules.job.common import JOB_ID_METADATA_KEY, JOB_NAME_METADATA_KEY, JOB_ACTOR_NAME_TEMPLATE, JOB_LOGS_PATH_TEMPLATE, SUPERVISOR_ACTOR_RAY_NAMESPACE, JobInfo, JobInfoStorageClient
  from ray.dashboard.modules.job.utils import file_tail_iterator
  from ray.exceptions import ActorUnschedulableError, RuntimeEnvSetupError
  from ray.job_submission import JobStatus
  from ray._private.event.event_logger import get_event_logger
  from ray.core.generated.event_pb2 import Event
label_desc: Use image node_id 4 for calling the JobManager obj's underlying member method code with example usage: obj._get_supervisor_runtime_env(user_runtime_env, resources_specified) and returns: runtime_env