code: def move_task_request(
      self, ts: TaskState, victim: WorkerState, thief: WorkerState
  ) -> str:
      try:
          if ts in self.in_flight:
              return "in-flight"
          # Stimulus IDs are used to verify the response, see
          # `move_task_confirm`. Therefore, this must be truly unique.
          stimulus_id = f"steal-{self._request_counter}"
          self._request_counter += 1
  
          key = ts.key
          self.remove_key_from_stealable(ts)
          logger.debug(
              "Request move %s, %s: %2f -> %s: %2f",
              key,
              victim,
              victim.occupancy,
              thief,
              thief.occupancy,
          )
  
          # TODO: occupancy no longer concats linearly so we can't easily
          # assume that the network cost would go down by that much
          victim_duration = self.scheduler.get_task_duration(
              ts
          ) + self.scheduler.get_comm_cost(ts, victim)
          thief_duration = self.scheduler.get_task_duration(
              ts
          ) + self.scheduler.get_comm_cost(ts, thief)
  
          self.scheduler.stream_comms[victim.address].send(
              {
                  "op": "steal-request",
                  "key": key,
                  "stimulus_id": stimulus_id,
              }
          )
          info: InFlightInfo = {
              "victim": victim,  # guaranteed to be processing_on
              "thief": thief,
              "victim_duration": victim_duration,
              "thief_duration": thief_duration,
              "stimulus_id": stimulus_id,
          }
          self._add_to_in_flight(ts, info)
          return stimulus_id
      except CommClosedError:
          logger.info(
              "Worker comm %r closed while stealing: %r", victim, ts
          )
          return "comm-closed"
      except Exception as e:  # pragma: no cover
          logger.exception(e)
          if LOG_PDB:
              import pdb
  
              pdb.set_trace()
          raise
  
imports: from __future__ import annotations
  import asyncio
  import logging
  from collections import defaultdict, deque
  from collections.abc import Container
  from functools import partial
  from math import log2
  from time import time
  from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast
  from tlz import topk
  import dask
  from dask.typing import Key
  from dask.utils import parse_timedelta
  from distributed.compatibility import PeriodicCallback
  from distributed.core import CommClosedError
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.utils import log_errors, recursive_to_dict
label_desc: Use image node_id 14 for calling the WorkStealing obj's underlying member method code with example usage: obj.move_task_request(ts, victim, thief) and returns: stimulus_id, str, str