fix(liquid-handling): report committed tip state from chatterbox sensors

pylabrobot/legacy/liquid_handling/backends/chatterbox.py

@@ -230,13 +230,16 @@ async def drop_resource(self, drop: ResourceDrop):
     print(f"Dropping resource: {drop}")
 
   async def request_tip_presence(self) -> List[Optional[bool]]:
-    """Return tip presence based on the tip tracker state.
+    """Return simulated sleeve-sensor tip presence from committed tracker state.
+
+    Pending pickup/drop operations are excluded so error recovery can distinguish
+    intended state from the last committed (simulated physical) state.
 
     Returns:
       A list of length `num_channels` where each element is `True` if a tip is mounted,
       `False` if not, or `None` if unknown.
     """
-    return [self.head[ch].has_tip for ch in range(self.num_channels)]
+    return [self.head[ch].has_committed_tip for ch in range(self.num_channels)]
 
   def can_pick_up_tip(self, channel_idx: int, tip: Tip) -> bool:
     return True

pylabrobot/legacy/liquid_handling/backends/chatterbox_tests.py

@@ -4,6 +4,7 @@
 from pylabrobot.legacy.liquid_handling.backends.chatterbox import (
   LiquidHandlerChatterboxBackend,
 )
+from pylabrobot.legacy.liquid_handling.errors import ChannelizedError
 from pylabrobot.resources import (
   Coordinate,
   cor_96_wellplate_360uL_Fb,
@@ -65,3 +66,44 @@ async def test_dispense96(self):
 
   async def test_move(self):
     await self.lh.move_resource(self.plate, Coordinate(0, 0, 0))
+
+  async def test_failed_pickup_does_not_commit_pending_tips(self):
+    async def fail_pickup(*args, **kwargs):
+      raise RuntimeError("simulated pickup failure")
+
+    self.backend.pick_up_tips = fail_pickup  # type: ignore[method-assign]
+    with self.assertRaises(RuntimeError):
+      await self.lh.pick_up_tips(self.tip_rack["A1"])
+    self.assertFalse(self.lh.head[0].has_tip)
+
+  async def test_failed_drop_does_not_commit_pending_remove(self):
+    await self.lh.pick_up_tips(self.tip_rack["A1"])
+    self.assertTrue(self.lh.head[0].has_tip)
+
+    async def fail_drop(*args, **kwargs):
+      raise RuntimeError("simulated drop failure")
+
+    self.backend.drop_tips = fail_drop  # type: ignore[method-assign]
+    with self.assertRaises(RuntimeError):
+      await self.lh.drop_tips(self.tip_rack["A1"])
+    self.assertTrue(self.lh.head[0].has_tip)
+
+  async def test_failed_multi_channel_pickup_rolls_back_all_channels(self):
+    async def fail_pickup(*args, **kwargs):
+      raise RuntimeError("simulated pickup failure")
+
+    self.backend.pick_up_tips = fail_pickup  # type: ignore[method-assign]
+    with self.assertRaises(RuntimeError):
+      await self.lh.pick_up_tips(self.tip_rack["A1", "B1"])
+    self.assertFalse(self.lh.head[0].has_tip)
+    self.assertFalse(self.lh.head[1].has_tip)
+
+  async def test_failed_pickup_presence_query_overrides_channelized_error(self):
+    async def fail_pickup(*args, **kwargs):
+      raise ChannelizedError(errors={0: Exception("channel 0 failed")})
+
+    self.backend.pick_up_tips = fail_pickup  # type: ignore[method-assign]
+    with self.assertRaises(ChannelizedError):
+      await self.lh.pick_up_tips(self.tip_rack["A1", "B1"])
+    self.assertFalse(self.lh.head[0].has_tip)
+    self.assertFalse(self.lh.head[1].has_tip)

pylabrobot/legacy/liquid_handling/backends/hamilton/STAR_chatterbox.py

@@ -288,13 +288,16 @@ async def request_working_envelopes_per_arm(
   # # # # # # # # 1_000 uL Channel: Basic Commands # # # # # # # #
 
   async def request_tip_presence(self) -> List[Optional[bool]]:
-    """Return mock tip presence based on the tip tracker state.
+    """Return mock sleeve-sensor tip presence from committed tracker state.
+
+    Pending pickup/drop operations are excluded so error recovery can distinguish
+    intended state from the last committed (simulated physical) state.
 
     Returns:
       A list of length `num_channels` where each element is `True` if a tip is mounted,
       `False` if not, or `None` if unknown.
     """
-    return [self.head[ch].has_tip for ch in range(self.num_channels)]
+    return [self.head[ch].has_committed_tip for ch in range(self.num_channels)]
 
   async def request_z_pos_channel_n(self, channel: int) -> float:
     return 285.0
@@ -403,7 +406,7 @@ async def head96_request_z_acceleration(self) -> float:
     return 400.0
 
   async def head96_request_tip_presence(self) -> int:
-    """Mock 96-head tip presence from the tip tracker: 1 if any channel holds a tip, else 0.
+    """Mock 96-head tip presence from committed tracker state: 1 if any channel holds a tip.
 
     Raises if tip tracking is disabled, since the tracker is then not updated and has no state to report.
     """
@@ -412,7 +415,7 @@ async def head96_request_tip_presence(self) -> int:
         "cannot report 96-head tip presence with tip tracking disabled in simulation; "
         "enable it with set_tip_tracking(True) or call with requires_tip=False"
       )
-    return int(any(tracker.has_tip for tracker in self.head96.values()))
+    return int(any(tracker.has_committed_tip for tracker in self.head96.values()))
 
   # # # # # # # # Extension: iSWAP # # # # # # # #
 

pylabrobot/resources/tip_tracker.py

@@ -56,6 +56,11 @@ def has_tip(self) -> bool:
     """Whether the tip tracker has a tip. Note that this includes pending operations."""
     return self._pending_tip is not None
 
+  @property
+  def has_committed_tip(self) -> bool:
+    """Whether the tip tracker has a committed tip. Pending operations are not included."""
+    return self._tip is not None
+
   def get_tip(self) -> "Tip":
     """Get the tip. Note that does includes pending operations.
 

pylabrobot/resources/tip_tracker_tests.py

@@ -43,3 +43,17 @@ def test_remove_tip(self):
 
     with self.assertRaises(NoTipError):
       tracker.get_tip()
+
+  def test_has_committed_tip_ignores_pending_add(self):
+    tracker = TipTracker(thing="tester")
+    tracker.add_tip(self.tip, commit=False)
+    self.assertEqual(tracker.has_tip, True)
+    self.assertEqual(tracker.has_committed_tip, False)
+
+  def test_has_committed_tip_ignores_pending_remove(self):
+    tracker = TipTracker(thing="tester")
+    tracker.add_tip(self.tip)
+    self.assertEqual(tracker.has_committed_tip, True)
+    tracker.remove_tip(commit=False)
+    self.assertEqual(tracker.has_tip, False)
+    self.assertEqual(tracker.has_committed_tip, True)