From 4317be17b645734357a1edc90a1425bfec8fafea Mon Sep 17 00:00:00 2001
From: Camillo Moschner <camillo.moschner@biocam.guide>
Date: Mon, 13 Jul 2026 14:57:31 +0100
Subject: [PATCH] `STARBackend`: reject aspirations whose pre-wetting or
 transport-air peak overfills the tip

An over-capacity aspiration was caught only by the firmware ("Maximum volume in tip reached"), after the channel had already moved to the well. `aspirate` now checks each channel's two transient peaks (volume + pre_wetting, volume + transport_air; blow-out air counts toward neither) against `tip.maximal_volume` before assembling the command, and raises `ValueError` naming every peak on every channel that exceeds it. The bound is exclusive. Adds three tests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 .../backends/hamilton/STAR_backend.py         | 21 +++++++++++
 .../backends/hamilton/STAR_tests.py           | 35 +++++++++++++++++++
 2 files changed, 56 insertions(+)

diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py
index 97ce7093dfd..f9d2ab5b192 100644
--- a/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py
+++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_backend.py
@@ -3479,6 +3479,27 @@ async def aspirate(
         f"Well bottom: {well_bottoms}, liquid height: {liquid_heights}, surface_following_distance: {surface_following_distance}, minimum_height: {minimum_height}"
       )
 
+    # A tip fills to one of two transient peaks that never coexist: volume + pre_wetting
+    # (pre-wet over-aspiration) and volume + transport_air (air gap drawn above the liquid);
+    # blow-out air counts toward neither.
+    over_capacity = []
+    for i, op in enumerate(ops):
+      for label, extra in (
+        ("pre-wetting", pre_wetting_volume[i]),
+        ("transport-air", transport_air_volume[i]),
+      ):
+        peak = volumes[i] + extra
+        if peak > op.tip.maximal_volume:
+          over_capacity.append((i, label, peak, op.tip.maximal_volume))
+    if over_capacity:
+      raise ValueError(
+        "Aspiration would exceed tip capacity: "
+        + "; ".join(
+          f"channel {i} {label} peak {peak:.1f} uL > tip maximal volume {tip_max:.1f} uL"
+          for i, label, peak, tip_max in over_capacity
+        )
+      )
+
     try:
       return await self.aspirate_pip(
         aspiration_type=[0 for _ in range(n)],
diff --git a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py
index 78574f81406..b5a6ce25591 100644
--- a/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py
+++ b/pylabrobot/liquid_handling/backends/hamilton/STAR_tests.py
@@ -988,6 +988,41 @@ async def test_single_channel_aspiration(self):
       ]
     )
 
+  async def test_aspirate_rejects_over_tip_capacity(self):
+    self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})  # 300 uL filter tip, max 360
+    assert self.plate.lid is not None
+    self.plate.lid.unassign()
+    well = self.plate.get_item("A1")
+    # pre-wetting peak (volume + pre_wetting) exceeds the tip max
+    well.tracker.set_volume(300)
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([well], vols=[100], pre_wetting_volume=[300])
+    # transport-air peak (volume + transport_air) exceeds the tip max
+    well.tracker.set_volume(300)
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([well], vols=[100], transport_air_volume=[300])
+
+  async def test_aspirate_capacity_ignores_blow_out(self):
+    self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})  # 300 uL filter tip, max 360
+    assert self.plate.lid is not None
+    self.plate.lid.unassign()
+    well = self.plate.get_item("A1")
+    well.tracker.set_volume(200)
+    # blow-out counts toward neither peak, so a large blow-out stays within capacity
+    await self.lh.aspirate([well], vols=[100], blow_out_air_volume=[300])
+
+  async def test_aspirate_capacity_boundary_is_exclusive(self):
+    self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})  # 300 uL filter tip, max 360
+    assert self.plate.lid is not None
+    self.plate.lid.unassign()
+    well = self.plate.get_item("A1")
+    # a peak exactly at the tip's maximal volume is allowed; just above it is rejected (>)
+    well.tracker.set_volume(400)
+    await self.lh.aspirate([well], vols=[360], disable_volume_correction=[True])
+    well.tracker.set_volume(400)
+    with self.assertRaises(ValueError):
+      await self.lh.aspirate([well], vols=[361], disable_volume_correction=[True])
+
   async def test_single_channel_aspiration_liquid_height(self):
     self.lh.update_head_state({0: self.tip_rack.get_tip("A1")})
     # TODO: Hamilton liquid classes