fix(api): fix run cancellation freezing on legacy core (#14312)

* communicate with smoothie via cancellable tasks during drop_tip movement and home_plunger

api/src/opentrons/hardware_control/api.py

@@ -619,6 +619,7 @@ async def _do_plunger_home(
                     home_flagged_axes=False,
                 )
 
+    @ExecutionManagerProvider.wait_for_running
     async def home_plunger(self, mount: top_types.Mount) -> None:
         """
         Home the plunger motor for a mount, and then return it to the 'bottom'
@@ -909,19 +910,18 @@ def engaged_axes(self) -> Dict[Axis, bool]:
     async def disengage_axes(self, which: List[Axis]) -> None:
         await self._backend.disengage_axes([ot2_axis_to_string(ax) for ax in which])
 
+    @ExecutionManagerProvider.wait_for_running
     async def _fast_home(self, axes: Sequence[str], margin: float) -> Dict[str, float]:
         converted_axes = "".join(axes)
         return await self._backend.fast_home(converted_axes, margin)
 
-    @ExecutionManagerProvider.wait_for_running
     async def retract(self, mount: top_types.Mount, margin: float = 10) -> None:
         """Pull the specified mount up to its home position.
 
         Works regardless of critical point or home status.
         """
         await self.retract_axis(Axis.by_mount(mount), margin)
 
-    @ExecutionManagerProvider.wait_for_running
     async def retract_axis(self, axis: Axis, margin: float = 10) -> None:
         """Pull the specified axis up to its home position.
 
@@ -1207,9 +1207,9 @@ async def drop_tip(self, mount: top_types.Mount, home_after: bool = True) -> Non
                 home_flagged_axes=False,
             )
             if move.home_after:
-                smoothie_pos = await self._backend.fast_home(
-                    [ot2_axis_to_string(ax) for ax in move.home_axes],
-                    move.home_after_safety_margin,
+                smoothie_pos = await self._fast_home(
+                    axes=[ot2_axis_to_string(ax) for ax in move.home_axes],
+                    margin=move.home_after_safety_margin,
                 )
                 self._current_position = deck_from_machine(
                     machine_pos=self._axis_map_from_string_map(smoothie_pos),

api/src/opentrons/hardware_control/execution_manager.py

@@ -1,6 +1,16 @@
 import asyncio
 import functools
-from typing import Set, TypeVar, Type, cast, Callable, Any, overload, Coroutine
+from typing import (
+    Set,
+    TypeVar,
+    Type,
+    cast,
+    Callable,
+    Any,
+    Coroutine,
+    ParamSpec,
+    Concatenate,
+)
 from .types import ExecutionState
 from opentrons_shared_data.errors.exceptions import ExecutionCancelledError
 
@@ -72,15 +82,9 @@ async def wait_for_is_running(self) -> None:
                 pass
 
 
-DecoratedReturn = TypeVar("DecoratedReturn")
-DecoratedMethodReturningValue = TypeVar(
-    "DecoratedMethodReturningValue",
-    bound=Callable[..., Coroutine[None, None, DecoratedReturn]],
-)
-DecoratedMethodNoReturn = TypeVar(
-    "DecoratedMethodNoReturn", bound=Callable[..., Coroutine[None, None, None]]
-)
 SubclassInstance = TypeVar("SubclassInstance", bound="ExecutionManagerProvider")
+DecoratedMethodParams = ParamSpec("DecoratedMethodParams")
+DecoratedReturn = TypeVar("DecoratedReturn")
 
 
 class ExecutionManagerProvider:
@@ -107,31 +111,22 @@ def taskify_movement_execution(self, cancellable: bool) -> None:
     def execution_manager(self) -> ExecutionManager:
         return self._execution_manager
 
-    @overload
     @classmethod
     def wait_for_running(
-        cls: Type[SubclassInstance], decorated: DecoratedMethodReturningValue
-    ) -> DecoratedMethodReturningValue:
-        ...
-
-    @overload
-    @classmethod
-    def wait_for_running(
-        cls: Type[SubclassInstance], decorated: DecoratedMethodNoReturn
-    ) -> DecoratedMethodNoReturn:
-        ...
-
-    # this type ignore and the overloads are because mypy requires that a function
-    # whose signature declares it returns None not have a return statement, whereas
-    # this function's implementation relies on python having None actually be the
-    # thing you return, and it's mad at that
-    @classmethod  # type: ignore
-    def wait_for_running(
-        cls: Type[SubclassInstance], decorated: DecoratedMethodReturningValue
-    ) -> DecoratedMethodReturningValue:
+        cls: Type["ExecutionManagerProvider"],
+        decorated: Callable[
+            Concatenate[SubclassInstance, DecoratedMethodParams],
+            Coroutine[Any, Any, DecoratedReturn],
+        ],
+    ) -> Callable[
+        Concatenate[SubclassInstance, DecoratedMethodParams],
+        Coroutine[Any, Any, DecoratedReturn],
+    ]:
         @functools.wraps(decorated)
         async def replace(
-            inst: SubclassInstance, *args: Any, **kwargs: Any
+            inst: SubclassInstance,
+            *args: DecoratedMethodParams.args,
+            **kwargs: DecoratedMethodParams.kwargs,
         ) -> DecoratedReturn:
             if not inst._em_simulate:
                 await inst.execution_manager.wait_for_is_running()
@@ -148,7 +143,13 @@ async def replace(
             else:
                 return await decorated(inst, *args, **kwargs)
 
-        return cast(DecoratedMethodReturningValue, replace)
+        return cast(
+            Callable[
+                Concatenate[SubclassInstance, DecoratedMethodParams],
+                Coroutine[Any, Any, DecoratedReturn],
+            ],
+            replace,
+        )
 
     async def do_delay(self, duration_s: float) -> None:
         if not self._em_simulate: