feat(api): Return a defined tipPhysicallyMissing error from pickUpTip commands

api/src/opentrons/protocol_engine/actions/actions.py

@@ -14,7 +14,12 @@
 
 from opentrons_shared_data.errors import EnumeratedError
 
-from ..commands import Command, CommandCreate, CommandPrivateResult
+from ..commands import (
+    Command,
+    CommandCreate,
+    CommandDefinedErrorData,
+    CommandPrivateResult,
+)
 from ..error_recovery_policy import ErrorRecoveryType
 from ..notes.notes import CommandNote
 from ..types import (
@@ -158,13 +163,25 @@ class FailCommandAction:
     """An ID to assign to the command's error.
 
     Must be unique to this occurrence of the error.
+
+    todo(mm, 2024-05-13): This is redundant with `error` when it's a defined error.
     """
 
     failed_at: datetime
-    """When the command failed."""
+    """When the command failed.
+
+    todo(mm, 2024-05-13): This is redundant with `error` when it's a defined error.
+    """
 
-    error: EnumeratedError
-    """The underlying exception that caused this command to fail."""
+    error: Union[CommandDefinedErrorData, EnumeratedError]
+    """The error that caused the command to fail.
+
+    If it was a defined error, this should be the `DefinedErrorData` that the command
+    returned.
+
+    If it was an undefined error, this should be the underlying exception
+    that caused the command to fail, represented as an `EnumeratedError`.
+    """
 
     notes: List[CommandNote]
     """Overwrite the command's `.notes` with these."""

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

@@ -37,7 +37,7 @@
     CommandResult,
     CommandType,
     CommandPrivateResult,
-    CommandT,
+    CommandDefinedErrorData,
 )
 
 from .aspirate import (
@@ -332,10 +332,9 @@
     "CommandResult",
     "CommandType",
     "CommandPrivateResult",
-    "CommandT",
+    "CommandDefinedErrorData",
     # base interfaces
     "AbstractCommandImpl",
-    "AbstractCommandWithPrivateResultImpl",
     "BaseCommand",
     "BaseCommandCreate",
     "CommandStatus",

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

@@ -22,6 +22,7 @@
 
 from opentrons.hardware_control import HardwareControlAPI
 
+from ..resources import ModelUtils
 from ..errors import ErrorOccurrence
 from ..notes import CommandNote, CommandNoteAdder
 
@@ -264,6 +265,7 @@ def __init__(
         tip_handler: execution.TipHandler,
         run_control: execution.RunControlHandler,
         rail_lights: execution.RailLightsHandler,
+        model_utils: ModelUtils,
         status_bar: execution.StatusBarHandler,
         command_note_adder: CommandNoteAdder,
     ) -> None:

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

@@ -1,10 +1,12 @@
 """Union types of concrete command definitions."""
 
-from typing import Union, TypeVar
+from typing import Union
 from typing_extensions import Annotated
 
 from pydantic import Field
 
+from .command import DefinedErrorData
+
 from . import heater_shaker
 from . import magnetic_module
 from . import temperature_module
@@ -203,6 +205,8 @@
     PickUpTipCreate,
     PickUpTipResult,
     PickUpTipCommandType,
+    TipPhysicallyMissingError,
+    TipPhysicallyMissingErrorInternalData,
 )
 
 from .touch_tip import (
@@ -624,4 +628,8 @@
     ConfigureNozzleLayoutPrivateResult,
 ]
 
-CommandT = TypeVar("CommandT", bound=Command)
+# All `DefinedErrorData`s that implementations will actually return in practice.
+# There's just one right now, but this will eventually be a Union.
+CommandDefinedErrorData = DefinedErrorData[
+    TipPhysicallyMissingError, TipPhysicallyMissingErrorInternalData
+]

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

@@ -1,17 +1,28 @@
 """Pick up tip command request, result, and implementation models."""
 from __future__ import annotations
+from dataclasses import dataclass
+from opentrons_shared_data.errors import ErrorCodes
 from pydantic import Field
-from typing import TYPE_CHECKING, Optional, Type
+from typing import TYPE_CHECKING, Optional, Type, Union
 from typing_extensions import Literal
 
+from opentrons.protocol_engine.errors.exceptions import TipNotAttachedError
+
+from ..errors import ErrorOccurrence
+from ..resources import ModelUtils
 from ..types import DeckPoint
 from .pipetting_common import (
     PipetteIdMixin,
     WellLocationMixin,
     DestinationPositionResult,
 )
-from .command import AbstractCommandImpl, BaseCommand, BaseCommandCreate, SuccessData
-from ..errors.error_occurrence import ErrorOccurrence
+from .command import (
+    AbstractCommandImpl,
+    BaseCommand,
+    BaseCommandCreate,
+    DefinedErrorData,
+    SuccessData,
+)
 
 if TYPE_CHECKING:
     from ..state import StateView
@@ -50,25 +61,56 @@ class PickUpTipResult(DestinationPositionResult):
     )
 
 
-class PickUpTipImplementation(
-    AbstractCommandImpl[PickUpTipParams, SuccessData[PickUpTipResult, None]]
-):
+class TipPhysicallyMissingError(ErrorOccurrence):
+    """Returned when sensors determine that no tip was physically picked up.
+
+    That space in the tip rack is marked internally as not having any tip,
+    as if the tip were consumed by a pickup.
+
+    The pipette will act as if no tip was picked up. So, you won't be able to aspirate
+    anything, and movement commands will assume there is no tip hanging off the bottom
+    of the pipette.
+    """
+
+    errorType: Literal["tipPhysicallyMissing"] = "tipPhysicallyMissing"
+    errorCode: str = ErrorCodes.TIP_PICKUP_FAILED.value.code
+    detail: str = "No tip detected."
+
+
+@dataclass
+class TipPhysicallyMissingErrorInternalData:
+    """Internal-to-ProtocolEngine data about a TipPhysicallyMissingError."""
+
+    pipette_id: str
+    labware_id: str
+    well_name: str
+
+
+_ExecuteReturn = Union[
+    SuccessData[PickUpTipResult, None],
+    DefinedErrorData[TipPhysicallyMissingError, TipPhysicallyMissingErrorInternalData],
+]
+
+
+class PickUpTipImplementation(AbstractCommandImpl[PickUpTipParams, _ExecuteReturn]):
     """Pick up tip command implementation."""
 
     def __init__(
         self,
         state_view: StateView,
         tip_handler: TipHandler,
+        model_utils: ModelUtils,
         movement: MovementHandler,
         **kwargs: object,
     ) -> None:
         self._state_view = state_view
         self._tip_handler = tip_handler
+        self._model_utils = model_utils
         self._movement = movement
 
     async def execute(
         self, params: PickUpTipParams
-    ) -> SuccessData[PickUpTipResult, None]:
+    ) -> Union[SuccessData[PickUpTipResult, None], _ExecuteReturn]:
         """Move to and pick up a tip using the requested pipette."""
         pipette_id = params.pipetteId
         labware_id = params.labwareId
@@ -82,24 +124,46 @@ async def execute(
             well_location=well_location,
         )
 
-        tip_geometry = await self._tip_handler.pick_up_tip(
-            pipette_id=pipette_id,
-            labware_id=labware_id,
-            well_name=well_name,
-        )
-
-        return SuccessData(
-            public=PickUpTipResult(
-                tipVolume=tip_geometry.volume,
-                tipLength=tip_geometry.length,
-                tipDiameter=tip_geometry.diameter,
-                position=DeckPoint(x=position.x, y=position.y, z=position.z),
-            ),
-            private=None,
-        )
-
-
-class PickUpTip(BaseCommand[PickUpTipParams, PickUpTipResult, ErrorOccurrence]):
+        try:
+            tip_geometry = await self._tip_handler.pick_up_tip(
+                pipette_id=pipette_id,
+                labware_id=labware_id,
+                well_name=well_name,
+            )
+        except TipNotAttachedError as e:
+            return DefinedErrorData(
+                public=TipPhysicallyMissingError(
+                    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,
+                        )
+                    ],
+                ),
+                private=TipPhysicallyMissingErrorInternalData(
+                    pipette_id=pipette_id,
+                    labware_id=labware_id,
+                    well_name=well_name,
+                ),
+            )
+        else:
+            return SuccessData(
+                public=PickUpTipResult(
+                    tipVolume=tip_geometry.volume,
+                    tipLength=tip_geometry.length,
+                    tipDiameter=tip_geometry.diameter,
+                    position=DeckPoint(x=position.x, y=position.y, z=position.z),
+                ),
+                private=None,
+            )
+
+
+class PickUpTip(
+    BaseCommand[PickUpTipParams, PickUpTipResult, TipPhysicallyMissingError]
+):
     """Pick up tip command model."""
 
     commandType: PickUpTipCommandType = "pickUpTip"

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

@@ -53,11 +53,11 @@ class SavePositionImplementation(
     def __init__(
         self,
         gantry_mover: GantryMover,
-        model_utils: Optional[ModelUtils] = None,
+        model_utils: ModelUtils,
         **kwargs: object,
     ) -> None:
         self._gantry_mover = gantry_mover
-        self._model_utils = model_utils or ModelUtils()
+        self._model_utils = model_utils
 
     async def execute(
         self, params: SavePositionParams

api/src/opentrons/protocol_engine/error_recovery_policy.py

@@ -1,12 +1,15 @@
 # noqa: D100
 
 import enum
-from typing import Protocol
-
-from opentrons_shared_data.errors import EnumeratedError, ErrorCodes
+from typing import Optional, Protocol
 
 from opentrons.config import feature_flags as ff
-from opentrons.protocol_engine.commands import Command
+from opentrons.protocol_engine.commands import (
+    Command,
+    CommandDefinedErrorData,
+    PickUpTip,
+)
+from opentrons.protocol_engine.commands.pick_up_tip import TipPhysicallyMissingError
 
 
 class ErrorRecoveryType(enum.Enum):
@@ -36,17 +39,24 @@ class ErrorRecoveryPolicy(Protocol):
     This describes a function that Protocol Engine calls after each command failure,
     with the details of that failure. The implementation should inspect those details
     and return an appropriate `ErrorRecoveryType`.
+
+    Args:
+        failed_command: The command that failed, in its final `status=="failed"` state.
+        defined_error_data: If the command failed with a defined error, details about
+            that error. If the command failed with an undefined error, `None`.
+            By design, this callable isn't given details about undefined errors,
+            since it would be fragile to rely on them.
     """
 
     @staticmethod
     def __call__(  # noqa: D102
-        failed_command: Command, exception: Exception
+        failed_command: Command, defined_error_data: Optional[CommandDefinedErrorData]
     ) -> ErrorRecoveryType:
         ...
 
 
 def error_recovery_by_ff(
-    failed_command: Command, exception: Exception
+    failed_command: Command, defined_error_data: Optional[CommandDefinedErrorData]
 ) -> ErrorRecoveryType:
     """Use API feature flags to decide how to handle an error.
 
@@ -56,20 +66,20 @@ def error_recovery_by_ff(
     # todo(mm, 2024-03-18): Do we need to do anything explicit here to disable
     # error recovery on the OT-2?
     if ff.enable_error_recovery_experiments() and _is_recoverable(
-        failed_command, exception
+        failed_command, defined_error_data
     ):
         return ErrorRecoveryType.WAIT_FOR_RECOVERY
     else:
         return ErrorRecoveryType.FAIL_RUN
 
 
-def _is_recoverable(failed_command: Command, exception: Exception) -> bool:
+def _is_recoverable(
+    failed_command: Command, error_data: Optional[CommandDefinedErrorData]
+) -> bool:
     if (
-        failed_command.commandType == "pickUpTip"
-        and isinstance(exception, EnumeratedError)
-        # Hack(?): It seems like this should be ErrorCodes.TIP_PICKUP_FAILED, but that's
-        # not what gets raised in practice.
-        and exception.code == ErrorCodes.UNEXPECTED_TIP_REMOVAL
+        isinstance(failed_command, PickUpTip)
+        and error_data is not None
+        and isinstance(error_data.public, TipPhysicallyMissingError)
     ):
         return True
     else: