feat(api): Pause when pick_up_tip() errors in a Python protocol

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -408,13 +408,19 @@ def pick_up_tip(
             well_name=well_name,
             well_location=well_location,
         )
-        self._engine_client.pick_up_tip(
+        self._engine_client.pick_up_tip_wait_for_recovery(
             pipette_id=self._pipette_id,
             labware_id=labware_id,
             well_name=well_name,
             well_location=well_location,
         )
+        # FIX BEFORE MERGE?: If the pickUpTip fails, the tip tracker (which is sort of
+        # kind of in Protocol Engine) doesn't think the tip has been consumed. So,
+        # the next PAPI pick_up_tip() after the error recovery will try to pick up
+        # the same tip, diverging from the protocol's analysis.
 
+        # FIX BEFORE MERGE: We should probably only set_last_location() if the
+        # pickUpTip succeeded.
         self._protocol_core.set_last_location(location=location, mount=self.get_mount())
 
     def drop_tip(

api/src/opentrons/protocol_engine/clients/sync_client.py

@@ -294,6 +294,25 @@ def pick_up_tip(
 
         return cast(commands.PickUpTipResult, result)
 
+    # FIX BEFORE MERGE?: See if we can cut down on this duplication.
+    # We do not want to have to do this for every method here.
+    def pick_up_tip_wait_for_recovery(
+        self,
+        pipette_id: str,
+        labware_id: str,
+        well_name: str,
+        well_location: WellLocation,
+    ) -> None:
+        request = commands.PickUpTipCreate(
+            params=commands.PickUpTipParams(
+                pipetteId=pipette_id,
+                labwareId=labware_id,
+                wellName=well_name,
+                wellLocation=well_location,
+            )
+        )
+        self._transport.execute_command_wait_for_recovery(request=request)
+
     def drop_tip(
         self,
         pipette_id: str,

api/src/opentrons/protocol_engine/clients/transports.py

@@ -79,6 +79,12 @@ def execute_command(self, request: CommandCreate) -> CommandResult:
 
         return command.result
 
+    def execute_command_wait_for_recovery(self, request: CommandCreate) -> None:
+        run_coroutine_threadsafe(
+            self._engine.add_and_execute_command_wait_for_recovery(request=request),
+            loop=self._loop,
+        ).result()
+
     @overload
     def call_method(
         self,

api/src/opentrons/protocol_engine/protocol_engine.py

@@ -242,6 +242,27 @@ async def add_and_execute_command(
         await self.wait_for_command(command.id)
         return self._state_store.commands.get(command.id)
 
+    async def add_and_execute_command_wait_for_recovery(
+        self, request: commands.CommandCreate
+    ) -> None:
+        """Like `add_and_execute_command()`, except wait for error recovery.
+
+        Unlike `add_and_execute_command()`, if the command fails, this will not
+        immediately return the failed command. Instead, if the error is recoverable,
+        it will wait until error recovery has completed (e.g. when some other task
+        calls `self.resume_from_recovery()`).
+        """
+        command = self.add_command(request)
+        await self.wait_for_command(command_id=command.id)
+        await self._state_store.wait_for(
+            # Per the warnings on `wait_for()`, we want our condition function to
+            # specifically check that *this* command's recovery has completed,
+            # rather than just checking that the overall run state
+            # != "awaiting-recovery."
+            self.state_view.commands.get_command_recovery_complete,
+            command_id=command.id,
+        )
+
     def estop(
         self,
         # TODO(mm, 2024-03-26): Maintenance runs are a robot-server concept that

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

@@ -715,6 +715,26 @@ def get_all_commands_final(self) -> bool:
 
         return no_command_running and no_command_to_execute
 
+    def get_command_recovery_complete(self, command_id: str) -> bool:
+        """Return whether we're finished with error recovery for the given command.
+
+        If the given command didn't have a recovery phase (because it hasn't completed
+        yet, or because it succeeded without an error, or because it failed with an
+        error that wasn't recoverable), that counts as `True`.
+
+        If the given command did have a recovery phase, but it was interrupted by a
+        stop, that also counts as `True`.
+        """
+        command_is_most_recent_to_fail = (
+            self._state.failed_command is not None
+            and command_id == self._state.failed_command.command.id
+        )
+        if command_is_most_recent_to_fail:
+            recovery_is_ongoing = self.get_status() == EngineStatus.AWAITING_RECOVERY
+            return not recovery_is_ongoing
+        else:
+            return True
+
     def raise_fatal_command_error(self) -> None:
         """Raise the run's fatal command error, if there was one, as an exception.
 

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

@@ -3,7 +3,8 @@
 
 from dataclasses import dataclass
 from functools import partial
-from typing import Any, Callable, Dict, List, Optional, Sequence, TypeVar
+from typing import Callable, Dict, List, Optional, Sequence, TypeVar
+from typing_extensions import ParamSpec
 
 from opentrons_shared_data.deck.dev_types import DeckDefinitionV4
 
@@ -30,7 +31,9 @@
 from .state_summary import StateSummary
 from ..types import DeckConfigurationType
 
-ReturnT = TypeVar("ReturnT")
+
+_ParamsT = ParamSpec("_ParamsT")
+_ReturnT = TypeVar("_ReturnT")
 
 
 @dataclass(frozen=True)
@@ -207,10 +210,10 @@ def handle_action(self, action: Action) -> None:
 
     async def wait_for(
         self,
-        condition: Callable[..., Optional[ReturnT]],
-        *args: Any,
-        **kwargs: Any,
-    ) -> ReturnT:
+        condition: Callable[_ParamsT, Optional[_ReturnT]],
+        *args: _ParamsT.args,
+        **kwargs: _ParamsT.kwargs,
+    ) -> _ReturnT:
         """Wait for a condition to become true, checking whenever state changes.
 
         If the condition is already true, return immediately.