code: def update_run_state(
      self,
      _id: ObjectId,
      run_state: ExecutionRunState,
      result: ExecutionResult = None,
      run_link: str = None,
      progress: ExecutionProgress = None,
  ) -> DelegatedOperationDocument:
      update = None
  
      execution_result = result
      if result is not None and not isinstance(result, ExecutionResult):
          execution_result = ExecutionResult(result=result)
  
      if run_state == ExecutionRunState.COMPLETED:
          update = {
              "$set": {
                  "run_state": run_state,
                  "completed_at": datetime.utcnow(),
                  "updated_at": datetime.utcnow(),
                  "result": execution_result.to_json()
                  if execution_result
                  else None,
              }
          }
      elif run_state == ExecutionRunState.FAILED:
          update = {
              "$set": {
                  "run_state": run_state,
                  "failed_at": datetime.utcnow(),
                  "updated_at": datetime.utcnow(),
                  "result": execution_result.to_json()
                  if execution_result
                  else None,
              }
          }
      elif run_state == ExecutionRunState.RUNNING:
          update = {
              "$set": {
                  "run_state": run_state,
                  "started_at": datetime.utcnow(),
                  "updated_at": datetime.utcnow(),
              }
          }
  
      if run_link is not None:
          update["$set"]["run_link"] = run_link
  
      if update is None:
          raise ValueError("Invalid run_state: {}".format(run_state))
  
      if progress is not None:
          update["$set"]["status"] = progress
          update["$set"]["status"]["updated_at"] = datetime.utcnow()
  
      doc = self._collection.find_one_and_update(
          filter={"_id": _id},
          update=update,
          return_document=pymongo.ReturnDocument.AFTER,
      )
  
      return DelegatedOperationDocument().from_pymongo(doc)
  
imports: import logging
  from datetime import datetime
  from typing import Any, List
  import pymongo
  from bson import ObjectId
  from pymongo import IndexModel
  from pymongo.collection import Collection
  from fiftyone.factory import DelegatedOperationPagingParams
  from fiftyone.factory.repos import DelegatedOperationDocument
  from fiftyone.operators.executor import ExecutionContext, ExecutionProgress, ExecutionResult, ExecutionRunState
label_desc: Use image node_id 7 for calling the MongoDelegatedOperationRepo obj's underlying member method code with example usage: obj.update_run_state(_id, run_state, result, run_link, progress) and returns: DelegatedOperationDocument