class_docstr: A job manager is a class that handles a job runner compute, for a specific processing step.
  
  Args:
      job_info (:obj:`JobInfo`):
          The job to process. It contains the job_id, the job type, the dataset, the revision, the config,
          the split and the priority level.
      app_config (:obj:`AppConfig`):
          The app config.
      job_runner (:obj:`JobRunner`):
          The job runner to use.
class_docstr_tok: ["A","job","manager","is","a","class","that","handles","a","job","runner","compute",",","for","a","specific","processing","step",".","Args",":","job_info","(",":","obj",":","`","JobInfo","`",")",":","The","job","to","process",".","It","contains","the","job_id",",","the","job","type",",","the","dataset",",","the","revision",",","the","config",",","the","split","and","the","priority","level",".","app_config","(",":","obj",":","`","AppConfig","`",")",":","The","app","config",".","job_runner","(",":","obj",":","`","JobRunner","`",")",":","The","job","runner","to","use","."]
code: def process(
      self,
  ) -> JobResult:
      self.info(f"compute {self}")
      try:
          try:
              self.job_runner.pre_compute()
              for (
                  parallel_step_name
              ) in self.job_runner.parallel_step_names:
                  self.raise_if_parallel_response_exists(
                      parallel_step_name=parallel_step_name
                  )
  
              job_result = self.job_runner.compute()
              content = job_result.content
  
              # Validate content size
              if (
                  len(orjson_dumps(content))
                  > self.worker_config.content_max_bytes
              ):
                  raise TooBigContentError(
                      "The computed response content exceeds the supported size in bytes"
                      f" ({self.worker_config.content_max_bytes})."
                  )
          except CachedArtifactNotFoundError as err:
              raise PreviousStepStillProcessingError(
                  message="The previous steps are still being processed",
                  cause=err,
              )
          finally:
              # ensure the post_compute hook is called even if the compute raises an exception
              self.job_runner.post_compute()
          self.debug(
              f"dataset={self.job_params['dataset']} revision={self.job_params['revision']} job_info={self.job_info}"
              " is valid"
          )
          return {
              "job_info": self.job_info,
              "job_runner_version": self.job_runner_version,
              "is_success": True,
              "output": {
                  "content": content,
                  "http_status": HTTPStatus.OK,
                  "error_code": None,
                  "details": None,
                  "progress": job_result.progress,
              },
          }
      except DatasetNotFoundError:
          # To avoid filling the cache, we don't save this error. Otherwise, DoS is possible.
          self.debug(
              f"the dataset={self.job_params['dataset']} could not be found, don't update the cache"
          )
          return {
              "job_info": self.job_info,
              "job_runner_version": self.job_runner_version,
              "is_success": False,
              "output": None,
          }
      except CachedArtifactError as err:
          # A previous step (cached artifact required by the job runner) is an error. We copy the cached entry,
          # so that users can see the underlying error (they are not interested in the internals of the graph).
          # We add an entry to details: "copied_from_artifact", with its identification details, to have a chance
          # to debug if needed.
          self.debug(
              f"response for job_info={self.job_info} had an error from a previous step"
          )
          return {
              "job_info": self.job_info,
              "job_runner_version": self.job_runner_version,
              "is_success": False,
              "output": {
                  "content": err.cache_entry_with_details["content"],
                  "http_status": err.cache_entry_with_details[
                      "http_status"
                  ],
                  "error_code": err.cache_entry_with_details[
                      "error_code"
                  ],
                  "details": err.enhanced_details,
                  "progress": None,
              },
          }
      except Exception as err:
          e = (
              err
              if isinstance(err, CustomError)
              else DatasetScriptError(str(err), err)
              if is_dataset_script_error()
              else UnexpectedError(str(err), err)
          )
          self.debug(
              f"response for job_info={self.job_info} had an error"
          )
          return {
              "job_info": self.job_info,
              "job_runner_version": self.job_runner_version,
              "is_success": False,
              "output": {
                  "content": dict(e.as_response()),
                  "http_status": e.status_code,
                  "error_code": e.code,
                  "details": dict(e.as_response_with_cause()),
                  "progress": None,
              },
          }
  
imports: import logging
  from http import HTTPStatus
  from typing import Optional
  from libcommon.config import CommonConfig
  from libcommon.exceptions import CustomError, DatasetInBlockListError, DatasetNotFoundError, DatasetScriptError, JobManagerCrashedError, JobManagerExceededMaximumDurationError, PreviousStepStillProcessingError, ResponseAlreadyComputedError, TooBigContentError, UnexpectedError
  from libcommon.orchestrator import finish_job
  from libcommon.processing_graph import processing_graph
  from libcommon.simple_cache import CachedArtifactError, CachedArtifactNotFoundError, CacheEntryDoesNotExistError, get_response_without_content_params
  from libcommon.utils import JobInfo, JobParams, JobResult, Priority, orjson_dumps
  from worker.config import AppConfig, WorkerConfig
  from worker.job_runner import JobRunner
  from worker.utils import is_dataset_script_error
label_desc: Use image node_id 13 for calling the JobManager obj's underlying member method code with example usage: obj.process() and returns: dict, dict, dict, dict