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,43 @@ 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_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)
+
+  async def test_request_tip_presence_uses_committed_state(self):
+    tip = self.tip_rack.get_item("A1").get_tip()
+    self.lh.head[0].add_tip(tip, commit=False)
+    self.assertEqual((await self.backend.request_tip_presence())[0], False)
+
+    self.lh.head[0].commit()
+    self.lh.head[0].remove_tip(commit=False)
+    self.assertEqual((await self.backend.request_tip_presence())[0], True)

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

@@ -291,13 +291,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
@@ -406,7 +409,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.
     """
@@ -415,7 +418,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 # # # # # # # #
 
@@ -470,18 +473,21 @@ async def slow_iswap(self, wrist_velocity: int = 20_000, gripper_velocity: int =
   # # # # # # # # Liquid Level Detection (LLD) # # # # # # # #
 
   async def request_tip_len_on_channel(self, channel_idx: int) -> float:
-    """Return tip length from the tip tracker.
+    """Return simulated measured tip length from committed tracker state.
+
+    Pending pickup/drop operations are excluded so the query matches the last
+    committed (simulated physical) tip.
 
     Args:
       channel_idx: Index of the pipetting channel (0-indexed).
 
     Returns:
-      The tip length in mm from the tip tracker.
+      The tip length in mm from the committed tip tracker state.
 
     Raises:
-      NoTipError: If no tip is present on the channel (via tip tracker).
+      NoTipError: If no committed tip is present on the channel.
     """
-    tip = self.head[channel_idx].get_tip()
+    tip = self.head[channel_idx].get_committed_tip()
     return tip.total_tip_length
 
   async def position_channels_in_y_direction(self, ys, make_space=True):

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

@@ -32,6 +32,7 @@
   set_tip_tracking,
 )
 from pylabrobot.resources.barcode import Barcode
+from pylabrobot.resources.errors import NoTipError
 from pylabrobot.resources.greiner import Greiner_384_wellplate_28ul_Fb
 from pylabrobot.resources.hamilton import STARDeck, STARLetDeck, hamilton_96_tiprack_300uL_filter
 
@@ -2770,3 +2771,39 @@ async def test_rejects_target_on_absent_right_arm(self):
     # The default single-arm STAR has no right X-arm, so a right-arm target is rejected.
     with self.assertRaises(ValueError):
       self.star._check_x_arm_reachable(400.0, "right")
+
+
+class TestSTARChatterboxTipPresenceRecovery(unittest.IsolatedAsyncioTestCase):
+  """Simulated STAR tip queries must report committed tracker state, not pending."""
+
+  async def asyncSetUp(self):
+    self.backend = STARChatterboxBackend()
+    self.deck = STARLetDeck()
+    self.lh = LiquidHandler(self.backend, deck=self.deck)
+    self.tip_car = TIP_CAR_480_A00(name="tip carrier")
+    self.tip_car[1] = self.tip_rack = hamilton_96_tiprack_300uL_filter(name="tip_rack_01")
+    self.deck.assign_child_resource(self.tip_car, rails=1)
+    await self.lh.setup()
+
+  async def asyncTearDown(self):
+    await self.lh.stop()
+
+  async def test_simulated_tip_queries_use_committed_state(self):
+    tip = self.tip_rack.get_item("A1").get_tip()
+    self.lh.head[0].add_tip(tip, commit=False)
+    self.assertEqual((await self.backend.request_tip_presence())[0], False)
+    with self.assertRaises(NoTipError):
+      await self.backend.request_tip_len_on_channel(0)
+
+    self.lh.head[0].commit()
+    self.assertEqual(await self.backend.request_tip_len_on_channel(0), tip.total_tip_length)
+    self.lh.head[0].remove_tip(commit=False)
+    self.assertEqual(await self.backend.request_tip_len_on_channel(0), tip.total_tip_length)
+
+    set_tip_tracking(enabled=True)
+    try:
+      self.assertEqual(await self.backend.head96_request_tip_presence(), 0)
+      self.lh.head96[0].add_tip(tip, commit=False)
+      self.assertEqual(await self.backend.head96_request_tip_presence(), 0)
+    finally:
+      set_tip_tracking(enabled=False)

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.
 
@@ -67,6 +72,16 @@ def get_tip(self) -> "Tip":
       raise NoTipError(f"{self.thing} does not have a tip.")
     return self._tip
 
+  def get_committed_tip(self) -> "Tip":
+    """Get the committed tip. Pending operations are not included.
+
+    Raises:
+      NoTipError: If the tip tracker does not have a committed tip.
+    """
+    if self._tip is None:
+      raise NoTipError(f"{self.thing} does not have a tip.")
+    return self._tip
+
   def disable(self) -> None:
     """Disable the tip tracker."""
     self._is_disabled = True

pylabrobot/resources/tip_tracker_tests.py

@@ -43,3 +43,29 @@ 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)
+
+  def test_get_committed_tip_ignores_pending_add(self):
+    tracker = TipTracker(thing="tester")
+    tracker.add_tip(self.tip, commit=False)
+    with self.assertRaises(NoTipError):
+      tracker.get_committed_tip()
+
+  def test_get_committed_tip_ignores_pending_remove(self):
+    tracker = TipTracker(thing="tester")
+    tracker.add_tip(self.tip)
+    tracker.remove_tip(commit=False)
+    self.assertEqual(tracker.get_committed_tip(), self.tip)