feat(api): Handle overpressures in dispense (#15754)

As in aspirate, we handle overpressure errors and turn them into
DefinedError return values, allowing us to recover from overpressures
via client-driven error recovery.

Closes EXEC-498

api/src/opentrons/protocol_engine/commands/dispense.py

@@ -1,8 +1,10 @@
 """Dispense command request, result, and implementation models."""
 from __future__ import annotations
-from typing import TYPE_CHECKING, Optional, Type
+from typing import TYPE_CHECKING, Optional, Type, Union
 from typing_extensions import Literal
 
+from opentrons_shared_data.errors.exceptions import PipetteOverpressureError
+
 from pydantic import Field
 
 from ..types import DeckPoint
@@ -13,12 +15,21 @@
     WellLocationMixin,
     BaseLiquidHandlingResult,
     DestinationPositionResult,
+    OverpressureError,
+    OverpressureErrorInternalData,
+)
+from .command import (
+    AbstractCommandImpl,
+    BaseCommand,
+    BaseCommandCreate,
+    DefinedErrorData,
+    SuccessData,
 )
-from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
 from ..errors.error_occurrence import ErrorOccurrence
 
 if TYPE_CHECKING:
     from ..execution import MovementHandler, PipettingHandler
+    from ..resources import ModelUtils
 
 
 DispenseCommandType = Literal["dispense"]
@@ -41,41 +52,69 @@ class DispenseResult(BaseLiquidHandlingResult, DestinationPositionResult):
     pass
 
 
-class DispenseImplementation(
-    AbstractCommandImpl[DispenseParams, SuccessData[DispenseResult, None]]
-):
+_ExecuteReturn = Union[
+    SuccessData[DispenseResult, None],
+    DefinedErrorData[OverpressureError, OverpressureErrorInternalData],
+]
+
+
+class DispenseImplementation(AbstractCommandImpl[DispenseParams, _ExecuteReturn]):
     """Dispense command implementation."""
 
     def __init__(
-        self, movement: MovementHandler, pipetting: PipettingHandler, **kwargs: object
+        self,
+        movement: MovementHandler,
+        pipetting: PipettingHandler,
+        model_utils: ModelUtils,
+        **kwargs: object,
     ) -> None:
         self._movement = movement
         self._pipetting = pipetting
+        self._model_utils = model_utils
 
-    async def execute(
-        self, params: DispenseParams
-    ) -> SuccessData[DispenseResult, None]:
+    async def execute(self, params: DispenseParams) -> _ExecuteReturn:
         """Move to and dispense to the requested well."""
         position = await self._movement.move_to_well(
             pipette_id=params.pipetteId,
             labware_id=params.labwareId,
             well_name=params.wellName,
             well_location=params.wellLocation,
         )
-        volume = await self._pipetting.dispense_in_place(
-            pipette_id=params.pipetteId,
-            volume=params.volume,
-            flow_rate=params.flowRate,
-            push_out=params.pushOut,
-        )
-
-        return SuccessData(
-            public=DispenseResult(
-                volume=volume,
-                position=DeckPoint(x=position.x, y=position.y, z=position.z),
-            ),
-            private=None,
-        )
+        try:
+            volume = await self._pipetting.dispense_in_place(
+                pipette_id=params.pipetteId,
+                volume=params.volume,
+                flow_rate=params.flowRate,
+                push_out=params.pushOut,
+            )
+        except PipetteOverpressureError as e:
+            return DefinedErrorData(
+                public=OverpressureError(
+                    id=self._model_utils.generate_id(),
+                    createdAt=self._model_utils.get_timestamp(),
+                    wrappedErrors=[
+                        ErrorOccurrence.from_failed(
+                            id=self._model_utils.generate_id(),
+                            createdAt=self._model_utils.get_timestamp(),
+                            error=e,
+                        )
+                    ],
+                    errorInfo={"retryLocation": (position.x, position.y, position.z)},
+                ),
+                private=OverpressureErrorInternalData(
+                    position=DeckPoint.construct(
+                        x=position.x, y=position.y, z=position.z
+                    )
+                ),
+            )
+        else:
+            return SuccessData(
+                public=DispenseResult(
+                    volume=volume,
+                    position=DeckPoint(x=position.x, y=position.y, z=position.z),
+                ),
+                private=None,
+            )
 
 
 class Dispense(BaseCommand[DispenseParams, DispenseResult, ErrorOccurrence]):

api/src/opentrons/protocol_engine/state/pipettes.py

@@ -13,6 +13,7 @@
 )
 from opentrons.protocol_engine.actions.actions import FailCommandAction
 from opentrons.protocol_engine.commands.aspirate import Aspirate
+from opentrons.protocol_engine.commands.dispense import Dispense
 from opentrons.protocol_engine.commands.command import DefinedErrorData
 from opentrons.protocol_engine.commands.pipetting_common import (
     OverpressureError,
@@ -316,7 +317,7 @@ def _update_current_location(  # noqa: C901
             )
         elif (
             isinstance(action, FailCommandAction)
-            and isinstance(action.running_command, Aspirate)
+            and isinstance(action.running_command, (Aspirate, Dispense))
             and isinstance(action.error, DefinedErrorData)
             and isinstance(action.error.public, OverpressureError)
         ):
@@ -412,7 +413,7 @@ def _update_deck_point(
             )
         elif (
             isinstance(action, FailCommandAction)
-            and isinstance(action.running_command, Aspirate)
+            and isinstance(action.running_command, (Aspirate, Dispense))
             and isinstance(action.error, DefinedErrorData)
             and isinstance(action.error.public, OverpressureError)
         ):

api/tests/opentrons/protocol_engine/commands/test_dispense.py

@@ -1,26 +1,47 @@
 """Test dispense commands."""
-from decoy import Decoy
+from datetime import datetime
+
+import pytest
+from decoy import Decoy, matchers
+
+from opentrons_shared_data.errors.exceptions import PipetteOverpressureError
 
 from opentrons.protocol_engine import WellLocation, WellOrigin, WellOffset, DeckPoint
 from opentrons.protocol_engine.execution import MovementHandler, PipettingHandler
 from opentrons.types import Point
 
-from opentrons.protocol_engine.commands.command import SuccessData
+from opentrons.protocol_engine.commands.command import SuccessData, DefinedErrorData
 from opentrons.protocol_engine.commands.dispense import (
     DispenseParams,
     DispenseResult,
     DispenseImplementation,
 )
+from opentrons.protocol_engine.resources.model_utils import ModelUtils
+from opentrons.protocol_engine.commands.pipetting_common import (
+    OverpressureError,
+    OverpressureErrorInternalData,
+)
+
+
+@pytest.fixture
+def subject(
+    movement: MovementHandler,
+    pipetting: PipettingHandler,
+    model_utils: ModelUtils,
+) -> DispenseImplementation:
+    """Get the implementation subject."""
+    return DispenseImplementation(
+        movement=movement, pipetting=pipetting, model_utils=model_utils
+    )
 
 
 async def test_dispense_implementation(
     decoy: Decoy,
     movement: MovementHandler,
     pipetting: PipettingHandler,
+    subject: DispenseImplementation,
 ) -> None:
     """It should move to the target location and then dispense."""
-    subject = DispenseImplementation(movement=movement, pipetting=pipetting)
-
     well_location = WellLocation(
         origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1)
     )
@@ -55,3 +76,65 @@ async def test_dispense_implementation(
         public=DispenseResult(volume=42, position=DeckPoint(x=1, y=2, z=3)),
         private=None,
     )
+
+
+async def test_overpressure_error(
+    decoy: Decoy,
+    movement: MovementHandler,
+    pipetting: PipettingHandler,
+    subject: DispenseImplementation,
+    model_utils: ModelUtils,
+) -> None:
+    """It should return an overpressure error if the hardware API indicates that."""
+    pipette_id = "pipette-id"
+    labware_id = "labware-id"
+    well_name = "well-name"
+    well_location = WellLocation(
+        origin=WellOrigin.BOTTOM, offset=WellOffset(x=0, y=0, z=1)
+    )
+
+    position = Point(x=1, y=2, z=3)
+
+    error_id = "error-id"
+    error_timestamp = datetime(year=2020, month=1, day=2)
+
+    data = DispenseParams(
+        pipetteId=pipette_id,
+        labwareId=labware_id,
+        wellName=well_name,
+        wellLocation=well_location,
+        volume=50,
+        flowRate=1.23,
+    )
+
+    decoy.when(
+        await movement.move_to_well(
+            pipette_id=pipette_id,
+            labware_id=labware_id,
+            well_name=well_name,
+            well_location=well_location,
+        ),
+    ).then_return(position)
+
+    decoy.when(
+        await pipetting.dispense_in_place(
+            pipette_id=pipette_id, volume=50, flow_rate=1.23, push_out=None
+        ),
+    ).then_raise(PipetteOverpressureError())
+
+    decoy.when(model_utils.generate_id()).then_return(error_id)
+    decoy.when(model_utils.get_timestamp()).then_return(error_timestamp)
+
+    result = await subject.execute(data)
+
+    assert result == DefinedErrorData(
+        public=OverpressureError.construct(
+            id=error_id,
+            createdAt=error_timestamp,
+            wrappedErrors=[matchers.Anything()],
+            errorInfo={"retryLocation": (position.x, position.y, position.z)},
+        ),
+        private=OverpressureErrorInternalData(
+            position=DeckPoint(x=position.x, y=position.y, z=position.z)
+        ),
+    )

api/tests/opentrons/protocol_engine/state/test_pipette_store.py

@@ -426,6 +426,43 @@ def test_blow_out_clears_volume(
                 well_name="move-to-well-well-name",
             ),
         ),
+        (
+            FailCommandAction(
+                running_command=cmd.Dispense(
+                    params=cmd.DispenseParams(
+                        pipetteId="pipette-id",
+                        labwareId="dispense-labware-id",
+                        wellName="dispense-well-name",
+                        volume=50,
+                        flowRate=1.23,
+                    ),
+                    id="command-id",
+                    key="command-key",
+                    createdAt=datetime.now(),
+                    status=cmd.CommandStatus.RUNNING,
+                ),
+                error=DefinedErrorData(
+                    public=OverpressureError(
+                        id="error-id",
+                        createdAt=datetime.now(),
+                        errorInfo={"retryLocation": (0, 0, 0)},
+                    ),
+                    private=OverpressureErrorInternalData(
+                        position=DeckPoint(x=0, y=0, z=0)
+                    ),
+                ),
+                command_id="command-id",
+                error_id="error-id",
+                failed_at=datetime.now(),
+                notes=[],
+                type=ErrorRecoveryType.WAIT_FOR_RECOVERY,
+            ),
+            CurrentWell(
+                pipette_id="pipette-id",
+                labware_id="dispense-labware-id",
+                well_name="dispense-well-name",
+            ),
+        ),
     ),
 )
 def test_movement_commands_update_current_well(
@@ -902,6 +939,37 @@ def test_add_pipette_config(
             ),
             private_result=None,
         ),
+        FailCommandAction(
+            running_command=cmd.Dispense(
+                params=cmd.DispenseParams(
+                    pipetteId="pipette-id",
+                    labwareId="labware-id",
+                    wellName="well-name",
+                    volume=125,
+                    flowRate=1.23,
+                ),
+                id="command-id",
+                key="command-key",
+                createdAt=datetime.now(),
+                status=cmd.CommandStatus.RUNNING,
+            ),
+            error=DefinedErrorData(
+                public=OverpressureError(
+                    id="error-id",
+                    detail="error-detail",
+                    createdAt=datetime.now(),
+                    errorInfo={"retryLocation": (11, 22, 33)},
+                ),
+                private=OverpressureErrorInternalData(
+                    position=DeckPoint(x=11, y=22, z=33)
+                ),
+            ),
+            command_id="command-id",
+            error_id="error-id",
+            failed_at=datetime.now(),
+            notes=[],
+            type=ErrorRecoveryType.WAIT_FOR_RECOVERY,
+        ),
     ),
 )
 def test_movement_commands_update_deck_point(