diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py
index bff19ae91c4..bf186e253db 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py
@@ -1,7 +1,10 @@
+import asyncio
 import inspect
 import logging
+import threading
+import time
 import uuid
-from typing import Any, Dict, List, Optional, Tuple, Union, cast
+from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast
 
 from pylabrobot import utils
 from pylabrobot.io import LOG_LEVEL_IO
@@ -46,6 +49,79 @@
 
 logger = logging.getLogger(__name__)
 
+# One request/response with the robot-server. A read of robot state, not a motion.
+DEFAULT_REQUEST_TIMEOUT = 30.0
+
+# One command, including the motion it performs. The OT-2's slowest single move is a
+# full-stroke aspirate or dispense at a viscous-liquid flow rate.
+DEFAULT_COMMAND_TIMEOUT = 120.0
+
+# Delay between two reads of a running command's status.
+DEFAULT_STATUS_POLL_INTERVAL = 0.05
+
+
+def _seconds_left(deadline: float) -> float:
+  return max(deadline - time.monotonic(), 0.0)
+
+
+def _well_location(x: float, y: float, z: float, origin: str = "bottom") -> Dict[str, Any]:
+  """The ``wellLocation`` shape every well-addressed command takes."""
+  return {"origin": origin, "offset": {"x": x, "y": y, "z": z}}
+
+
+def _in_place(
+  volume: float,
+  flow_rate: float,
+  pipette_id: str,
+  push_out: Optional[bool] = None,
+) -> Dict[str, Any]:
+  """Params for an ``aspirateInPlace``/``dispenseInPlace`` command."""
+  params: Dict[str, Any] = {"flowRate": flow_rate, "volume": volume, "pipetteId": pipette_id}
+  if push_out is not None:
+    params["pushOut"] = push_out
+  return params
+
+
+async def _call_off_loop(call: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+  """Run a blocking call on a thread of its own and await its result.
+
+  Not ``asyncio.to_thread``: that borrows the loop's shared executor, whose threads
+  ``asyncio.run()`` joins on the way out, so one request nobody ever answers would
+  hang the process at shutdown and burn a pool slot every other backend shares. A
+  daemon thread holds neither.
+
+  A thread per call, not a queue of one, and that is deliberate. ``ot_api`` gives
+  ``urlopen`` no socket timeout, so a request the robot never answers leaks its
+  thread for the life of the process. Queueing behind it would mean the stop that
+  contains a timed-out move could never reach the robot either, which is the worse
+  failure: the robot would keep executing what we stopped waiting for. The cost is
+  that after a timeout a second call can be in flight against the same robot while
+  the first is still stuck.
+  """
+  loop = asyncio.get_running_loop()
+  future: "asyncio.Future[Any]" = loop.create_future()
+
+  def _settle(setter: Callable[[Any], None], value: Any) -> None:
+    if not future.done():
+      setter(value)
+
+  def _deliver(setter: Callable[[Any], None], value: Any) -> None:
+    try:
+      loop.call_soon_threadsafe(_settle, setter, value)
+    except RuntimeError:
+      pass  # the loop is gone, so nobody is waiting for this answer any more
+
+  def _run() -> None:
+    try:
+      result = call(*args, **kwargs)
+    except BaseException as exc:
+      _deliver(future.set_exception, exc)
+    else:
+      _deliver(future.set_result, result)
+
+  threading.Thread(target=_run, name="opentrons-request", daemon=True).start()
+  return await future
+
 
 class _IOLogger:
   """Transparent proxy over the ``ot_api`` module that logs every call at
@@ -73,6 +149,9 @@ def _logged(*args, **kwargs):
         logger.log(LOG_LEVEL_IO, "%s(%s)", qualified, ", ".join(parts))
         return attr(*args, **kwargs)
 
+      # Without this every wrapped call answers to "_logged", and anything that
+      # names the call it is reporting on (a timeout message) names the proxy.
+      _logged.__name__ = _logged.__qualname__ = qualified
       return _logged
     return attr
 
@@ -97,9 +176,27 @@ class OpentronsOT2Backend(LiquidHandlerBackend):
     "p1000_single_gen3": 1000,
   }
 
-  def __init__(self, host: str, port: int = 31950):
+  def __init__(
+    self,
+    host: str,
+    port: int = 31950,
+    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
+    command_timeout: float = DEFAULT_COMMAND_TIMEOUT,
+    status_poll_interval: float = DEFAULT_STATUS_POLL_INTERVAL,
+  ):
+    """Args:
+    host: the robot's address.
+    port: the robot-server's port.
+    request_timeout: how long one request/response with the robot may take, in
+      seconds. Includes the wait for whatever request is already in flight.
+    command_timeout: how long a command that moves the robot may take, in seconds.
+      Covers the motion itself, not just the request that started it.
+    status_poll_interval: delay between two reads of a running command's status.
+    """
     super().__init__()
 
+    self._init_wire_state(request_timeout, command_timeout, status_poll_interval)
+
     if not USE_OT:
       raise RuntimeError(
         "Opentrons is not installed. Please run pip install pylabrobot[opentrons]."
@@ -125,6 +222,34 @@ def __init__(self, host: str, port: int = 31950):
     self._tip_racks: Dict[str, int] = {}  # tip_rack.name -> slot index
     self._plr_name_to_load_name: Dict[str, str] = {}
 
+  def _init_wire_state(
+    self,
+    request_timeout: float,
+    command_timeout: float,
+    status_poll_interval: float,
+  ) -> None:
+    """Check and record the three budgets, and the state the wire layer keeps.
+
+    Shared with the chatterbox, which skips this ``__init__`` because it has no
+    ``ot_api`` to talk to and would otherwise drift from what the wire layer expects.
+    """
+    for name, value in (
+      ("request_timeout", request_timeout),
+      ("command_timeout", command_timeout),
+      ("status_poll_interval", status_poll_interval),
+    ):
+      if value <= 0:
+        raise ValueError(f"{name} must be greater than 0, got {value}")
+    self.request_timeout = request_timeout
+    self.command_timeout = command_timeout
+    self.status_poll_interval = status_poll_interval
+    # Built on first use, not here: on 3.9 a Lock binds to whatever loop is current
+    # when it is constructed, and a backend is routinely built before asyncio.run().
+    self._request_lock: Optional[asyncio.Lock] = None
+    # Set by a command timeout: the robot is still holding a command we stopped
+    # waiting for, so its pose is no longer ours to describe.
+    self._robot_state_unknown = False
+
   def serialize(self) -> dict:
     return {
       **super().serialize(),
@@ -132,18 +257,175 @@ def serialize(self) -> dict:
       "port": self.port,
     }
 
+  async def _request(
+    self,
+    call: Callable[..., Any],
+    *args: Any,
+    timeout: Optional[float] = None,
+    **kwargs: Any,
+  ) -> Any:
+    """Issue one ``ot_api`` call off the event loop, and give up after ``timeout``.
+
+    ``ot_api`` reaches the robot with ``urlopen`` and passes it no socket timeout, so
+    an unanswered request blocks its thread for good. Running it off the loop keeps
+    the rest of the process going, and the wait_for ends OUR wait: a thread cannot be
+    cancelled, so the abandoned one finishes on its own.
+
+    The budget covers the wait for the lock as well as the request. A caller told a
+    read is bounded at seven seconds must not sit behind someone else's mix for ten
+    minutes first, so the clock starts before the queue, not after it.
+    """
+
+    budget = self.request_timeout if timeout is None else timeout
+    try:
+      return await asyncio.wait_for(self._locked_call(call, *args, **kwargs), timeout=budget)
+    except asyncio.TimeoutError as exc:
+      # Before 3.11 asyncio.TimeoutError is not builtins.TimeoutError, so re-raising
+      # is what gives this backend one give-up type on every supported version.
+      raise TimeoutError(
+        f"{getattr(call, '__name__', call)} did not answer within {budget:g}s"
+      ) from exc
+
+  async def _locked_call(self, call: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
+    """One ``ot_api`` call at a time: the robot runs one queue and ``ot_api`` keeps the
+    run id in a module global.
+
+    Per call, not per command. The blocking calls this replaced held the event loop for
+    a whole command; here the lock is released between an enqueue and its polls, so
+    keeping two commands off one robot is the caller's job.
+    """
+    if self._request_lock is None:
+      self._request_lock = asyncio.Lock()
+    async with self._request_lock:
+      return await _call_off_loop(call, *args, **kwargs)
+
+  async def _command(
+    self,
+    command_type: str,
+    params: Dict[str, Any],
+    timeout: Optional[float] = None,
+    abandon_run_on_timeout: bool = True,
+  ) -> Dict[str, Any]:
+    """Enqueue one run command and wait for the robot to finish it.
+
+    ``ot_api``'s own command wrappers are not used for anything that moves: their
+    decorator hard-codes a 30s ceiling no caller can raise, and polls with no delay
+    between reads. Enqueue-and-poll here honours ``command_timeout`` instead.
+
+    ``timeout`` bounds the waiting, not the whole call: the enqueue and each status
+    read carry a request budget of their own, so one unanswered request can carry the
+    call past its deadline by up to ``request_timeout``. A read that does not answer
+    is retried until the deadline: one lost GET says nothing about the motion, and
+    ending a ten-minute mix at the eight-second mark would take the abort decision
+    away from whoever asked for ten minutes.
+
+    Set ``abandon_run_on_timeout`` False for a command that moves nothing. Giving up
+    on one of those leaves no motion outstanding, so halting the robot and refusing
+    everything after it would cost more than the timeout did.
+    """
+
+    self._refuse_if_state_unknown()
+    budget = self.command_timeout if timeout is None else timeout
+    deadline = time.monotonic() + budget
+    try:
+      command_id = await self._request(
+        self._ot.runs.enqueue_command,
+        command_type,
+        params,
+        intent="setup",
+        timeout=min(self.request_timeout, _seconds_left(deadline)),
+      )
+      while True:
+        # A read is a request, so it gets a request budget. Handing it whatever is
+        # left of the deadline gives it milliseconds it cannot answer in, and then a
+        # command the robot finished reads as a timeout.
+        try:
+          result: Dict[str, Any] = await self._request(self._ot.runs.get_command, command_id)
+        except TimeoutError as exc:
+          remaining = _seconds_left(deadline)
+          if remaining <= 0:
+            raise TimeoutError(f"{command_type} did not finish within {budget:g}s") from exc
+          logger.warning(
+            "status read for %s did not answer; %.0fs of its budget left", command_type, remaining
+          )
+          await asyncio.sleep(min(self.status_poll_interval, remaining))
+          continue
+        data = result["data"]
+        status = data["status"]
+        if status == "failed":
+          error = data["error"]
+          raise RuntimeError(f"{command_type} failed with {error['errorType']}: {error['detail']}")
+        if status not in ("queued", "running"):
+          return result
+        # The deadline is checked after the read, so the read that follows the last
+        # sleep still happens: that is the one that sees a command which finished
+        # while we were asleep.
+        remaining = _seconds_left(deadline)
+        if remaining <= 0:
+          raise TimeoutError(f"{command_type} did not finish within {budget:g}s")
+        await asyncio.sleep(min(self.status_poll_interval, remaining))
+    except TimeoutError:
+      if abandon_run_on_timeout:
+        await self._abandon_run()
+      raise
+
+  def _refuse_if_state_unknown(self) -> None:
+    if self._robot_state_unknown:
+      raise RuntimeError(
+        "A command timed out and the OT-2 was left holding it, so its pose is "
+        "unknown. Recover with setup(), which starts a fresh run the stale command "
+        "cannot execute in, and homes. Check the pipettes by eye first: the OT-2 has "
+        "no tip sensor, ending a run does not drop tips, and setup() records both "
+        "mounts as empty, so a tip left on will be pressed into the rack."
+      )
+
+  async def _abandon_run(self) -> None:
+    """Stop the run a timed-out command is still sitting in, and refuse the next one.
+
+    Giving up on the wait does not take the command out of the robot's queue: it will
+    still execute, so a caller who retries makes the robot aspirate twice from one
+    well. The stop action is what prevents that, but the robot-server only schedules
+    it and answers 201 straight away, so nothing here can confirm the run halted. The
+    refusal is the part that holds: it stands until ``setup()`` builds a new run,
+    which the stale command cannot execute in whatever the old one did.
+    """
+    self._robot_state_unknown = True
+    run_id = getattr(self._ot, "run_id", None)
+    if not run_id:
+      return
+    try:
+      await self._stop_run(run_id)
+    except Exception:
+      logger.warning("could not stop run %s after a command timed out", run_id, exc_info=True)
+
+  async def _stop_run(self, run_id: str) -> None:
+    """Halt a run, so nothing still queued in it executes."""
+    await self._request(
+      self._ot.requestor.post,
+      f"/runs/{run_id}/actions",
+      {"data": {"actionType": "stop"}},
+    )
+
   async def setup(self, skip_home: bool = False):
     # create run
-    run_id = self._ot.runs.create()
+    run_id = await self._request(self._ot.runs.create)
     self._ot.set_run(run_id)
 
-    # get pipettes, then assign them
-    self.left_pipette, self.right_pipette = self._ot.lh.add_mounted_pipettes()
+    # Only now is the unknown-state refusal lifted. Creating the run is what orphans
+    # whatever an earlier timeout left queued, and it is also the step most likely to
+    # fail here: the robot-server answers 409 while it still holds the old run.
+    self._robot_state_unknown = False
+
+    # get pipettes, then assign them. This reads /pipettes and then loads each one,
+    # so it needs the command budget rather than a single request's.
+    self.left_pipette, self.right_pipette = await self._request(
+      self._ot.lh.add_mounted_pipettes, timeout=self.command_timeout
+    )
 
     self.left_pipette_has_tip = self.right_pipette_has_tip = False
 
     # get api version
-    health = self._ot.health.get()
+    health = await self._request(self._ot.health.get)
     self.ot_api_version = health["api_version"]
 
     if not skip_home:
@@ -160,19 +442,18 @@ async def stop(self):
     self.left_pipette = None
     self.right_pipette = None
 
-    # cancel the HTTP-API run if it exists (helpful to make device available again in official Opentrons app)
+    # release the run so the official Opentrons app can drive the robot again. Halt
+    # first: deleting a run the robot is still working through leaves it working.
     run_id = getattr(self._ot, "run_id", None)
     if run_id:
       try:
-        self._ot.requestor.post(f"/runs/{run_id}/cancel")
+        await self._stop_run(run_id)
       except Exception:
-        try:
-          self._ot.requestor.post(f"/runs/{run_id}/actions/cancel")
-        except Exception:
-          try:
-            self._ot.requestor.delete(f"/runs/{run_id}")
-          except Exception:
-            pass
+        logger.warning("could not stop run %s", run_id, exc_info=True)
+      try:
+        await self._request(self._ot.requestor.delete, f"/runs/{run_id}")
+      except Exception:
+        logger.warning("could not delete run %s", run_id, exc_info=True)
 
   def get_ot_name(self, plr_resource_name: str) -> str:
     """Opentrons only allows names in ^[a-z0-9._]+$, but in PLR we are flexible.
@@ -268,7 +549,7 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip):
       ],
     }
 
-    data = self._ot.labware.define(lw)
+    data = await self._request(self._ot.labware.define, lw)
     namespace, definition, version = data["data"]["definitionUri"].split("/")
 
     # assign labware to robot
@@ -281,13 +562,16 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip):
     slot = deck.get_slot(tip_rack)
     assert slot is not None, "tip rack must be on deck"
 
-    self._ot.labware.add(
-      load_name=definition,
-      namespace=namespace,
-      ot_location=slot,
-      version=version,
-      labware_id=labware_uuid,
-      display_name=self.get_ot_name(tip_rack.name),
+    await self._command(
+      "loadLabware",
+      {
+        "location": {"slotName": str(slot)},
+        "loadName": definition,
+        "namespace": namespace,
+        "version": version,
+        "labwareId": labware_uuid,
+        "displayName": self.get_ot_name(tip_rack.name),
+      },
     )
 
     self._tip_racks[tip_rack.name] = slot
@@ -360,13 +644,14 @@ async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]):
 
     offset_z += op.tip.total_tip_length
 
-    self._ot.lh.pick_up_tip(
-      labware_id=self.get_ot_name(tip_rack.name),
-      well_name=self.get_ot_name(op.resource.name),
-      pipette_id=pipette_id,
-      offset_x=offset_x,
-      offset_y=offset_y,
-      offset_z=offset_z,
+    await self._command(
+      "pickUpTip",
+      {
+        "labwareId": self.get_ot_name(tip_rack.name),
+        "wellName": self.get_ot_name(op.resource.name),
+        "wellLocation": _well_location(offset_x, offset_y, offset_z),
+        "pipetteId": pipette_id,
+      },
     )
 
     self._set_tip_state(pipette_id, True)
@@ -400,21 +685,26 @@ async def drop_tips(self, ops: List[Drop], use_channels: List[int]):
     offset_z += 10
 
     if use_fixed_trash:
-      self._ot.lh.move_to_addressable_area_for_drop_tip(
-        pipette_id=pipette_id,
-        offset_x=offset_x,
-        offset_y=offset_y,
-        offset_z=offset_z,
+      await self._command(
+        "moveToAddressableAreaForDropTip",
+        {
+          "pipetteId": pipette_id,
+          "addressableAreaName": "fixedTrash",
+          "wellName": "A1",
+          "wellLocation": _well_location(offset_x, offset_y, offset_z, origin="default"),
+          "alternateDropLocation": False,
+        },
       )
-      self._ot.lh.drop_tip_in_place(pipette_id=pipette_id)
+      await self._command("dropTipInPlace", {"pipetteId": pipette_id})
     else:
-      self._ot.lh.drop_tip(
-        labware_id,
-        well_name=self.get_ot_name(op.resource.name),
-        pipette_id=pipette_id,
-        offset_x=offset_x,
-        offset_y=offset_y,
-        offset_z=offset_z,
+      await self._command(
+        "dropTip",
+        {
+          "labwareId": labware_id,
+          "wellName": self.get_ot_name(op.resource.name),
+          "wellLocation": _well_location(offset_x, offset_y, offset_z),
+          "pipetteId": pipette_id,
+        },
       )
 
     self._set_tip_state(pipette_id, False)
@@ -509,22 +799,14 @@ async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[
 
     if op.mix is not None:
       for _ in range(op.mix.repetitions):
-        self._ot.lh.aspirate_in_place(
-          volume=op.mix.volume,
-          flow_rate=op.mix.flow_rate,
-          pipette_id=pipette_id,
+        await self._command(
+          "aspirateInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id)
         )
-        self._ot.lh.dispense_in_place(
-          volume=op.mix.volume,
-          flow_rate=op.mix.flow_rate,
-          pipette_id=pipette_id,
+        await self._command(
+          "dispenseInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id, push_out=False)
         )
 
-    self._ot.lh.aspirate_in_place(
-      volume=volume,
-      flow_rate=flow_rate,
-      pipette_id=pipette_id,
-    )
+    await self._command("aspirateInPlace", _in_place(volume, flow_rate, pipette_id))
 
     traversal_location = self._deck_to_robot_frame(
       op.resource.get_location_wrt(self.deck, "c", "c", "cavity_bottom") + op.offset
@@ -581,23 +863,15 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in
       pipette_id=pipette_id,
     )
 
-    self._ot.lh.dispense_in_place(
-      volume=volume,
-      flow_rate=flow_rate,
-      pipette_id=pipette_id,
-    )
+    await self._command("dispenseInPlace", _in_place(volume, flow_rate, pipette_id, push_out=False))
 
     if op.mix is not None:
       for _ in range(op.mix.repetitions):
-        self._ot.lh.aspirate_in_place(
-          volume=op.mix.volume,
-          flow_rate=op.mix.flow_rate,
-          pipette_id=pipette_id,
+        await self._command(
+          "aspirateInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id)
         )
-        self._ot.lh.dispense_in_place(
-          volume=op.mix.volume,
-          flow_rate=op.mix.flow_rate,
-          pipette_id=pipette_id,
+        await self._command(
+          "dispenseInPlace", _in_place(op.mix.volume, op.mix.flow_rate, pipette_id, push_out=False)
         )
 
     traversal_location = self._deck_to_robot_frame(
@@ -611,7 +885,7 @@ async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[in
     )
 
   async def home(self):
-    self._ot.health.home()
+    await self._request(self._ot.health.home, timeout=self.command_timeout)
 
   async def pick_up_tips96(self, pickup: PickupTipRack):
     raise NotImplementedError("The Opentrons backend does not support the 96 head.")
@@ -638,7 +912,7 @@ async def drop_resource(self, drop: ResourceDrop):
 
   async def list_connected_modules(self) -> List[dict]:
     """List all connected temperature modules."""
-    return cast(List[dict], self._ot.modules.list_connected_modules())
+    return cast(List[dict], await self._request(self._ot.modules.list_connected_modules))
 
   def _pipette_id_for_channel(self, channel: int) -> str:
     pipettes = []
@@ -650,16 +924,30 @@ def _pipette_id_for_channel(self, channel: int) -> str:
       raise NoChannelError(f"Channel {channel} not available on this OT-2 setup.")
     return pipettes[channel]
 
-  def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]:
+  async def _save_position(self, pipette_id: str) -> Dict[str, Any]:
+    """Ask the robot where a pipette is, and wait for the answer.
+
+    A read rather than a move, so it gets the request budget rather than the command
+    one: nothing here waits on motion.
+    """
+
+    return await self._command(
+      "savePosition",
+      {"pipetteId": pipette_id},
+      timeout=self.request_timeout,
+      abandon_run_on_timeout=False,
+    )
+
+  async def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]:
     """Return the pipette id and current coordinate for a given channel."""
 
     pipette_id = self._pipette_id_for_channel(channel)
     try:
-      res = self._ot.lh.save_position(pipette_id=pipette_id)
+      res = await self._save_position(pipette_id)
       pos = res["data"]["result"]["position"]
       current = Coordinate(pos["x"], pos["y"], pos["z"])
     except Exception as exc:
-      raise RuntimeError("Failed to query current pipette position") from exc
+      raise RuntimeError(f"Failed to query current pipette position: {exc}") from exc
 
     return pipette_id, current
 
@@ -668,10 +956,41 @@ async def prepare_for_manual_channel_operation(self, channel: int):
 
     _ = self._pipette_id_for_channel(channel)
 
+  async def get_channel_position(self, channel: int) -> Coordinate:
+    """Where a channel is right now, in the OT-2 robot frame (this file's own name
+    for the frame ``_deck_to_robot_frame`` converts PLR deck coordinates into)."""
+
+    _, current = await self._current_channel_position(channel)
+    return current
+
+  async def move_channel_to(
+    self,
+    channel: int,
+    x: Optional[float] = None,
+    y: Optional[float] = None,
+    z: Optional[float] = None,
+  ):
+    """Move a channel to an absolute position, holding the axes left out.
+
+    One coordinated move rather than the per-axis calls chained: the robot lifts to the traversal
+    height and travels once, where three separate moves each descend and can clip labware between
+    them.
+    """
+
+    pipette_id, current = await self._current_channel_position(channel)
+    target = Coordinate(
+      x=current.x if x is None else x,
+      y=current.y if y is None else y,
+      z=current.z if z is None else z,
+    )
+    await self.move_pipette_head(
+      location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id
+    )
+
   async def move_channel_x(self, channel: int, x: float):
     """Move a channel to an absolute x coordinate using savePosition to seed pose."""
 
-    pipette_id, current = self._current_channel_position(channel)
+    pipette_id, current = await self._current_channel_position(channel)
     target = Coordinate(x=x, y=current.y, z=current.z)
     await self.move_pipette_head(
       location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id
@@ -680,7 +999,7 @@ async def move_channel_x(self, channel: int, x: float):
   async def move_channel_y(self, channel: int, y: float):
     """Move a channel to an absolute y coordinate using savePosition to seed pose."""
 
-    pipette_id, current = self._current_channel_position(channel)
+    pipette_id, current = await self._current_channel_position(channel)
     target = Coordinate(x=current.x, y=y, z=current.z)
     await self.move_pipette_head(
       location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id
@@ -689,7 +1008,7 @@ async def move_channel_y(self, channel: int, y: float):
   async def move_channel_z(self, channel: int, z: float):
     """Move a channel to an absolute z coordinate using savePosition to seed pose."""
 
-    pipette_id, current = self._current_channel_position(channel)
+    pipette_id, current = await self._current_channel_position(channel)
     target = Coordinate(x=current.x, y=current.y, z=z)
     await self.move_pipette_head(
       location=target, minimum_z_height=self.traversal_height, pipette_id=pipette_id
@@ -724,15 +1043,16 @@ async def move_pipette_head(
     if pipette_id is None:
       raise ValueError("No pipette id given or left/right pipette not available.")
 
-    self._ot.lh.move_arm(
-      pipette_id=pipette_id,
-      location_x=location.x,
-      location_y=location.y,
-      location_z=location.z,
-      minimum_z_height=minimum_z_height,
-      speed=speed,
-      force_direct=force_direct,
-    )
+    params: Dict[str, Any] = {
+      "pipetteId": pipette_id,
+      "coordinates": {"x": location.x, "y": location.y, "z": location.z},
+      "forceDirect": force_direct,
+    }
+    if minimum_z_height is not None:
+      params["minimumZHeight"] = minimum_z_height
+    if speed is not None:
+      params["speed"] = speed
+    await self._command("moveToCoordinates", params)
 
   def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool:
     def supports_tip(channel_vol: float, tip_vol: float) -> bool:
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py
index 7f753803de8..df65364e0f7 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py
@@ -1,4 +1,8 @@
+import asyncio
+import threading
+import time
 import unittest
+from typing import Any, Dict, List, Optional, Tuple
 from unittest.mock import patch
 
 import pytest
@@ -21,46 +25,67 @@
 from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul
 from pylabrobot.resources.well import Well
 
+_PIPETTES = (
+  {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"},
+  {"pipetteId": "right-pipette-id", "name": "p20_single_gen2"},
+)
+
 
 def _mock_define(lw):
   return {"data": {"definitionUri": f'lw["namespace"]/{lw["metadata"]["displayName"]}/1'}}
 
 
-def _mock_add(load_name, namespace, ot_location, version, labware_id, display_name):
-  return labware_id
-
-
 def _mock_health_get():
   return {
     "api_version": "7.0.1",
   }
 
 
+class _FakeRobot:
+  """Stands in for the robot-server's command queue.
+
+  Records what the backend enqueues and answers every poll "succeeded", so a test
+  can read the exact wire params the robot would have been handed.
+  """
+
+  def __init__(self, position: Tuple[float, float, float] = (0.0, 0.0, 0.0)) -> None:
+    self.commands: List[Tuple[str, Dict[str, Any]]] = []
+    self.position = position
+    self.error: Optional[Dict[str, str]] = None
+
+  def enqueue_command(self, command_type, params, intent="setup", **kwargs) -> str:
+    self.commands.append((command_type, dict(params)))
+    return f"cmd-{len(self.commands)}"
+
+  def get_command(self, command_id, **kwargs) -> Dict[str, Any]:
+    if self.error is not None:
+      return {"data": {"status": "failed", "error": self.error}}
+    x, y, z = self.position
+    return {"data": {"status": "succeeded", "result": {"position": {"x": x, "y": y, "z": z}}}}
+
+  def command_types(self) -> List[str]:
+    return [command_type for command_type, _params in self.commands]
+
+  def params(self, command_type: str) -> Dict[str, Any]:
+    return dict(self.commands)[command_type]
+
+
 class OpentronsBackendSetupTests(unittest.IsolatedAsyncioTestCase):
   """Tests for setup and stop"""
 
   @patch("ot_api.runs.create")
   @patch("ot_api.health.home")
   @patch("ot_api.lh.add_mounted_pipettes")
-  @patch("ot_api.labware.add")
-  @patch("ot_api.labware.define")
   @patch("ot_api.health.get")
   async def test_setup(
     self,
     mock_health_get,
-    mock_define,
-    mock_add,
     mock_add_mounted_pipettes,
     mock_home,
     mock_create,
   ):
     mock_create.return_value = "run-id"
-    mock_add_mounted_pipettes.return_value = (
-      {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"},
-      {"pipetteId": "right-pipette-id", "name": "p20_single_gen2"},
-    )
-    mock_add.side_effect = _mock_add
-    mock_define.side_effect = _mock_define
+    mock_add_mounted_pipettes.return_value = _PIPETTES
     mock_health_get.side_effect = _mock_health_get
 
     self.backend = OpentronsOT2Backend(host="localhost", port=1338)
@@ -78,33 +103,40 @@ def test_serialize(self):
       "OpentronsOT2Backend",
     )
 
+  def test_a_budget_of_zero_or_less_is_refused(self):
+    """A zero poll interval spins the robot-server; a zero command budget never polls."""
+    for kwargs in (
+      {"request_timeout": 0.0},
+      {"command_timeout": -1.0},
+      {"status_poll_interval": 0.0},
+    ):
+      with self.subTest(**kwargs):
+        with self.assertRaises(ValueError):
+          OpentronsOT2Backend(host="localhost", port=1338, **kwargs)
 
-class OpentronsBackendCommandTests(unittest.IsolatedAsyncioTestCase):
-  """Tests Opentrons commands"""
 
-  @patch("ot_api.runs.create")
-  @patch("ot_api.health.home")
-  @patch("ot_api.lh.add_mounted_pipettes")
-  @patch("ot_api.labware.add")
-  @patch("ot_api.labware.define")
-  @patch("ot_api.health.get")
-  async def asyncSetUp(
-    self,
-    mock_health_get,
-    mock_define,
-    mock_add,
-    mock_add_mounted_pipettes,
-    mock_home,
-    mock_create,
-  ):
-    mock_add.side_effect = _mock_add
-    mock_define.side_effect = _mock_define
-    mock_add_mounted_pipettes.return_value = (
-      {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"},
-      {"pipetteId": "right-pipette-id", "name": "p20_single_gen2"},
-    )
-    mock_create.return_value = "run-id"
-    mock_health_get.side_effect = _mock_health_get
+class OpentronsBackendCommandTests(unittest.IsolatedAsyncioTestCase):
+  """Tests Opentrons commands.
+
+  The robot is faked at the command queue (``runs.enqueue_command`` /
+  ``runs.get_command``), which is where the backend actually reaches it, so every
+  assertion here is about the params a real robot would receive.
+  """
+
+  async def asyncSetUp(self):
+    self.robot = _FakeRobot()
+    for target, kwargs in (
+      ("ot_api.runs.create", {"return_value": "run-id"}),
+      ("ot_api.runs.enqueue_command", {"side_effect": self.robot.enqueue_command}),
+      ("ot_api.runs.get_command", {"side_effect": self.robot.get_command}),
+      ("ot_api.lh.add_mounted_pipettes", {"return_value": _PIPETTES}),
+      ("ot_api.health.get", {"side_effect": _mock_health_get}),
+      ("ot_api.health.home", {}),
+      ("ot_api.labware.define", {"side_effect": _mock_define}),
+    ):
+      patcher = patch(target, **kwargs)
+      patcher.start()
+      self.addCleanup(patcher.stop)
 
     self.backend = OpentronsOT2Backend(host="localhost", port=1338)
     self.deck = OTDeck()
@@ -115,132 +147,156 @@ async def asyncSetUp(
     self.deck.assign_child_at_slot(self.tip_rack, slot=1)
     self.plate = celltreat_96_wellplate_350uL_Fb(name="plate")
     self.deck.assign_child_at_slot(self.plate, slot=11)
+    self.robot.commands.clear()
 
-  @patch("ot_api.lh.pick_up_tip")
-  @patch("ot_api.labware.define")
-  @patch("ot_api.labware.add")
-  async def test_tip_pick_up(self, mock_add=None, mock_define=None, mock_pick_up_tip=None):
-    assert mock_pick_up_tip is not None and mock_define is not None and mock_add is not None
-    mock_define.side_effect = _mock_define
-    mock_add.side_effect = _mock_add
+  async def test_tip_pick_up(self):
+    await self.lh.pick_up_tips(self.tip_rack["A1"])
 
-    def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, offset_z):
-      self.assertEqual(labware_id, self.backend.get_ot_name("tip_rack"))
-      self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1"))
-      self.assertEqual(pipette_id, "left-pipette-id")
-      self.assertEqual(offset_x, offset_x)
-      self.assertEqual(offset_y, offset_y)
-      self.assertEqual(offset_z, offset_z)
+    params = self.robot.params("pickUpTip")
+    self.assertEqual(params["labwareId"], self.backend.get_ot_name("tip_rack"))
+    self.assertEqual(
+      params["wellName"], self.backend.get_ot_name(self.tip_rack.get_item("A1").name)
+    )
+    self.assertEqual(params["pipetteId"], "left-pipette-id")
+    self.assertEqual(params["wellLocation"]["origin"], "bottom")
+
+  async def test_a_tip_rack_is_loaded_into_the_run_before_its_first_pickup(self):
+    await self.lh.pick_up_tips(self.tip_rack["A1"])
 
-    mock_pick_up_tip.side_effect = assert_parameters
+    params = self.robot.params("loadLabware")
+    self.assertEqual(params["location"], {"slotName": "1"})
+    self.assertEqual(params["labwareId"], self.backend.get_ot_name("tip_rack"))
+    self.assertLess(
+      self.robot.command_types().index("loadLabware"),
+      self.robot.command_types().index("pickUpTip"),
+    )
 
+  async def test_tip_drop(self):
     await self.lh.pick_up_tips(self.tip_rack["A1"])
+    await self.lh.drop_tips(self.tip_rack["A1"])
 
-  @patch("ot_api.lh.drop_tip")
-  async def test_tip_drop(self, mock_drop_tip):
-    def assert_parameters(labware_id, well_name, pipette_id, offset_x, offset_y, offset_z):
-      self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1"))
-      self.assertEqual(well_name, self.backend.get_ot_name("tip_rack_A1"))
-      self.assertEqual(pipette_id, "left-pipette-id")
-      self.assertEqual(offset_x, offset_x)
-      self.assertEqual(offset_y, offset_y)
-      self.assertEqual(offset_z, offset_z)
+    params = self.robot.params("dropTip")
+    self.assertEqual(
+      params["wellName"], self.backend.get_ot_name(self.tip_rack.get_item("A1").name)
+    )
+    self.assertEqual(params["pipetteId"], "left-pipette-id")
 
-    mock_drop_tip.side_effect = assert_parameters
+  async def test_get_channel_position_asks_the_robot_to_save_its_position(self):
+    self.robot.position = (11.0, 22.0, 33.0)
 
-    await self.test_tip_pick_up()
-    await self.lh.drop_tips(self.tip_rack["A1"])
+    position = await self.backend.get_channel_position(0)
 
-  @patch("ot_api.lh.aspirate_in_place")
-  @patch("ot_api.lh.move_arm")
-  async def test_aspirate(self, mock_move=None, mock_aspirate=None):
-    assert mock_aspirate is not None and mock_move is not None
+    self.assertEqual(position, Coordinate(11.0, 22.0, 33.0))
+    self.assertEqual(self.robot.params("savePosition"), {"pipetteId": "left-pipette-id"})
 
-    def assert_parameters(
-      volume,
-      flow_rate,
-      pipette_id,
-    ):
-      self.assertEqual(pipette_id, "left-pipette-id")
-      self.assertEqual(volume, 10)
-      self.assertEqual(flow_rate, 3.78)
+  async def test_a_failed_save_position_names_the_robot_error(self):
+    self.robot.error = {"errorType": "MustHomeError", "detail": "Must home first"}
+
+    with self.assertRaises(RuntimeError) as caught:
+      await self.backend.get_channel_position(0)
+
+    self.assertIn("MustHomeError", str(caught.exception))
+
+  async def test_move_channel_to_travels_once_holding_the_axes_left_out(self):
+    self.robot.position = (11.0, 22.0, 33.0)
+
+    await self.backend.move_channel_to(0, x=50.0, z=5.0)
 
-    mock_aspirate.side_effect = assert_parameters
+    self.assertEqual(self.robot.command_types().count("moveToCoordinates"), 1)
+    params = self.robot.params("moveToCoordinates")
+    self.assertEqual(params["coordinates"]["x"], 50.0)
+    self.assertEqual(params["coordinates"]["y"], 22.0)  # not named, so held
+    self.assertEqual(params["coordinates"]["z"], 5.0)
+    self.assertEqual(params["minimumZHeight"], self.backend.traversal_height)
 
-    await self.test_tip_pick_up()
+  async def test_aspirate(self):
+    await self.lh.pick_up_tips(self.tip_rack["A1"])
     self.plate.get_well("A1").tracker.set_volume(10)
-    await self.lh.aspirate(self.plate["A1"], vols=[10])
 
-  @patch("ot_api.lh.dispense_in_place")
-  @patch("ot_api.lh.move_arm")
-  async def test_dispense(self, mock_move, mock_dispense):
-    def assert_parameters(
-      volume,
-      flow_rate,
-      pipette_id,
-    ):
-      self.assertEqual(pipette_id, "left-pipette-id")
-      self.assertEqual(volume, 10)
-      self.assertEqual(flow_rate, 7.56)
+    await self.lh.aspirate(self.plate["A1"], vols=[10])
 
-    mock_dispense.side_effect = assert_parameters
+    self.assertEqual(
+      self.robot.params("aspirateInPlace"),
+      {"flowRate": 3.78, "volume": 10, "pipetteId": "left-pipette-id"},
+    )
 
-    await self.test_aspirate()  # aspirate first
+  async def test_dispense(self):
+    await self.test_aspirate()
     with no_volume_tracking():
       await self.lh.dispense(self.plate["A1"], vols=[10])
 
-  # -- characterization of the remaining ot_api call sites (Phase 0 safety net) --
+    self.assertEqual(
+      self.robot.params("dispenseInPlace"),
+      {"flowRate": 7.56, "volume": 10, "pipetteId": "left-pipette-id", "pushOut": False},
+    )
 
-  @patch("ot_api.health.home")
-  async def test_home_calls_health_home(self, mock_home):
+  async def test_a_motion_command_never_goes_through_ot_apis_capped_wrapper(self):
+    """``ot_api``'s ``@command`` decorator hard-codes a 30s ceiling and takes no
+    ``timeout`` kwarg, so anything routed through it ignores ``command_timeout``."""
+    with patch("ot_api.lh.aspirate_in_place") as capped_wrapper:
+      await self.lh.pick_up_tips(self.tip_rack["A1"])
+      self.plate.get_well("A1").tracker.set_volume(10)
+      await self.lh.aspirate(self.plate["A1"], vols=[10])
+
+    capped_wrapper.assert_not_called()
+    self.assertIn("aspirateInPlace", self.robot.command_types())
+
+  async def test_a_command_may_outlast_a_single_requests_budget(self):
+    """The request budget bounds one exchange; the command budget bounds the motion.
+    A move that inherited the request budget could never wait out a real motion."""
+    answers = ["running", "running", "running", "succeeded"]
+
+    def still_moving(command_id, **kwargs):
+      return {"data": {"status": answers.pop(0), "result": {}}}
+
+    self.backend.request_timeout = 0.05
+    self.backend.command_timeout = 5.0
+    self.backend.status_poll_interval = 0.05
+    with patch("ot_api.runs.get_command", side_effect=still_moving):
+      started = time.monotonic()
+      await self.backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+
+    self.assertGreater(time.monotonic() - started, self.backend.request_timeout)
+    self.assertEqual(answers, [])
+
+  async def test_home_calls_health_home(self):
     """home() issues exactly one ot_api.health.home() call."""
-    await self.backend.home()
+    with patch("ot_api.health.home") as mock_home:
+      await self.backend.home()
     mock_home.assert_called_once_with()
 
-  @patch("ot_api.modules.list_connected_modules")
-  async def test_list_connected_modules_passthrough(self, mock_modules):
+  async def test_list_connected_modules_passthrough(self):
     """list_connected_modules() returns ot_api.modules.list_connected_modules() verbatim."""
-    mock_modules.return_value = [{"id": "tempdeck"}]
-    result = await self.backend.list_connected_modules()
+    with patch("ot_api.modules.list_connected_modules") as mock_modules:
+      mock_modules.return_value = [{"id": "tempdeck"}]
+      result = await self.backend.list_connected_modules()
     mock_modules.assert_called_once_with()
     self.assertEqual(result, [{"id": "tempdeck"}])
 
   @patch("ot_api.run_id", "run-id", create=True)
-  @patch("ot_api.requestor.post")
-  async def test_stop_cancels_active_run_and_clears_pipettes(self, mock_post):
-    """stop() cancels the active run through the requestor and clears mounted pipettes."""
-    await self.backend.stop()
-    mock_post.assert_called_once_with("/runs/run-id/cancel")
+  async def test_stop_halts_the_run_then_releases_it(self):
+    """stop() halts the run before deleting it: deleting a run the robot is still
+    working through leaves it working."""
+    with patch("ot_api.requestor.post") as mock_post, patch("ot_api.requestor.delete") as mock_del:
+      await self.backend.stop()
+
+    mock_post.assert_called_once_with("/runs/run-id/actions", {"data": {"actionType": "stop"}})
+    mock_del.assert_called_once_with("/runs/run-id")
     self.assertIsNone(self.backend.left_pipette)
     self.assertIsNone(self.backend.right_pipette)
 
-  @patch("ot_api.lh.drop_tip_in_place")
-  @patch("ot_api.lh.move_to_addressable_area_for_drop_tip")
-  @patch("ot_api.lh.drop_tip")
-  @patch("ot_api.lh.pick_up_tip")
-  @patch("ot_api.labware.define")
-  @patch("ot_api.labware.add")
-  async def test_tip_drop_to_trash_uses_addressable_area(
-    self,
-    mock_add,
-    mock_define,
-    mock_pick_up_tip,
-    mock_drop_tip,
-    mock_to_trash,
-    mock_drop_in_place,
-  ):
+  async def test_tip_drop_to_trash_uses_addressable_area(self):
     """At api_version >= 7.1.0 a discard to the deck trash routes via the addressable
-    area (move_to_addressable_area_for_drop_tip + drop_tip_in_place), not drop_tip."""
-    mock_define.side_effect = _mock_define
-    mock_add.side_effect = _mock_add
+    area (moveToAddressableAreaForDropTip + dropTipInPlace), not dropTip."""
     self.backend.ot_api_version = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION
 
     await self.lh.pick_up_tips(self.tip_rack["A1"])
     await self.lh.discard_tips()
 
-    mock_to_trash.assert_called_once()
-    mock_drop_in_place.assert_called_once()
-    mock_drop_tip.assert_not_called()
+    types = self.robot.command_types()
+    self.assertEqual(types.count("moveToAddressableAreaForDropTip"), 1)
+    self.assertEqual(types.count("dropTipInPlace"), 1)
+    self.assertNotIn("dropTip", types)
 
 
 def _make_backend_with_pipettes(left_name="p300_single_gen2", right_name="p20_single_gen2"):
@@ -382,3 +438,349 @@ def test_set_tip_state_right(self):
     self.backend._set_tip_state("right-id", True)
     self.assertFalse(self.backend.left_pipette_has_tip)
     self.assertTrue(self.backend.right_pipette_has_tip)
+
+
+class OpentronsBackendTimeoutTests(unittest.IsolatedAsyncioTestCase):
+  """A robot that stops answering must not hang the process or freeze the event loop.
+
+  ``ot_api`` calls ``urlopen`` with no socket timeout, so an unanswered request blocks
+  its thread for good. These pin that the backend stops waiting on its own, that the
+  rest of the process keeps running while it waits, and that giving up leaves neither
+  the loop nor the robot in a state the next caller can walk into.
+  """
+
+  def _backend(self, **kwargs: float) -> OpentronsOT2Backend:
+    backend = OpentronsOT2Backend(host="localhost", port=1338, **kwargs)
+    backend.left_pipette = {"pipetteId": "left-pipette-id", "name": "p20_single_gen2"}
+    backend.right_pipette = None
+    return backend
+
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  async def test_a_position_read_that_never_answers_fails_at_the_deadline(
+    self, mock_enqueue, mock_get_command
+  ):
+    mock_enqueue.return_value = "cmd-1"
+    released = threading.Event()
+    mock_get_command.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2)
+    started = time.monotonic()
+    try:
+      with self.assertRaises(RuntimeError) as caught:
+        await backend.get_channel_position(0)
+    finally:
+      released.set()  # let the abandoned thread finish, so the suite can exit
+
+    self.assertIsInstance(caught.exception.__cause__, TimeoutError)
+    self.assertLess(time.monotonic() - started, 5.0)
+
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  async def test_a_slow_position_read_leaves_the_event_loop_free(
+    self, mock_enqueue, mock_get_command
+  ):
+    mock_enqueue.return_value = "cmd-1"
+
+    def slow_answer(*args, **kwargs):
+      time.sleep(0.3)
+      return {
+        "data": {"status": "succeeded", "result": {"position": {"x": 1.0, "y": 2.0, "z": 3.0}}}
+      }
+
+    mock_get_command.side_effect = slow_answer
+
+    ticks = 0
+
+    async def tick():
+      nonlocal ticks
+      while True:
+        await asyncio.sleep(0.01)
+        ticks += 1
+
+    ticker = asyncio.ensure_future(tick())
+    try:
+      position = await self._backend().get_channel_position(0)
+    finally:
+      ticker.cancel()
+
+    self.assertEqual(position, Coordinate(1.0, 2.0, 3.0))
+    self.assertGreater(ticks, 1)  # zero or one means the read blocked the loop
+
+  @patch("ot_api.health.home")
+  async def test_a_robot_command_gets_the_longer_budget(self, mock_home):
+    mock_home.side_effect = lambda *a, **kw: time.sleep(0.3)
+
+    # A move outlasts the plain request budget on purpose: it waits on the motion.
+    await self._backend(request_timeout=0.05, command_timeout=5.0).home()
+
+    mock_home.assert_called_once()
+
+  @patch("ot_api.modules.list_connected_modules")
+  async def test_a_plain_read_gets_the_shorter_budget(self, mock_list):
+    released = threading.Event()
+    mock_list.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2, command_timeout=60.0)
+    started = time.monotonic()
+    try:
+      with self.assertRaises(TimeoutError):
+        await backend.list_connected_modules()
+    finally:
+      released.set()
+
+    self.assertLess(time.monotonic() - started, 5.0)
+
+  @patch("ot_api.modules.list_connected_modules")
+  async def test_a_request_queued_behind_another_still_gives_up_at_its_own_budget(self, mock_list):
+    """The budget is what a caller is told a read costs. Starting the clock only after
+    the queue lets a 7s read block for as long as whatever is ahead of it takes."""
+    released = threading.Event()
+    mock_list.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.3)
+    holder = asyncio.ensure_future(backend.list_connected_modules())
+    await asyncio.sleep(0.05)  # let the holder take the lock
+
+    started = time.monotonic()
+    try:
+      with self.assertRaises(TimeoutError):
+        await asyncio.wait_for(backend.list_connected_modules(), timeout=5.0)
+      elapsed = time.monotonic() - started
+    finally:
+      released.set()
+      holder.cancel()
+
+    self.assertLess(elapsed, 2.0)
+
+  @patch("ot_api.modules.list_connected_modules")
+  async def test_an_abandoned_request_does_not_hold_the_loops_default_executor(self, mock_list):
+    """``asyncio.run()`` joins the default executor's threads on the way out, so a
+    request parked there moves the hang from mid-command to shutdown, and burns a
+    pool slot every other backend in the process shares."""
+    released = threading.Event()
+    mock_list.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2)
+    try:
+      with self.assertRaises(TimeoutError):
+        await backend.list_connected_modules()
+      loop = asyncio.get_running_loop()
+      await asyncio.wait_for(loop.shutdown_default_executor(), timeout=3.0)
+    finally:
+      released.set()
+
+  @patch("ot_api.requestor.post")
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  @patch("ot_api.run_id", "run-id", create=True)
+  async def test_a_timed_out_move_stops_the_run_and_refuses_the_next_command(
+    self, mock_enqueue, mock_get_command, mock_post
+  ):
+    """Giving up on the wait leaves the command in the robot's queue, where it still
+    runs. A caller who retries would make the pipette aspirate twice from one well."""
+    mock_enqueue.return_value = "cmd-1"
+    released = threading.Event()
+    mock_get_command.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2, command_timeout=0.2)
+    try:
+      with self.assertRaises(TimeoutError):
+        await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+
+      mock_post.assert_called_once_with("/runs/run-id/actions", {"data": {"actionType": "stop"}})
+
+      with self.assertRaises(RuntimeError) as refused:
+        await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+    finally:
+      released.set()
+
+    self.assertIn("setup()", str(refused.exception))
+    self.assertEqual(mock_enqueue.call_count, 1)  # the second attempt reached no robot
+
+  @patch("ot_api.requestor.post")
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  @patch("ot_api.run_id", "run-id", create=True)
+  async def test_a_command_that_finished_during_the_last_sleep_is_not_a_timeout(
+    self, mock_enqueue, mock_get_command, mock_post
+  ):
+    """The read after the final sleep is the one that sees a command the robot has
+    just finished. Handing it the sliver of deadline that is left cannot answer, and
+    then a move that worked halts the robot and latches the refusal."""
+    mock_enqueue.return_value = "cmd-1"
+    answers = ["running"]
+
+    def status(*args, **kwargs):
+      state = answers.pop(0) if answers else "succeeded"
+      return {"data": {"status": state, "result": {}}}
+
+    mock_get_command.side_effect = status
+
+    # A poll interval wider than the budget puts the whole remainder into one sleep,
+    # so the read that follows it lands exactly on the deadline.
+    backend = self._backend(request_timeout=5.0, command_timeout=0.2, status_poll_interval=1.0)
+
+    await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+
+    mock_post.assert_not_called()  # nothing halted a run that completed
+    await backend.move_pipette_head(Coordinate(4.0, 5.0, 6.0), pipette_id="left")
+    self.assertEqual(mock_enqueue.call_count, 2)  # and nothing latched the refusal
+
+  @patch("ot_api.requestor.post")
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  @patch("ot_api.run_id", "run-id", create=True)
+  async def test_a_status_read_that_never_answers_does_not_end_the_command(
+    self, mock_enqueue, mock_get_command, mock_post
+  ):
+    """A lost GET says nothing about the motion. Ending the command on one takes the
+    abort decision away from whoever asked for the longer budget: with a ten-minute
+    mix and a seven-second request budget, the driver would fire at eight seconds."""
+    mock_enqueue.return_value = "cmd-1"
+    released = threading.Event()
+    reads = []
+
+    def status(*args, **kwargs):
+      reads.append(1)
+      if len(reads) == 1:
+        released.wait()  # this one read never comes back
+      return {"data": {"status": "succeeded", "result": {}}}
+
+    mock_get_command.side_effect = status
+
+    backend = self._backend(request_timeout=0.2, command_timeout=5.0, status_poll_interval=0.01)
+    try:
+      await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+    finally:
+      released.set()
+
+    self.assertEqual(len(reads), 2)  # it polled again rather than giving up
+    mock_post.assert_not_called()  # and nothing halted a run that was still running
+    await backend.move_pipette_head(Coordinate(4.0, 5.0, 6.0), pipette_id="left")
+
+  @patch("ot_api.requestor.post")
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  @patch("ot_api.run_id", "run-id", create=True)
+  async def test_a_slow_position_read_does_not_halt_the_robot(
+    self, mock_enqueue, mock_get_command, mock_post
+  ):
+    """savePosition moves nothing, so giving up on it leaves no motion outstanding.
+    Halting the run and refusing everything after would cost more than the timeout."""
+    mock_enqueue.return_value = "cmd-1"
+    released = threading.Event()
+    mock_get_command.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2)
+    try:
+      with self.assertRaises(RuntimeError):
+        await backend.get_channel_position(0)
+
+      mock_post.assert_not_called()
+
+      # still usable: the refusal did not latch
+      with self.assertRaises(RuntimeError) as second:
+        await backend.get_channel_position(0)
+    finally:
+      released.set()
+
+    self.assertNotIn("setup()", str(second.exception))
+    self.assertEqual(mock_enqueue.call_count, 2)
+
+  @patch("ot_api.health.home")
+  @patch("ot_api.health.get")
+  @patch("ot_api.lh.add_mounted_pipettes")
+  @patch("ot_api.runs.create")
+  @patch("ot_api.requestor.post")
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  @patch("ot_api.run_id", "run-id", create=True)
+  async def test_a_setup_that_cannot_create_a_run_leaves_the_refusal_in_place(
+    self,
+    mock_enqueue,
+    mock_get_command,
+    mock_post,
+    mock_create,
+    mock_pipettes,
+    mock_health,
+    mock_home,
+  ):
+    """A fresh run is what the stale command cannot execute in, so until one exists
+    there is nothing to lift the refusal. The robot-server answering 409 while it
+    still holds the old run is the likeliest reason an operator is here at all."""
+    mock_enqueue.return_value = "cmd-1"
+    released = threading.Event()
+    mock_get_command.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2, command_timeout=0.2)
+    try:
+      with self.assertRaises(TimeoutError):
+        await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+
+      mock_create.side_effect = RuntimeError("RunConflictError")
+      with self.assertRaises(RuntimeError):
+        await backend.setup()
+
+      with self.assertRaises(RuntimeError) as refused:
+        await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+    finally:
+      released.set()
+
+    self.assertIn("setup()", str(refused.exception))
+    self.assertEqual(mock_enqueue.call_count, 1)  # nothing went into the stale run
+
+  @patch("ot_api.health.home")
+  @patch("ot_api.health.get")
+  @patch("ot_api.lh.add_mounted_pipettes")
+  @patch("ot_api.runs.create")
+  @patch("ot_api.requestor.post")
+  @patch("ot_api.runs.get_command")
+  @patch("ot_api.runs.enqueue_command")
+  @patch("ot_api.run_id", "run-id", create=True)
+  async def test_setup_is_the_way_back_from_a_latched_refusal(
+    self,
+    mock_enqueue,
+    mock_get_command,
+    mock_post,
+    mock_create,
+    mock_pipettes,
+    mock_health,
+    mock_home,
+  ):
+    """The refusal names setup(); this is what makes that a real instruction."""
+    mock_enqueue.return_value = "cmd-1"
+    mock_create.return_value = "run-2"
+    mock_pipettes.return_value = _PIPETTES
+    mock_health.side_effect = _mock_health_get
+    released = threading.Event()
+    mock_get_command.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2, command_timeout=0.2)
+    try:
+      with self.assertRaises(TimeoutError):
+        await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+    finally:
+      released.set()
+
+    mock_get_command.side_effect = lambda *a, **kw: {"data": {"status": "succeeded", "result": {}}}
+    await backend.setup()
+
+    await backend.move_pipette_head(Coordinate(1.0, 2.0, 3.0), pipette_id="left")
+
+  @patch("ot_api.modules.list_connected_modules")
+  async def test_a_timeout_names_the_call_and_a_readable_budget(self, mock_list):
+    """An operator debugging a lossy OT-2 reads this line; the proxy the backend logs
+    through must not be what it names."""
+    released = threading.Event()
+    mock_list.side_effect = lambda *a, **kw: released.wait()
+
+    backend = self._backend(request_timeout=0.2)
+    try:
+      with self.assertRaises(TimeoutError) as caught:
+        await backend.list_connected_modules()
+    finally:
+      released.set()
+
+    self.assertIn("modules.list_connected_modules", str(caught.exception))
+    self.assertIn("0.2s", str(caught.exception))
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py
index 628fd67113e..98da141b2b0 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py
@@ -3,6 +3,8 @@
 Dry-runs the real OpentronsOT2Backend without hardware or the ``ot_api`` library
 by swapping the backend's transport handle (``self._ot``) for a recorder that
 logs every call and returns canned data for the few reads the backend makes back.
+Robot commands arrive as ``runs.enqueue_command(command_type, params)``, which is
+what the backend puts on the wire, and each is answered "succeeded" on the poll.
 
 This mirrors how ``STARChatterboxBackend`` dry-runs ``STARBackend``: only the
 transport is replaced, so all the real high-level logic (pipette selection, tip
@@ -17,6 +19,9 @@
 from pylabrobot.legacy.liquid_handling.backends.backend import LiquidHandlerBackend
 from pylabrobot.legacy.liquid_handling.backends.opentrons_backend import (
   _OT_DECK_IS_ADDRESSABLE_AREA_VERSION,
+  DEFAULT_COMMAND_TIMEOUT,
+  DEFAULT_REQUEST_TIMEOUT,
+  DEFAULT_STATUS_POLL_INTERVAL,
   OpentronsOT2Backend,
 )
 
@@ -55,18 +60,35 @@ class _OTChatterboxModule:
   """Stand-in for the ``ot_api`` module that records calls instead of issuing them.
 
   Provides the sub-namespaces and reads the real backend touches: ``runs.create``,
-  ``lh.add_mounted_pipettes``, ``health.get``, ``labware.define``,
-  ``modules.list_connected_modules`` and ``lh.save_position`` return canned data;
-  everything else is recorded and returns ``None``. ``run_id`` stays ``None`` so
-  ``stop()`` skips the cancel request.
+  ``runs.enqueue_command``/``runs.get_command``, ``lh.add_mounted_pipettes``,
+  ``health.get``, ``labware.define`` and ``modules.list_connected_modules`` return
+  canned data; everything else is recorded and returns ``None``. ``run_id`` stays
+  ``None`` so ``stop()`` skips the release requests.
+
+  A recorded ``moveToCoordinates`` updates the position ``savePosition`` reports,
+  so a dry run that reads a channel back gets where it just sent it rather than
+  the origin every time.
   """
 
   def __init__(self, left_pipette, right_pipette, api_version: str, verbose: bool = True):
     self.calls: List[Tuple[str, tuple, dict]] = []
+    self.commands: List[Tuple[str, dict]] = []  # (command_type, params) in send order
     self.run_id: Optional[str] = None
     self._verbose = verbose
 
-    self.runs = _RecordingNamespace(self, "runs", {"create": lambda: "chatterbox-run"})
+    self.position: Dict[str, float] = {"x": 0.0, "y": 0.0, "z": 0.0}
+
+    self.runs = _RecordingNamespace(
+      self,
+      "runs",
+      {
+        "create": lambda: "chatterbox-run",
+        "enqueue_command": lambda: "chatterbox-command",
+        "get_command": lambda: {
+          "data": {"status": "succeeded", "result": {"position": dict(self.position)}}
+        },
+      },
+    )
     self.health = _RecordingNamespace(self, "health", {"get": lambda: {"api_version": api_version}})
     self.labware = _RecordingNamespace(
       self, "labware", {"define": lambda: {"data": {"definitionUri": "pylabrobot/chatterbox/1"}}}
@@ -76,10 +98,7 @@ def __init__(self, left_pipette, right_pipette, api_version: str, verbose: bool
     self.lh = _RecordingNamespace(
       self,
       "lh",
-      {
-        "add_mounted_pipettes": lambda: (left_pipette, right_pipette),
-        "save_position": lambda: {"data": {"result": {"position": {"x": 0, "y": 0, "z": 0}}}},
-      },
+      {"add_mounted_pipettes": lambda: (left_pipette, right_pipette)},
     )
 
   def log(self, qualified: str, args: tuple, kwargs: dict):
@@ -90,6 +109,17 @@ def log(self, qualified: str, args: tuple, kwargs: dict):
     logger.log(LOG_LEVEL_IO, "%s", rendered)
     if self._verbose:
       print(rendered)
+    if qualified == "runs.enqueue_command" and len(args) >= 2:
+      self._track_command(args[0], args[1])
+
+  def _track_command(self, command_type: str, params: dict) -> None:
+    """Record one robot command, and follow a move so ``savePosition`` reports it."""
+    self.commands.append((command_type, dict(params)))
+    if command_type != "moveToCoordinates":
+      return
+    for axis, moved_to in params.get("coordinates", {}).items():
+      if moved_to is not None:
+        self.position[axis] = moved_to
 
   def __getattr__(self, name: str):
     # top-level functions the backend calls directly: set_host, set_port, set_run
@@ -108,7 +138,8 @@ class OpentronsOT2ChatterboxBackend(OpentronsOT2Backend):
 
   Runs the real OpentronsOT2Backend logic with its transport replaced by a
   recorder - no hardware and no ``ot_api`` library required. Every issued call is
-  printed and collected in :attr:`commands`.
+  printed; :attr:`commands` collects the robot commands and :attr:`calls` every
+  ``ot_api`` call underneath them.
 
   Example:
     >>> from pylabrobot.legacy.liquid_handling import LiquidHandler
@@ -126,6 +157,9 @@ def __init__(
     port: int = 31950,
     api_version: str = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION,
     verbose: bool = True,
+    request_timeout: float = DEFAULT_REQUEST_TIMEOUT,
+    command_timeout: float = DEFAULT_COMMAND_TIMEOUT,
+    status_poll_interval: float = DEFAULT_STATUS_POLL_INTERVAL,
   ):
     """Initialize the chatterbox.
 
@@ -135,6 +169,9 @@ def __init__(
       api_version: reported Opentrons API version; defaults to the version at
         which tip drops route through the addressable-area trash.
       verbose: if True, print every recorded call.
+      request_timeout: how long one request/response with the robot may take, in seconds.
+      command_timeout: how long a command that moves the robot may take, in seconds.
+      status_poll_interval: delay between two reads of a running command's status.
     """
     # Skip OpentronsOT2Backend.__init__ (it requires ot_api); set up state directly.
     LiquidHandlerBackend.__init__(self)
@@ -149,6 +186,7 @@ def __init__(
     self._right_pipette_name = right_pipette_name
     self.host = host
     self.port = port
+    self._init_wire_state(request_timeout, command_timeout, status_poll_interval)
 
     left = (
       {"name": left_pipette_name, "pipetteId": "chatterbox-left"} if left_pipette_name else None
@@ -166,10 +204,15 @@ def __init__(
     self._plr_name_to_load_name: Dict[str, str] = {}
 
   @property
-  def commands(self) -> List[Tuple[str, tuple, dict]]:
-    """Recorded ``(qualified_name, args, kwargs)`` for every call issued so far."""
+  def calls(self) -> List[Tuple[str, tuple, dict]]:
+    """Recorded ``(qualified_name, args, kwargs)`` for every ``ot_api`` call issued."""
     return cast(List[Tuple[str, tuple, dict]], self._ot.calls)
 
+  @property
+  def commands(self) -> List[Tuple[str, dict]]:
+    """Recorded ``(command_type, params)`` for every robot command issued so far."""
+    return cast(List[Tuple[str, dict]], self._ot.commands)
+
   def serialize(self) -> dict:
     return {
       **LiquidHandlerBackend.serialize(self),
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py
index 1866ead93f5..78c5b6a1bdb 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py
@@ -11,13 +11,13 @@
   OpentronsOT2ChatterboxBackend,
   OpentronsOT2Simulator,
 )
-from pylabrobot.resources import set_tip_tracking, set_volume_tracking
+from pylabrobot.resources import Coordinate, set_tip_tracking, set_volume_tracking
 from pylabrobot.resources.celltreat import CellTreat_96_wellplate_350ul_Fb
 from pylabrobot.resources.opentrons import OTDeck, opentrons_96_filtertiprack_20ul
 
 
-def _names(backend: OpentronsOT2ChatterboxBackend):
-  return [call[0] for call in backend.commands]
+def _command_types(backend: OpentronsOT2ChatterboxBackend):
+  return [command_type for command_type, _params in backend.commands]
 
 
 class OpentronsChatterboxTests(unittest.IsolatedAsyncioTestCase):
@@ -49,28 +49,55 @@ async def test_setup_resolves_two_channels_without_ot_api(self):
     assert self.backend.left_pipette is not None and self.backend.right_pipette is not None
     self.assertEqual(self.backend.left_pipette["name"], "p20_single_gen2")
 
-  async def test_full_protocol_records_one_wire_call_per_operation(self):
-    """A pickup -> aspirate -> dispense -> trash-discard records exactly one
-    wire call each, via the real backend logic."""
+  async def test_a_channel_reads_back_where_the_dry_run_last_moved_it(self):
+    """savePosition reports the recorded move, so a dry run can drive a channel."""
+    await self.backend.move_channel_to(0, x=50.0, y=60.0, z=70.0)
+
+    self.assertEqual(await self.backend.get_channel_position(0), Coordinate(50.0, 60.0, 70.0))
+
+    await self.backend.move_channel_x(0, x=15.0)
+
+    self.assertEqual(await self.backend.get_channel_position(0), Coordinate(15.0, 60.0, 70.0))
+
+  async def test_full_protocol_sends_one_robot_command_per_operation(self):
+    """A pickup -> aspirate -> dispense -> trash-discard puts exactly one command
+    of each kind on the wire, via the real backend logic."""
     self.plate.get_well("A1").tracker.set_volume(15)
     await self.lh.pick_up_tips(self.tips["A1"])
     await self.lh.aspirate(self.plate["A1"], vols=[10])
     await self.lh.dispense(self.plate["B1"], vols=[10])
     await self.lh.discard_tips()
 
-    names = _names(self.backend)
-    self.assertEqual(names.count("lh.pick_up_tip"), 1)
-    self.assertEqual(names.count("lh.aspirate_in_place"), 1)
-    self.assertEqual(names.count("lh.dispense_in_place"), 1)
+    types = _command_types(self.backend)
+    self.assertEqual(types.count("pickUpTip"), 1)
+    self.assertEqual(types.count("aspirateInPlace"), 1)
+    self.assertEqual(types.count("dispenseInPlace"), 1)
     # api_version defaults to 7.1.0, so the discard routes through the trash addressable area
-    self.assertEqual(names.count("lh.move_to_addressable_area_for_drop_tip"), 1)
-    self.assertEqual(names.count("lh.drop_tip_in_place"), 1)
+    self.assertEqual(types.count("moveToAddressableAreaForDropTip"), 1)
+    self.assertEqual(types.count("dropTipInPlace"), 1)
+
+  async def test_a_pick_up_names_the_well_and_the_pipette_the_robot_needs(self):
+    """The backend builds the wire params itself now, so they are pinned here."""
+    await self.lh.pick_up_tips(self.tips["A1"])
+
+    params = dict(self.backend.commands)["pickUpTip"]
+    self.assertEqual(params["labwareId"], self.backend.get_ot_name("tips"))
+    tip_spot = self.tips.get_item("A1")
+    self.assertEqual(params["wellName"], self.backend.get_ot_name(tip_spot.name))
+    assert self.backend.left_pipette is not None
+    self.assertEqual(params["pipetteId"], self.backend.left_pipette["pipetteId"])
+    self.assertEqual(params["wellLocation"]["origin"], "bottom")
 
   def test_unknown_pipette_name_raises(self):
     """An unrecognised pipette name is rejected at construction."""
     with self.assertRaises(ValueError):
       OpentronsOT2ChatterboxBackend(left_pipette_name="not_a_pipette")
 
+  def test_the_chatterbox_checks_the_budgets_the_real_backend_checks(self):
+    """It skips the real __init__, so the two share one helper rather than drifting."""
+    with self.assertRaises(ValueError):
+      OpentronsOT2ChatterboxBackend(status_poll_interval=0.0, verbose=False)
+
   def test_serialize_includes_pipettes(self):
     """serialize() captures the mounted-pipette names (None for an empty mount)."""
     backend = OpentronsOT2ChatterboxBackend(
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py
index 262c4d3ef60..32c666ef803 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py
@@ -109,7 +109,7 @@ async def stop(self):
     self.right_pipette_has_tip = False
     logger.info("OpentronsOT2Simulator stopped.")
 
-  def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]:
+  async def _current_channel_position(self, channel: int) -> Tuple[str, Coordinate]:
     pipette_id = self._pipette_id_for_channel(channel)
     return pipette_id, self._positions.get(pipette_id, Coordinate.zero())