diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py
index f24c867553a..db64c11a8e2 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend.py
@@ -1,8 +1,9 @@
 import inspect
 import logging
+import math
 import re
 import uuid
-from typing import Any, Dict, List, Optional, Tuple, Union, cast
+from typing import Any, Dict, List, Literal, Optional, Sequence, Tuple, Union, cast
 
 from pylabrobot import utils
 from pylabrobot.io import LOG_LEVEL_IO
@@ -19,6 +20,7 @@
   MultiHeadDispensePlate,
   Pickup,
   PickupTipRack,
+  PipettingOp,
   ResourceDrop,
   ResourceMove,
   ResourcePickup,
@@ -27,8 +29,12 @@
 )
 from pylabrobot.resources import (
   Coordinate,
+  Plate,
+  Resource,
   Tip,
+  Trash,
 )
+from pylabrobot.resources.itemized_resource import ItemizedResource
 from pylabrobot.resources.opentrons import OTDeck
 from pylabrobot.resources.tip_rack import TipRack
 
@@ -100,9 +106,30 @@ def _logged(*args, **kwargs):
     return attr
 
 
+def _fixed_head_tip_geometry(tip: Tip) -> Tuple[object, ...]:
+  """Compare physical tip properties independently of resource identity and placement."""
+  return (
+    tip.get_size_x(),
+    tip.get_size_y(),
+    tip.get_size_z(),
+    tip.fitting_depth,
+    tip.has_filter,
+    tip.maximal_volume,
+    tip.nominal_volume,
+    tip.collar_height if tip.has_collar_height else None,
+    tip.pick_up_location,
+  )
+
+
 class OpentronsOT2Backend(LiquidHandlerBackend):
   """Backends for the Opentrons OT2 liquid handling robots."""
 
+  _FIXED_HEAD_CHANNELS = list(range(8))
+  _FIXED_HEAD_VOLUME_RANGES = {
+    "p20_multi_gen2": (1, 20),
+    "p300_multi_gen2": (20, 300),
+  }
+
   pipette_name2volume = {
     "p10_single": 10,
     "p10_multi": 10,
@@ -120,7 +147,23 @@ class OpentronsOT2Backend(LiquidHandlerBackend):
     "p1000_single_gen3": 1000,
   }
 
-  def __init__(self, host: str, port: int = 31950):
+  def __init__(
+    self,
+    host: str,
+    port: int = 31950,
+    fixed_head_mount: Optional[Literal["left", "right"]] = None,
+  ):
+    """Create an OT-2 backend.
+
+    Args:
+      host: Hostname or IP address of the OT-2.
+      port: Robot-server port.
+      fixed_head_mount: Mount containing a supported OT-2 GEN2 eight-channel pipette. When set,
+        the backend exposes the eight fixed nozzles as one full-head operation. All eight channels
+        must be selected together. P20 and P300 GEN2 models are supported, with per-nozzle volumes
+        of 1–20 and 20–300 µL, respectively. The other mount is not exposed in this mode. Resources
+        must be unrotated A-to-H columns of a tip rack or plate at 9 mm pitch.
+    """
     super().__init__()
 
     if not USE_OT:
@@ -131,6 +174,9 @@ def __init__(self, host: str, port: int = 31950):
 
     self.host = host
     self.port = port
+    if fixed_head_mount not in {None, "left", "right"}:
+      raise ValueError("fixed_head_mount must be left, right, or None")
+    self.fixed_head_mount = fixed_head_mount
 
     # All hardware I/O goes through this handle so a subclass (e.g. the chatterbox)
     # can dry-run the backend by swapping it for a recording stand-in. The real handle
@@ -149,11 +195,27 @@ def __init__(self, host: str, port: int = 31950):
     self._plr_name_to_load_name: Dict[str, str] = {}
 
   def serialize(self) -> dict:
-    return {
+    data = {
       **super().serialize(),
       "host": self.host,
       "port": self.port,
     }
+    if self.fixed_head_mount is not None:
+      data["fixed_head_mount"] = self.fixed_head_mount
+    return data
+
+  def _fixed_head_pipette(self) -> Dict[str, str]:
+    """Return the selected fixed-head pipette or reject the mounted configuration."""
+
+    if self.fixed_head_mount is None:
+      raise RuntimeError("fixed-head operation is not enabled")
+    pipette = self.left_pipette if self.fixed_head_mount == "left" else self.right_pipette
+    if pipette is None or pipette["name"] not in self._FIXED_HEAD_VOLUME_RANGES:
+      raise NoChannelError(
+        f"The {self.fixed_head_mount} mount must contain a supported OT-2 GEN2 "
+        "eight-channel pipette."
+      )
+    return pipette
 
   async def setup(self, skip_home: bool = False):
     # create run
@@ -164,6 +226,8 @@ async def setup(self, skip_home: bool = False):
     self.left_pipette, self.right_pipette = self._ot.lh.add_mounted_pipettes()
 
     self.left_pipette_has_tip = self.right_pipette_has_tip = False
+    if self.fixed_head_mount is not None:
+      self._fixed_head_pipette()
 
     # get api version
     health = self._ot.health.get()
@@ -174,8 +238,118 @@ async def setup(self, skip_home: bool = False):
 
   @property
   def num_channels(self) -> int:
+    if self.fixed_head_mount is not None:
+      return len(self._FIXED_HEAD_CHANNELS)
     return len([p for p in [self.left_pipette, self.right_pipette] if p is not None])
 
+  def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool:
+    def supports_tip(channel_vol: float, tip_vol: float) -> bool:
+      if channel_vol == 20:
+        return tip_vol in {10, 20}
+      if channel_vol == 300:
+        return tip_vol in {200, 300}
+      if channel_vol == 1000:
+        return tip_vol in {1000}
+      raise ValueError(f"Unknown channel volume: {channel_vol}")
+
+    if self.fixed_head_mount is not None:
+      if channel_idx not in self._FIXED_HEAD_CHANNELS:
+        return False
+      pipette = self._fixed_head_pipette()
+      return supports_tip(self.pipette_name2volume[pipette["name"]], tip.maximal_volume)
+    if channel_idx == 0:
+      if self.left_pipette is None:
+        return False
+      left_volume = OpentronsOT2Backend.pipette_name2volume[self.left_pipette["name"]]
+      return supports_tip(left_volume, tip.maximal_volume)
+    if channel_idx == 1:
+      if self.right_pipette is None:
+        return False
+      right_volume = OpentronsOT2Backend.pipette_name2volume[self.right_pipette["name"]]
+      return supports_tip(right_volume, tip.maximal_volume)
+    return False
+
+  def _validate_fixed_head_ops(self, ops: Sequence[PipettingOp], use_channels: List[int]) -> None:
+    """Validate one ordered, 9 mm-pitch column for the fixed eight-nozzle head."""
+
+    if use_channels != self._FIXED_HEAD_CHANNELS or len(ops) != len(self._FIXED_HEAD_CHANNELS):
+      raise ValueError("fixed-head operations require channels 0 through 7 in order")
+
+    geometry = _fixed_head_tip_geometry(ops[0].tip)
+    if any(_fixed_head_tip_geometry(op.tip) != geometry for op in ops[1:]):
+      raise ValueError("fixed-head operations require one shared tip geometry")
+    if not all(self.can_pick_up_tip(channel, op.tip) for channel, op in zip(use_channels, ops)):
+      raise NoChannelError("The fixed-head pipette cannot use the selected tips.")
+
+    resources = [op.resource for op in ops]
+    if len({id(resource) for resource in resources}) != len(resources):
+      if all(
+        isinstance(op, Drop)
+        and op.resource is resources[0]
+        and isinstance(op.resource, Trash)
+        and op.resource.name == "trash"
+        and op.resource.is_in_subtree_of(self.deck)
+        for op in ops
+      ):
+        anchor = ops[0].offset
+        if not all(op.offset == anchor + Coordinate(y=-9 * index) for index, op in enumerate(ops)):
+          raise ValueError("fixed-head trash offsets must form one 9 mm-pitch column")
+        return
+      raise ValueError("fixed-head operations require eight distinct positions")
+
+    if any(op.offset != ops[0].offset for op in ops[1:]):
+      raise ValueError("fixed-head operations require one shared offset")
+
+    parent = resources[0].parent
+    if parent is None or any(resource.parent is not parent for resource in resources):
+      raise ValueError("fixed-head positions must share one itemized resource")
+    if not isinstance(parent, (TipRack, Plate)):
+      raise ValueError("fixed-head positions must belong to a tip rack or plate")
+
+    itemized_parent = cast(ItemizedResource[Resource], parent)
+    rows = [itemized_parent.get_child_row(resource) for resource in resources]
+    columns = [itemized_parent.get_child_column(resource) for resource in resources]
+    if rows != self._FIXED_HEAD_CHANNELS or len(set(columns)) != 1:
+      raise ValueError("fixed-head positions must be one ordered A-to-H column")
+    z_anchor = (
+      "cavity_bottom"
+      if isinstance(ops[0], (SingleChannelAspiration, SingleChannelDispense))
+      else "b"
+    )
+    positions = [resource.get_location_wrt(self.deck, "c", "c", z_anchor) for resource in resources]
+    anchor = positions[0]
+    for index, (resource, position) in enumerate(zip(resources, positions)):
+      rotation = resource.get_absolute_rotation()
+      if any(angle % 360 != 0 for angle in (rotation.x, rotation.y, rotation.z)):
+        raise ValueError("fixed-head positions must be unrotated")
+      if not all(
+        math.isclose(actual, expected, rel_tol=0, abs_tol=1e-6)
+        for actual, expected in (
+          (position.x, anchor.x),
+          (position.y, anchor.y - 9 * index),
+          (position.z, anchor.z),
+        )
+      ):
+        raise ValueError("fixed-head positions must use aligned 9 mm row pitch at one height")
+
+  @staticmethod
+  def _validate_equal_fixed_head_values(
+    ops: Sequence[Union[SingleChannelAspiration, SingleChannelDispense]],
+  ) -> None:
+    """Reject per-nozzle settings that one fixed-head command cannot represent."""
+
+    first = ops[0]
+    if any(op.volume != first.volume for op in ops[1:]):
+      raise ValueError("fixed-head operations require one shared volume")
+    if any(op.flow_rate != first.flow_rate for op in ops[1:]):
+      raise ValueError("fixed-head operations require one shared flow_rate")
+    if any(op.liquid_height != first.liquid_height for op in ops[1:]):
+      raise ValueError("fixed-head operations require one shared liquid_height")
+    if any(op.blow_out_air_volume != first.blow_out_air_volume for op in ops[1:]):
+      raise ValueError("fixed-head operations require one shared blow_out_air_volume")
+    if any(op.mix != first.mix for op in ops[1:]):
+      raise ValueError("fixed-head operations require one shared mix")
+
   async def stop(self):
     """Cancel any active OT run, then clear labware definitions."""
     self._plr_name_to_load_name = {}
@@ -315,8 +489,18 @@ async def _assign_tip_rack(self, tip_rack: TipRack, tip: Tip):
 
     self._tip_racks[tip_rack.name] = slot
 
-  def _get_pickup_pipette(self, ops: List[Pickup]) -> str:
+  def _get_pickup_pipette(self, ops: List[Pickup], use_channels: Optional[List[int]] = None) -> str:
     """Get the pipette for a tip pick-up, or raise."""
+    use_channels = list(range(len(ops))) if use_channels is None else use_channels
+    if self.fixed_head_mount is not None:
+      self._validate_fixed_head_ops(ops, use_channels)
+      pipette = self._fixed_head_pipette()
+      has_tip = (
+        self.left_pipette_has_tip if self.fixed_head_mount == "left" else self.right_pipette_has_tip
+      )
+      if has_tip:
+        raise NoChannelError("The fixed-head pipette already has tips.")
+      return cast(str, pipette["pipetteId"])
     assert len(ops) == 1, "only one channel supported for now"
     op = ops[0]
     assert op.resource.parent is not None, "must not be a floating resource"
@@ -325,8 +509,18 @@ def _get_pickup_pipette(self, ops: List[Pickup]) -> str:
       raise NoChannelError("No pipette channel of right type with no tip available.")
     return pipette_id
 
-  def _get_drop_pipette(self, ops: List[Drop]) -> str:
+  def _get_drop_pipette(self, ops: List[Drop], use_channels: Optional[List[int]] = None) -> str:
     """Get the pipette for a tip drop, or raise."""
+    use_channels = list(range(len(ops))) if use_channels is None else use_channels
+    if self.fixed_head_mount is not None:
+      self._validate_fixed_head_ops(ops, use_channels)
+      pipette = self._fixed_head_pipette()
+      has_tip = (
+        self.left_pipette_has_tip if self.fixed_head_mount == "left" else self.right_pipette_has_tip
+      )
+      if not has_tip:
+        raise NoChannelError("The fixed-head pipette has no tips.")
+      return cast(str, pipette["pipetteId"])
     assert len(ops) == 1, "only one channel supported for now"
     op = ops[0]
     assert op.resource.parent is not None, "must not be a floating resource"
@@ -336,9 +530,37 @@ def _get_drop_pipette(self, ops: List[Drop]) -> str:
     return pipette_id
 
   def _get_liquid_pipette(
-    self, ops: Union[List[SingleChannelAspiration], List[SingleChannelDispense]]
+    self,
+    ops: Union[List[SingleChannelAspiration], List[SingleChannelDispense]],
+    use_channels: Optional[List[int]] = None,
   ) -> str:
     """Get the pipette for an aspirate/dispense, or raise."""
+    use_channels = list(range(len(ops))) if use_channels is None else use_channels
+    if self.fixed_head_mount is not None:
+      self._validate_fixed_head_ops(ops, use_channels)
+      self._validate_equal_fixed_head_values(ops)
+      pipette = self._fixed_head_pipette()
+      minimum, maximum = self._FIXED_HEAD_VOLUME_RANGES[pipette["name"]]
+      if not minimum <= ops[0].volume <= maximum:
+        raise NoChannelError(
+          f"{pipette['name']} supports {minimum} through {maximum} µL per nozzle."
+        )
+      if ops[0].volume > ops[0].tip.maximal_volume:
+        raise ValueError("fixed-head volume exceeds the selected tip capacity")
+      if ops[0].mix is not None:
+        mix = ops[0].mix
+        if not minimum <= mix.volume <= min(maximum, ops[0].tip.maximal_volume):
+          raise ValueError(f"fixed-head mix volume must be within {minimum} through {maximum} µL")
+        if mix.surface_following_distance not in (None, 0) or mix.auto_surface_following:
+          raise ValueError("fixed-head mixing does not support surface following")
+      if ops[0].blow_out_air_volume not in (None, 0):
+        raise ValueError("fixed-head operations do not support blow_out_air_volume")
+      has_tip = (
+        self.left_pipette_has_tip if self.fixed_head_mount == "left" else self.right_pipette_has_tip
+      )
+      if not has_tip:
+        raise NoChannelError("The fixed-head pipette has no tips.")
+      return cast(str, pipette["pipetteId"])
     assert len(ops) == 1, "only one channel supported for now"
     pipette_id = self.select_liquid_pipette(ops[0].volume)
     if pipette_id is None:
@@ -366,7 +588,7 @@ def _set_tip_state(self, pipette_id: str, has_tip: bool):
   async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]):
     """Pick up tips from the specified resource."""
 
-    pipette_id = self._get_pickup_pipette(ops)
+    pipette_id = self._get_pickup_pipette(ops, use_channels)
     op = ops[0]
 
     offset_x, offset_y, offset_z = (
@@ -397,7 +619,7 @@ async def pick_up_tips(self, ops: List[Pickup], use_channels: List[int]):
   async def drop_tips(self, ops: List[Drop], use_channels: List[int]):
     """Drop tips from the specified resource."""
 
-    pipette_id = self._get_drop_pipette(ops)
+    pipette_id = self._get_drop_pipette(ops, use_channels)
     op = ops[0]
 
     use_fixed_trash = op.resource.name == "trash" and _fixed_trash_is_addressable(
@@ -510,7 +732,7 @@ def _deck_to_robot_frame(self, location: Coordinate) -> Coordinate:
   async def aspirate(self, ops: List[SingleChannelAspiration], use_channels: List[int]):
     """Aspirate liquid from the specified resource using pip."""
 
-    pipette_id = self._get_liquid_pipette(ops)
+    pipette_id = self._get_liquid_pipette(ops, use_channels)
     op = ops[0]
     volume = op.volume
 
@@ -585,7 +807,7 @@ def _get_default_dispense_flow_rate(self, pipette_name: str) -> float:
   async def dispense(self, ops: List[SingleChannelDispense], use_channels: List[int]):
     """Dispense liquid from the specified resource using pip."""
 
-    pipette_id = self._get_liquid_pipette(ops)
+    pipette_id = self._get_liquid_pipette(ops, use_channels)
     op = ops[0]
     volume = op.volume
 
@@ -663,6 +885,10 @@ async def list_connected_modules(self) -> List[dict]:
     return cast(List[dict], self._ot.modules.list_connected_modules())
 
   def _pipette_id_for_channel(self, channel: int) -> str:
+    if self.fixed_head_mount is not None:
+      if channel != 0:
+        raise NoChannelError("Position the fixed head through anchor channel 0.")
+      return cast(str, self._fixed_head_pipette()["pipetteId"])
     pipettes = []
     if self.left_pipette is not None:
       pipettes.append(self.left_pipette["pipetteId"])
@@ -755,25 +981,3 @@ async def move_pipette_head(
       speed=speed,
       force_direct=force_direct,
     )
-
-  def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool:
-    def supports_tip(channel_vol: float, tip_vol: float) -> bool:
-      if channel_vol == 20:
-        return tip_vol in {10, 20}
-      if channel_vol == 300:
-        return tip_vol in {200, 300}
-      if channel_vol == 1000:
-        return tip_vol in {1000}
-      raise ValueError(f"Unknown channel volume: {channel_vol}")
-
-    if channel_idx == 0:
-      if self.left_pipette is None:
-        return False
-      left_volume = OpentronsOT2Backend.pipette_name2volume[self.left_pipette["name"]]
-      return supports_tip(left_volume, tip.maximal_volume)
-    if channel_idx == 1:
-      if self.right_pipette is None:
-        return False
-      right_volume = OpentronsOT2Backend.pipette_name2volume[self.right_pipette["name"]]
-      return supports_tip(right_volume, tip.maximal_volume)
-    return False
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py
index 9de5a5a914e..e69b7c54a69 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_backend_tests.py
@@ -1,4 +1,5 @@
 import unittest
+from dataclasses import replace
 from unittest.mock import patch
 
 import pytest
@@ -13,11 +14,21 @@
 from pylabrobot.legacy.liquid_handling.errors import NoChannelError
 from pylabrobot.legacy.liquid_handling.standard import (
   Drop,
+  Mix,
   Pickup,
   SingleChannelAspiration,
 )
-from pylabrobot.resources import Coordinate, Tip, no_volume_tracking
+from pylabrobot.resources import (
+  Coordinate,
+  Tip,
+  does_tip_tracking,
+  does_volume_tracking,
+  no_volume_tracking,
+  set_tip_tracking,
+  set_volume_tracking,
+)
 from pylabrobot.resources.celltreat import celltreat_96_wellplate_350uL_Fb
+from pylabrobot.resources.errors import HasTipError, TooLittleLiquidError, TooLittleVolumeError
 from pylabrobot.resources.opentrons import (
   OTDeck,
   opentrons_96_filtertiprack_20ul,
@@ -267,9 +278,369 @@ async def test_tip_drop_to_trash_uses_addressable_area(
     mock_drop_tip.assert_not_called()
 
 
+class OpentronsFixedHeadTests(unittest.IsolatedAsyncioTestCase):
+  """Tests for supported OT-2 GEN2 eight-channel full-head dispatch."""
+
+  @patch("ot_api.runs.create", return_value="run-id")
+  @patch("ot_api.lh.add_mounted_pipettes")
+  @patch("ot_api.health.get", side_effect=_mock_health_get)
+  async def asyncSetUp(self, _mock_health, mock_pipettes, _mock_create):
+    """Build a full head and a deck with eight confirmed source volumes."""
+    self._previous_volume_tracking = does_volume_tracking()
+    self._previous_tip_tracking = does_tip_tracking()
+    set_volume_tracking(True)
+    set_tip_tracking(True)
+    mock_pipettes.return_value = (
+      {"pipetteId": "fixed-head-id", "name": "p300_multi_gen2"},
+      {"pipetteId": "right-id", "name": "p20_single_gen2"},
+    )
+    self.backend = OpentronsOT2Backend(host="localhost", port=1338, fixed_head_mount="left")
+    self.deck = OTDeck()
+    self.lh = LiquidHandler(backend=self.backend, deck=self.deck)
+    await self.lh.setup(skip_home=True)
+    self.tip_rack = opentrons_96_tiprack_300ul(name="tip_rack")
+    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)
+    for well in self.plate["A1:H1"]:
+      well.tracker.set_volume(100)
+
+  async def asyncTearDown(self):
+    """Restore the caller's tip- and volume-tracking settings."""
+    set_volume_tracking(self._previous_volume_tracking)
+    set_tip_tracking(self._previous_tip_tracking)
+
+  @patch("ot_api.lh.drop_tip")
+  @patch("ot_api.lh.dispense_in_place")
+  @patch("ot_api.lh.aspirate_in_place")
+  @patch("ot_api.lh.move_arm")
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_full_head_uses_one_dispatch_per_primitive(
+    self,
+    _mock_define,
+    _mock_add,
+    mock_pick_up,
+    _mock_move,
+    mock_aspirate,
+    mock_dispense,
+    mock_drop,
+  ):
+    """Transfer eight per-nozzle volumes and return eight tips through one command each."""
+    self.assertEqual(len({spot.get_tip().name for spot in self.tip_rack["A1:H1"]}), 8)
+    await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+    await self.lh.aspirate(self.plate["A1:H1"], vols=[20] * 8)
+    self.assertEqual([well.tracker.get_used_volume() for well in self.plate["A1:H1"]], [80] * 8)
+    self.assertEqual(
+      [tracker.get_tip().tracker.get_used_volume() for tracker in self.lh.head.values()], [20] * 8
+    )
+    await self.lh.dispense(self.plate["A2:H2"], vols=[20] * 8)
+    self.assertEqual([well.tracker.get_used_volume() for well in self.plate["A2:H2"]], [20] * 8)
+    await self.lh.return_tips()
+
+    mock_pick_up.assert_called_once()
+    self.assertEqual(
+      mock_pick_up.call_args.kwargs["well_name"], self.backend.get_ot_name("tip_rack_A1")
+    )
+    mock_aspirate.assert_called_once_with(
+      volume=20.0,
+      flow_rate=94,
+      pipette_id="fixed-head-id",
+    )
+    mock_dispense.assert_called_once_with(
+      volume=20.0,
+      flow_rate=94,
+      pipette_id="fixed-head-id",
+    )
+    mock_drop.assert_called_once()
+    self.assertEqual(
+      mock_drop.call_args.kwargs["well_name"], self.backend.get_ot_name("tip_rack_A1")
+    )
+    self.assertTrue(all(not tracker.has_tip for tracker in self.lh.head.values()))
+    self.assertTrue(all(spot.has_tip() for spot in self.tip_rack["A1:H1"]))
+
+  @patch("ot_api.lh.pick_up_tip")
+  async def test_fixed_head_rejects_different_tip_geometry_before_dispatch(self, mock_pick_up):
+    """Distinct tip identities are permitted, but every physical property must agree."""
+    spots = self.tip_rack["A1:H1"]
+    ops = [Pickup(spot, Coordinate.zero(), spot.get_tip()) for spot in spots]
+    original = ops[-1].tip
+    parameters = dict(
+      name="different_tip",
+      diameter=original.get_size_x(),
+      size_z=original.get_size_z(),
+      has_filter=original.has_filter,
+      maximal_volume=original.maximal_volume,
+      fitting_depth=original.fitting_depth,
+      nominal_volume=original.nominal_volume,
+    )
+    for changes in (
+      {"diameter": original.get_size_x() + 1},
+      {"size_z": original.get_size_z() + 1},
+      {"fitting_depth": original.fitting_depth + 1},
+      {"has_filter": not original.has_filter},
+      {"maximal_volume": original.maximal_volume - 1},
+      {"nominal_volume": original.nominal_volume - 1},
+      {"collar_height": 1},
+      {"pick_up_location": Coordinate(z=1)},
+    ):
+      with self.subTest(changes=changes), self.assertRaisesRegex(ValueError, "shared tip geometry"):
+        await self.backend.pick_up_tips(
+          ops[:-1] + [replace(ops[-1], tip=Tip(**{**parameters, **changes}))], list(range(8))
+        )
+    mock_pick_up.assert_not_called()
+
+  @patch("ot_api.lh.pick_up_tip", side_effect=RuntimeError("dispatch failed"))
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_full_head_pickup_failure_rolls_back_all_tip_state(
+    self, _mock_define, _mock_add, mock_pick_up
+  ):
+    """A transport failure commits none of the eight queued tip changes."""
+    with self.assertRaisesRegex(RuntimeError, "dispatch failed"):
+      await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+
+    mock_pick_up.assert_called_once()
+    self.assertTrue(all(not tracker.has_tip for tracker in self.lh.head.values()))
+    self.assertTrue(all(spot.has_tip() for spot in self.tip_rack["A1:H1"]))
+
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_fixed_head_rejects_partial_and_reversed_columns_before_dispatch(
+    self, _mock_define, _mock_add, mock_pick_up
+  ):
+    """Reject partial or reversed nozzle targets without a tip command."""
+    with self.assertRaisesRegex(ValueError, "channels 0 through 7"):
+      await self.lh.pick_up_tips(self.tip_rack["A1:G1"])
+    with self.assertRaisesRegex(ValueError, "ordered A-to-H column"):
+      await self.lh.pick_up_tips(list(reversed(self.tip_rack["A1:H1"])))
+
+    mock_pick_up.assert_not_called()
+
+  @patch("ot_api.lh.aspirate_in_place")
+  @patch("ot_api.lh.move_arm")
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_fixed_head_rejects_mixed_per_nozzle_volumes_before_liquid_dispatch(
+    self, _mock_define, _mock_add, _mock_pick_up, _mock_move, mock_aspirate
+  ):
+    """Unequal volumes reject the complete action and preserve mounted tips."""
+    await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+    with self.assertRaisesRegex(ValueError, "one shared volume"):
+      await self.lh.aspirate(self.plate["A1:H1"], vols=[20] * 7 + [21])
+
+    mock_aspirate.assert_not_called()
+    self.assertTrue(all(tracker.has_tip for tracker in self.lh.head.values()))
+
+  @patch("ot_api.runs.create", return_value="run-id")
+  @patch("ot_api.lh.add_mounted_pipettes")
+  async def test_fixed_head_setup_rejects_a_legacy_mounted_model(self, mock_pipettes, _mock_create):
+    """Model discovery must match an explicitly supported GEN2 identity."""
+    mock_pipettes.return_value = (
+      {"pipetteId": "left-id", "name": "p50_multi"},
+      None,
+    )
+    backend = OpentronsOT2Backend("localhost", fixed_head_mount="left")
+    liquid_handler = LiquidHandler(backend=backend, deck=OTDeck())
+
+    with self.assertRaisesRegex(NoChannelError, "supported OT-2 GEN2 eight-channel"):
+      await liquid_handler.setup(skip_home=True)
+
+  @patch("ot_api.lh.aspirate_in_place")
+  @patch("ot_api.lh.move_arm")
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_p20_full_head_uses_its_model_range_and_default_flow_rate(
+    self, _mock_define, _mock_add, mock_pick_up, _mock_move, mock_aspirate
+  ):
+    """P20 uses its per-nozzle minimum and default flow through the normal frontend."""
+    self.backend.left_pipette = {"pipetteId": "fixed-head-id", "name": "p20_multi_gen2"}
+    tip_rack = opentrons_96_filtertiprack_20ul(name="tip_rack_20")
+    self.deck.assign_child_at_slot(tip_rack, slot=2)
+
+    await self.lh.pick_up_tips(tip_rack["A1:H1"])
+    await self.lh.aspirate(self.plate["A1:H1"], vols=[1] * 8)
+    self.assertEqual([well.tracker.get_used_volume() for well in self.plate["A1:H1"]], [99] * 8)
+
+    mock_pick_up.assert_called_once()
+    mock_aspirate.assert_called_once_with(
+      volume=1.0,
+      flow_rate=7.6,
+      pipette_id="fixed-head-id",
+    )
+
+  @patch("ot_api.lh.aspirate_in_place")
+  @patch("ot_api.lh.move_arm")
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_fixed_head_rejects_volume_outside_the_detected_model_range(
+    self, _mock_define, _mock_add, _mock_pick_up, _mock_move, mock_aspirate
+  ):
+    """P300 rejects a volume below its operating minimum before liquid dispatch."""
+    await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+
+    with self.assertRaisesRegex(NoChannelError, "20 through 300"):
+      await self.lh.aspirate(self.plate["A1:H1"], vols=[19] * 8)
+
+    mock_aspirate.assert_not_called()
+
+  @patch("ot_api.lh.aspirate_in_place", side_effect=RuntimeError("dispatch failed"))
+  @patch("ot_api.lh.move_arm")
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_full_head_liquid_failure_rolls_back_every_source_and_tip(
+    self, _mock_define, _mock_add, _mock_pick_up, _mock_move, mock_aspirate
+  ):
+    """One failed liquid command leaves all eight software volumes uncommitted."""
+    await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+    with self.assertRaisesRegex(RuntimeError, "dispatch failed"):
+      await self.lh.aspirate(self.plate["A1:H1"], vols=[20] * 8)
+    mock_aspirate.assert_called_once()
+    self.assertEqual([well.tracker.get_used_volume() for well in self.plate["A1:H1"]], [100] * 8)
+    self.assertEqual(
+      [tracker.get_tip().tracker.get_used_volume() for tracker in self.lh.head.values()], [0] * 8
+    )
+
+  @patch("ot_api.lh.dispense_in_place")
+  @patch("ot_api.lh.aspirate_in_place")
+  @patch("ot_api.lh.move_arm")
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_late_invalid_volume_rolls_back_queued_state_without_dispatch(
+    self, _mock_define, _mock_add, _mock_pick_up, mock_move, mock_aspirate, mock_dispense
+  ):
+    """Reject an invalid eighth well without leaving changes queued on the first seven."""
+    await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+    self.plate.get_well("H1").tracker.set_volume(0)
+    with self.assertRaises(TooLittleLiquidError):
+      await self.lh.aspirate(self.plate["A1:H1"], vols=[20] * 8)
+    self.assertEqual(
+      [well.tracker.get_used_volume() for well in self.plate["A1:H1"]], [100] * 7 + [0]
+    )
+    self.assertEqual(
+      [tracker.get_tip().tracker.get_used_volume() for tracker in self.lh.head.values()], [0] * 8
+    )
+    mock_move.assert_not_called()
+    mock_aspirate.assert_not_called()
+
+    self.plate.get_well("H1").tracker.set_volume(100)
+    await self.lh.aspirate(self.plate["A1:H1"], vols=[20] * 8)
+    self.plate.get_well("H2").tracker.set_volume(350)
+    mock_move.reset_mock()
+    with self.assertRaises(TooLittleVolumeError):
+      await self.lh.dispense(self.plate["A2:H2"], vols=[20] * 8)
+    self.assertEqual(
+      [well.tracker.get_used_volume() for well in self.plate["A2:H2"]], [0] * 7 + [350]
+    )
+    self.assertEqual(
+      [tracker.get_tip().tracker.get_used_volume() for tracker in self.lh.head.values()], [20] * 8
+    )
+    mock_move.assert_not_called()
+    mock_dispense.assert_not_called()
+
+  @patch("ot_api.lh.drop_tip")
+  @patch("ot_api.lh.pick_up_tip")
+  @patch("ot_api.labware.add", side_effect=_mock_add)
+  @patch("ot_api.labware.define", side_effect=_mock_define)
+  async def test_late_occupied_return_spot_rolls_back_every_tip_without_dispatch(
+    self, _mock_define, _mock_add, _mock_pick_up, mock_drop
+  ):
+    """An occupied eighth return spot must preserve all mounted tips."""
+    await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+    self.tip_rack.get_item("H1").tracker.add_tip(self.tip_rack.get_item("H2").get_tip())
+    with self.assertRaises(HasTipError):
+      await self.lh.return_tips()
+    self.assertTrue(all(tracker.has_tip for tracker in self.lh.head.values()))
+    self.assertEqual([spot.has_tip() for spot in self.tip_rack["A1:H1"]], [False] * 7 + [True])
+    mock_drop.assert_not_called()
+
+  @patch("ot_api.lh.pick_up_tip")
+  async def test_late_mounted_tip_rejects_pickup_without_leaving_queued_changes(self, mock_pick_up):
+    """An occupied eighth channel preserves the first seven empty channels and rack tips."""
+    tip = self.tip_rack.get_item("H2").get_tip()
+    self.lh.head[7].add_tip(tip)
+    with self.assertRaises(HasTipError):
+      await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+    self.assertEqual([tracker.has_tip for tracker in self.lh.head.values()], [False] * 7 + [True])
+    self.assertTrue(all(spot.has_tip() for spot in self.tip_rack["A1:H1"]))
+    mock_pick_up.assert_not_called()
+
+  @patch("ot_api.lh.pick_up_tip")
+  async def test_fixed_head_rejects_displaced_or_rotated_columns_before_dispatch(
+    self, mock_pick_up
+  ):
+    """Index ordering alone cannot authorize a physically misaligned head."""
+    for displacement in (Coordinate(x=1), Coordinate(y=1), Coordinate(z=1)):
+      spot = self.tip_rack.get_item("H1")
+      original = spot.location
+      assert original is not None
+      spot.location = original + displacement
+      with self.assertRaisesRegex(ValueError, "aligned 9 mm"):
+        await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+      spot.location = original
+    self.tip_rack.rotate(z=180)
+    with self.assertRaisesRegex(ValueError, "unrotated"):
+      await self.lh.pick_up_tips(self.tip_rack["A1:H1"])
+    mock_pick_up.assert_not_called()
+
+  @patch("ot_api.lh.aspirate_in_place")
+  @patch("ot_api.lh.move_arm")
+  async def test_fixed_head_rejects_unrepresentable_parameters_before_movement(
+    self, mock_move, mock_aspirate
+  ):
+    """A single physical command cannot encode different settings for each nozzle."""
+    ops = [
+      SingleChannelAspiration(
+        well, Coordinate.zero(), self.tip_rack.get_item("A1").get_tip(), 20, None, None, None, None
+      )
+      for well in self.plate["A1:H1"]
+    ]
+    mismatches = (
+      {"offset": Coordinate(z=1)},
+      {"flow_rate": 1},
+      {"liquid_height": 1},
+      {"blow_out_air_volume": 1},
+      {"mix": Mix(20, 2, 10)},
+    )
+    for changes in mismatches:
+      with self.subTest(changes=changes), self.assertRaisesRegex(ValueError, "one shared"):
+        await self.backend.aspirate(ops[:-1] + [replace(ops[-1], **changes)], list(range(8)))
+    mock_move.assert_not_called()
+    mock_aspirate.assert_not_called()
+
+  @patch("ot_api.lh.aspirate_in_place")
+  @patch("ot_api.lh.move_arm")
+  async def test_fixed_head_rejects_surface_following_before_movement(
+    self, mock_move, mock_aspirate
+  ):
+    """Fixed-head mixing cannot follow the liquid surface."""
+    for mix in (
+      Mix(20, 2, 10, surface_following_distance=1),
+      Mix(20, 2, 10, auto_surface_following=True),
+    ):
+      ops = [
+        SingleChannelAspiration(
+          well, Coordinate.zero(), self.tip_rack.get_item("A1").get_tip(), 20, None, None, None, mix
+        )
+        for well in self.plate["A1:H1"]
+      ]
+      with self.subTest(mix=mix), self.assertRaisesRegex(ValueError, "surface following"):
+        await self.backend.aspirate(ops, list(range(8)))
+    mock_move.assert_not_called()
+    mock_aspirate.assert_not_called()
+
+
 def _make_backend_with_pipettes(left_name="p300_single_gen2", right_name="p20_single_gen2"):
   """Create a backend with pipette state set directly (no ot_api needed)."""
   backend = OpentronsOT2Backend.__new__(OpentronsOT2Backend)
+  backend.fixed_head_mount = None
   backend.left_pipette = {"name": left_name, "pipetteId": "left-id"} if left_name else None
   backend.right_pipette = {"name": right_name, "pipetteId": "right-id"} if right_name else None
   backend.left_pipette_has_tip = False
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py
index 628fd67113e..dadbb7e14e0 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox.py
@@ -11,7 +11,7 @@
 """
 
 import logging
-from typing import Dict, List, Optional, Tuple, cast
+from typing import Dict, List, Literal, Optional, Tuple, cast
 
 from pylabrobot.io import LOG_LEVEL_IO
 from pylabrobot.legacy.liquid_handling.backends.backend import LiquidHandlerBackend
@@ -126,6 +126,7 @@ def __init__(
     port: int = 31950,
     api_version: str = _OT_DECK_IS_ADDRESSABLE_AREA_VERSION,
     verbose: bool = True,
+    fixed_head_mount: Optional[Literal["left", "right"]] = None,
   ):
     """Initialize the chatterbox.
 
@@ -135,6 +136,7 @@ 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.
+      fixed_head_mount: Mount whose eight nozzles operate together through the fixed-head route.
     """
     # Skip OpentronsOT2Backend.__init__ (it requires ot_api); set up state directly.
     LiquidHandlerBackend.__init__(self)
@@ -149,6 +151,9 @@ def __init__(
     self._right_pipette_name = right_pipette_name
     self.host = host
     self.port = port
+    if fixed_head_mount not in {None, "left", "right"}:
+      raise ValueError("fixed_head_mount must be left, right, or None")
+    self.fixed_head_mount = fixed_head_mount
 
     left = (
       {"name": left_pipette_name, "pipetteId": "chatterbox-left"} if left_pipette_name else None
@@ -171,10 +176,13 @@ def commands(self) -> List[Tuple[str, tuple, dict]]:
     return cast(List[Tuple[str, tuple, dict]], self._ot.calls)
 
   def serialize(self) -> dict:
-    return {
+    data = {
       **LiquidHandlerBackend.serialize(self),
       "left_pipette_name": self._left_pipette_name,
       "right_pipette_name": self._right_pipette_name,
       "host": self.host,
       "port": self.port,
     }
+    if self.fixed_head_mount is not None:
+      data["fixed_head_mount"] = self.fixed_head_mount
+    return data
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py
index 1866ead93f5..eed5e409c90 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_chatterbox_tests.py
@@ -11,9 +11,22 @@
   OpentronsOT2ChatterboxBackend,
   OpentronsOT2Simulator,
 )
-from pylabrobot.resources import 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
+from pylabrobot.legacy.liquid_handling.standard import Mix
+from pylabrobot.resources import (
+  does_tip_tracking,
+  does_volume_tracking,
+  set_tip_tracking,
+  set_volume_tracking,
+)
+from pylabrobot.resources.celltreat import (
+  CellTreat_96_wellplate_350ul_Fb,
+  celltreat_96_wellplate_350uL_Fb,
+)
+from pylabrobot.resources.opentrons import (
+  OTDeck,
+  opentrons_96_filtertiprack_20ul,
+  opentrons_96_tiprack_300ul,
+)
 
 
 def _names(backend: OpentronsOT2ChatterboxBackend):
@@ -131,5 +144,89 @@ async def test_chatterbox_matches_simulator_single_channel(self):
     self.assertEqual(chatterbox_outcome, simulator_outcome)
 
 
+class OpentronsFixedHeadChatterboxTests(unittest.IsolatedAsyncioTestCase):
+  """Exercise full-head operation through the normal frontend and recorded transport."""
+
+  async def test_both_models_and_mounts_preserve_eight_well_state_and_command_sequence(self):
+    """Pickup, mixing, transfer, and discard use the selected mount once per primitive."""
+    previous_tip_tracking = does_tip_tracking()
+    previous_volume_tracking = does_volume_tracking()
+    set_tip_tracking(True)
+    set_volume_tracking(True)
+    try:
+      for model, volume, flow, rack_factory in (
+        ("p20_multi_gen2", 1, 7.6, opentrons_96_filtertiprack_20ul),
+        ("p20_multi_gen2", 20, 7.6, opentrons_96_filtertiprack_20ul),
+        ("p300_multi_gen2", 20, 94, opentrons_96_tiprack_300ul),
+        ("p300_multi_gen2", 300, 94, opentrons_96_tiprack_300ul),
+      ):
+        for mount in ("left", "right"):
+          with self.subTest(model=model, mount=mount, volume=volume):
+            backend = OpentronsOT2ChatterboxBackend(
+              left_pipette_name=model if mount == "left" else None,
+              right_pipette_name=model if mount == "right" else None,
+              fixed_head_mount=mount,
+              verbose=False,
+            )
+            deck = OTDeck()
+            lh = LiquidHandler(backend=backend, deck=deck)
+            await lh.setup(skip_home=True)
+            self.assertEqual(list(lh.head), list(range(8)))
+            tips = rack_factory(name="tips")
+            deck.assign_child_at_slot(tips, slot=1)
+            plate = celltreat_96_wellplate_350uL_Fb(name="plate")
+            deck.assign_child_at_slot(plate, slot=2)
+            for well in plate["A1:H1"]:
+              well.tracker.set_volume(volume + 10)
+            await lh.pick_up_tips(tips["A1:H1"])
+            await lh.aspirate(plate["A1:H1"], vols=[volume] * 8, mix=[Mix(volume, 2, flow)] * 8)
+            await lh.dispense(plate["A2:H2"], vols=[volume] * 8)
+            await lh.discard_tips()
+
+            primitives = [
+              (name, args, kwargs)
+              for name, args, kwargs in backend.commands
+              if name
+              in {
+                "lh.pick_up_tip",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.drop_tip_in_place",
+              }
+            ]
+            self.assertEqual(
+              [name for name, _, _ in primitives],
+              [
+                "lh.pick_up_tip",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.aspirate_in_place",
+                "lh.dispense_in_place",
+                "lh.drop_tip_in_place",
+              ],
+            )
+            self.assertTrue(
+              all(kwargs["pipette_id"] == f"chatterbox-{mount}" for _, _, kwargs in primitives)
+            )
+            self.assertTrue(
+              all(
+                kwargs["volume"] == volume and kwargs["flow_rate"] == flow
+                for name, _, kwargs in primitives
+                if name in {"lh.aspirate_in_place", "lh.dispense_in_place"}
+              )
+            )
+            self.assertEqual([well.tracker.get_used_volume() for well in plate["A1:H1"]], [10] * 8)
+            self.assertEqual(
+              [well.tracker.get_used_volume() for well in plate["A2:H2"]], [volume] * 8
+            )
+            self.assertTrue(all(not tracker.has_tip for tracker in lh.head.values()))
+            self.assertEqual(backend.serialize()["fixed_head_mount"], mount)
+    finally:
+      set_tip_tracking(previous_tip_tracking)
+      set_volume_tracking(previous_volume_tracking)
+
+
 if __name__ == "__main__":
   unittest.main()
diff --git a/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py b/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py
index 262c4d3ef60..64711a63e0b 100644
--- a/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py
+++ b/pylabrobot/legacy/liquid_handling/backends/opentrons_simulator.py
@@ -50,6 +50,7 @@ def __init__(
     """
     # Skip OpentronsOT2Backend.__init__ (requires ot_api); call grandparent directly.
     LiquidHandlerBackend.__init__(self)
+    self.fixed_head_mount = None
 
     pv = OpentronsOT2Backend.pipette_name2volume
     if left_pipette_name is not None and left_pipette_name not in pv:
diff --git a/pylabrobot/legacy/liquid_handling/liquid_handler.py b/pylabrobot/legacy/liquid_handling/liquid_handler.py
index 8a6a0ed2ef2..15d41a96a34 100644
--- a/pylabrobot/legacy/liquid_handling/liquid_handler.py
+++ b/pylabrobot/legacy/liquid_handling/liquid_handler.py
@@ -752,14 +752,6 @@ async def pick_up_tips(
       for tip_spot, offset, tip in zip(tip_spots, offsets, tips)
     ]
 
-    # queue operations on the trackers
-    for channel, op in zip(use_channels, pickups):
-      if self.head[channel].has_tip:
-        raise HasTipError("Channel has tip")
-      if does_tip_tracking() and not op.resource.tracker.is_disabled:
-        op.resource.tracker.remove_tip()
-      self.head[channel].add_tip(op.tip, origin=op.resource, commit=False)
-
     # fix the backend kwargs
     extras = self._check_args(
       self.backend.pick_up_tips,
@@ -770,6 +762,21 @@ async def pick_up_tips(
     for extra in extras:
       del backend_kwargs[extra]
 
+    # Queue the complete request before entering the hardware boundary.
+    try:
+      for channel, op in zip(use_channels, pickups):
+        if self.head[channel].has_tip:
+          raise HasTipError("Channel has tip")
+        if does_tip_tracking() and not op.resource.tracker.is_disabled:
+          op.resource.tracker.remove_tip()
+        self.head[channel].add_tip(op.tip, origin=op.resource, commit=False)
+    except BaseException:
+      for channel, op in zip(use_channels, pickups):
+        if does_tip_tracking() and not op.resource.tracker.is_disabled:
+          op.resource.tracker.rollback()
+        self.head[channel].rollback()
+      raise
+
     # actually pick up the tips
     error: Optional[BaseException] = None
     try:
@@ -898,16 +905,6 @@ async def drop_tips(
       for tip_spot, tip, offset in zip(tip_spots, tips, offsets)
     ]
 
-    # queue operations on the trackers
-    for channel, op in zip(use_channels, drops):
-      if (
-        does_tip_tracking()
-        and isinstance(op.resource, TipSpot)
-        and not op.resource.tracker.is_disabled
-      ):
-        op.resource.tracker.add_tip(op.tip, commit=False)
-      self.head[channel].remove_tip()
-
     # fix the backend kwargs
     extras = self._check_args(
       self.backend.drop_tips,
@@ -918,6 +915,26 @@ async def drop_tips(
     for extra in extras:
       del backend_kwargs[extra]
 
+    try:
+      for channel, op in zip(use_channels, drops):
+        if (
+          does_tip_tracking()
+          and isinstance(op.resource, TipSpot)
+          and not op.resource.tracker.is_disabled
+        ):
+          op.resource.tracker.add_tip(op.tip, commit=False)
+        self.head[channel].remove_tip()
+    except BaseException:
+      for channel, op in zip(use_channels, drops):
+        if (
+          does_tip_tracking()
+          and isinstance(op.resource, TipSpot)
+          and not op.resource.tracker.is_disabled
+        ):
+          op.resource.tracker.rollback()
+        self.head[channel].rollback()
+      raise
+
     # actually drop the tips
     error: Optional[BaseException] = None
     try:
diff --git a/pylabrobot/legacy/tip_tracker.py b/pylabrobot/legacy/tip_tracker.py
index e010d1b5429..f901abd6bad 100644
--- a/pylabrobot/legacy/tip_tracker.py
+++ b/pylabrobot/legacy/tip_tracker.py
@@ -199,15 +199,20 @@ def serialize(self) -> dict:
     }
 
   def load_state(self, state: dict) -> None:
-    """Load a saved tip tracker state."""
+    """Load committed and pending tips, including the committed tip's liquid state."""
 
     tip_data, pending_tip_data = state.get("tip"), state.get("pending_tip")
     tip = Tip.deserialize(tip_data) if tip_data is not None else None
     pending_tip = Tip.deserialize(pending_tip_data) if pending_tip_data is not None else None
-    self._put(pending_tip)
     same = (tip is None) == (pending_tip is None) and (
       tip is None or tip.serialize() == cast(Tip, pending_tip).serialize()
     )
+    tip_state = state.get("tip_state")
+    if tip is not None and tip_state is not None:
+      tip.tracker.load_state(tip_state)
+    if same:
+      pending_tip = tip
+    self._put(pending_tip)
     self._before = _NOTHING_PENDING if same else tip
 
   def get_tip_origin(self) -> Optional["TipSpot"]:
diff --git a/pylabrobot/legacy/tip_tracker_tests.py b/pylabrobot/legacy/tip_tracker_tests.py
index b79ac977b85..e23ad317a16 100644
--- a/pylabrobot/legacy/tip_tracker_tests.py
+++ b/pylabrobot/legacy/tip_tracker_tests.py
@@ -8,6 +8,7 @@
 from pylabrobot.resources import Coordinate
 from pylabrobot.resources.hamilton import HamiltonTip, hamilton_tip_300uL
 from pylabrobot.resources.hamilton.tip_creators import TIP_DIAMETER, TipSize
+from pylabrobot.resources.opentrons import opentrons_96_tiprack_300ul
 from pylabrobot.resources.tip import Tip
 
 
@@ -76,6 +77,64 @@ def test_load_empty_state(self):
     with self.assertRaises(NoTipError):
       tracker.get_tip()
 
+  def test_load_committed_tip_spot_state_remains_committed(self):
+    """A restored occupied spot retains its committed tip and liquid volume."""
+    spot = opentrons_96_tiprack_300ul("tips").get_item("A1")
+    spot.tracker.get_tip().tracker.set_volume(17)
+    state = spot.serialize_state()
+
+    spot.load_state(state)
+
+    self.assertEqual(spot.serialize_state(), state)
+    spot.tracker.rollback()
+    self.assertEqual(spot.serialize_state(), state)
+
+  def test_load_pending_tip_spot_removal_preserves_rollback(self):
+    """Rollback restores a removed tip with its saved liquid volume."""
+    spot = opentrons_96_tiprack_300ul("tips").get_item("A1")
+    spot.tracker.get_tip().tracker.set_volume(17)
+    committed = spot.serialize_state()
+    spot.tracker.remove_tip(commit=False)
+    pending = spot.serialize_state()
+
+    spot.load_state(pending)
+
+    self.assertFalse(spot.tracker.has_tip)
+    self.assertEqual(spot.tracker.get_tip().name, committed["tip"]["name"])
+    spot.tracker.rollback()
+    self.assertEqual(spot.serialize_state(), committed)
+
+  def test_load_channel_tip_preserves_pending_liquid_state(self):
+    """Restoration preserves uncommitted liquid changes until explicit rollback."""
+    tracker = TipTracker("channel")
+    tracker.add_tip(self.tip)
+    self.tip.tracker.set_volume(4)
+    self.tip.tracker.add_liquid(3)
+    saved = tracker.serialize()
+
+    restored = TipTracker("restored")
+    restored.load_state(saved)
+
+    self.assertEqual(restored.serialize(), saved)
+    restored.get_tip().tracker.rollback()
+    self.assertEqual(restored.get_tip().tracker.get_used_volume(), 4)
+
+  def test_load_pending_pickup_preserves_tip_transaction(self):
+    """A pending pickup remains pending and can be rolled back after restoration."""
+    tracker = TipTracker("channel")
+    tracker.add_tip(self.tip, commit=False)
+    saved = tracker.serialize()
+
+    restored = TipTracker("restored")
+    restored.load_state(saved)
+
+    self.assertEqual(restored.serialize(), saved)
+    self.assertTrue(restored.has_tip)
+    with self.assertRaises(NoTipError):
+      restored.get_tip()
+    restored.rollback()
+    self.assertFalse(restored.has_tip)
+
   def test_a_pending_removal_keeps_the_tip_readable_until_it_commits(self):
     """A liquid handler removes a spot's tip before its backend asks the spot which tip it is."""
     tracker = TipTracker(thing="tester")