code: def _wait_job(self, timeout: int):
      # The context ensures the job always either finishes normally
      # or is terminated.
      with self._terminate_job_context():
          assert self.job_id, "Job must have been started"
  
          start_time = time.monotonic()
          # Waiting for cluster needs to be a part of the whole
          # run.
          timeout_at = start_time + self.cluster_startup_timeout
          next_status = start_time + 30
          job_running = False
  
          while True:
              now = time.monotonic()
              if now >= timeout_at:
                  self._terminate_job()
                  if not job_running:
                      raise JobStartupTimeout(
                          "Cluster did not start within "
                          f"{self.cluster_startup_timeout} seconds."
                      )
                  raise CommandTimeout(
                      f"Job timed out after {timeout} seconds."
                  )
  
              if now >= next_status:
                  if job_running:
                      msg = "... job still running ..."
                  else:
                      msg = "... job not yet running ..."
                  logger.info(
                      f"{msg}({int(now - start_time)} seconds, "
                      f"{int(timeout_at - now)} seconds to job timeout) ..."
                  )
                  next_status += 30
  
              result = self._get_job_status_with_retry()
              self.last_job_result = result
              status = self.last_job_status
  
              if not job_running and status in {
                  HaJobStates.RUNNING,
                  HaJobStates.ERRORED,
              }:
                  logger.info(
                      f"... job started ...({int(now - start_time)} seconds) ..."
                  )
                  job_running = True
                  # If job has started, we switch from waiting for cluster
                  # to the actual command (incl. prepare commands) timeout.
                  timeout_at = now + timeout
  
              if status in terminal_state:
                  logger.info(f"Job entered terminal state {status}.")
                  break
              time.sleep(1)
  
      result = self._get_job_status_with_retry()
      self.last_job_result = result
      status = self.last_job_status
      assert status in terminal_state
      if status == HaJobStates.TERMINATED and not job_running:
          # Soft infra error
          retcode = -4
      else:
          retcode = job_status_to_return_code[status]
      self._duration = time.time() - self.start_time
      return retcode, self._duration
  
imports: import os
  import time
  import subprocess
  import tempfile
  from collections import deque
  from contextlib import contextmanager
  from typing import Any, Dict, Optional, Tuple, List
  from anyscale.sdk.anyscale_client.models import CreateProductionJob, HaJobStates
  from ray_release.anyscale_util import LAST_LOGS_LENGTH, get_cluster_name
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.exception import CommandTimeout, JobStartupTimeout, JobStartupFailed
  from ray_release.logger import logger
  from ray_release.signal_handling import register_handler, unregister_handler
  from ray_release.util import ANYSCALE_HOST, ERROR_LOG_PATTERNS, exponential_backoff_retry, anyscale_job_url, format_link
label_desc: Use image node_id 13 for calling the AnyscaleJobManager obj's underlying member method code with example usage: obj._wait_job(timeout) and returns: retcode,  self