diff --git a/api/src/opentrons/protocol_engine/actions/__init__.py b/api/src/opentrons/protocol_engine/actions/__init__.py
index dfd497817c0..26dfb0df8e0 100644
--- a/api/src/opentrons/protocol_engine/actions/__init__.py
+++ b/api/src/opentrons/protocol_engine/actions/__init__.py
@@ -30,7 +30,7 @@
     SetPipetteMovementSpeedAction,
     AddAbsorbanceReaderLidAction,
 )
-from .get_state_update import get_state_update
+from .get_state_update import get_state_updates
 
 __all__ = [
     # action pipeline interface
@@ -63,5 +63,5 @@
     "PauseSource",
     "FinishErrorDetails",
     # helper functions
-    "get_state_update",
+    "get_state_updates",
 ]
diff --git a/api/src/opentrons/protocol_engine/actions/actions.py b/api/src/opentrons/protocol_engine/actions/actions.py
index 4569f7866ef..4cdcb771616 100644
--- a/api/src/opentrons/protocol_engine/actions/actions.py
+++ b/api/src/opentrons/protocol_engine/actions/actions.py
@@ -69,7 +69,7 @@ class StopAction:
 class ResumeFromRecoveryAction:
     """See `ProtocolEngine.resume_from_recovery()`."""
 
-    pass
+    state_update: StateUpdate
 
 
 @dataclasses.dataclass(frozen=True)
diff --git a/api/src/opentrons/protocol_engine/actions/get_state_update.py b/api/src/opentrons/protocol_engine/actions/get_state_update.py
index e0ddadc3222..3ab984ab850 100644
--- a/api/src/opentrons/protocol_engine/actions/get_state_update.py
+++ b/api/src/opentrons/protocol_engine/actions/get_state_update.py
@@ -1,18 +1,35 @@
 # noqa: D100
 
 
-from .actions import Action, SucceedCommandAction, FailCommandAction
+from .actions import (
+    Action,
+    ResumeFromRecoveryAction,
+    SucceedCommandAction,
+    FailCommandAction,
+)
 from ..commands.command import DefinedErrorData
+from ..error_recovery_policy import ErrorRecoveryType
 from ..state.update_types import StateUpdate
 
 
-def get_state_update(action: Action) -> StateUpdate | None:
-    """Extract the StateUpdate from an action, if there is one."""
+def get_state_updates(action: Action) -> list[StateUpdate]:
+    """Extract all the StateUpdates that the StateStores should apply when they apply an action."""
     if isinstance(action, SucceedCommandAction):
-        return action.state_update
+        return [action.state_update]
+
     elif isinstance(action, FailCommandAction) and isinstance(
         action.error, DefinedErrorData
     ):
-        return action.error.state_update
+        if action.type == ErrorRecoveryType.ASSUME_FALSE_POSITIVE_AND_CONTINUE:
+            return [
+                action.error.state_update,
+                action.error.state_update_if_false_positive,
+            ]
+        else:
+            return [action.error.state_update]
+
+    elif isinstance(action, ResumeFromRecoveryAction):
+        return [action.state_update]
+
     else:
-        return None
+        return []
diff --git a/api/src/opentrons/protocol_engine/commands/command.py b/api/src/opentrons/protocol_engine/commands/command.py
index 9ba9404af1f..813a038d7ec 100644
--- a/api/src/opentrons/protocol_engine/commands/command.py
+++ b/api/src/opentrons/protocol_engine/commands/command.py
@@ -147,6 +147,10 @@ class DefinedErrorData(Generic[_ErrorT_co]):
     )
     """How the engine state should be updated to reflect this command failure."""
 
+    state_update_if_false_positive: StateUpdate = dataclasses.field(
+        default_factory=StateUpdate
+    )
+
 
 class BaseCommand(
     GenericModel,
diff --git a/api/src/opentrons/protocol_engine/commands/drop_tip.py b/api/src/opentrons/protocol_engine/commands/drop_tip.py
index f4917a82195..114a97b0467 100644
--- a/api/src/opentrons/protocol_engine/commands/drop_tip.py
+++ b/api/src/opentrons/protocol_engine/commands/drop_tip.py
@@ -146,7 +146,15 @@ async def execute(self, params: DropTipParams) -> _ExecuteReturn:
                     )
                 ],
             )
-            return DefinedErrorData(public=error, state_update=state_update)
+            state_update_if_false_positive = update_types.StateUpdate()
+            state_update_if_false_positive.update_pipette_tip_state(
+                pipette_id=params.pipetteId, tip_geometry=None
+            )
+            return DefinedErrorData(
+                public=error,
+                state_update=state_update,
+                state_update_if_false_positive=state_update_if_false_positive,
+            )
         else:
             state_update.update_pipette_tip_state(
                 pipette_id=params.pipetteId, tip_geometry=None
diff --git a/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py b/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py
index 81b47e05c08..49d44d6b563 100644
--- a/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py
+++ b/api/src/opentrons/protocol_engine/commands/drop_tip_in_place.py
@@ -72,6 +72,10 @@ async def execute(self, params: DropTipInPlaceParams) -> _ExecuteReturn:
                 pipette_id=params.pipetteId, home_after=params.homeAfter
             )
         except TipAttachedError as exception:
+            state_update_if_false_positive = update_types.StateUpdate()
+            state_update_if_false_positive.update_pipette_tip_state(
+                pipette_id=params.pipetteId, tip_geometry=None
+            )
             error = TipPhysicallyAttachedError(
                 id=self._model_utils.generate_id(),
                 createdAt=self._model_utils.get_timestamp(),
@@ -83,7 +87,11 @@ async def execute(self, params: DropTipInPlaceParams) -> _ExecuteReturn:
                     )
                 ],
             )
-            return DefinedErrorData(public=error, state_update=state_update)
+            return DefinedErrorData(
+                public=error,
+                state_update=state_update,
+                state_update_if_false_positive=state_update_if_false_positive,
+            )
         else:
             state_update.update_pipette_tip_state(
                 pipette_id=params.pipetteId, tip_geometry=None
diff --git a/api/src/opentrons/protocol_engine/commands/pick_up_tip.py b/api/src/opentrons/protocol_engine/commands/pick_up_tip.py
index 5ccdcfc6f3a..bf8492cc74b 100644
--- a/api/src/opentrons/protocol_engine/commands/pick_up_tip.py
+++ b/api/src/opentrons/protocol_engine/commands/pick_up_tip.py
@@ -6,7 +6,7 @@
 from typing_extensions import Literal
 
 
-from ..errors import ErrorOccurrence, TipNotAttachedError
+from ..errors import ErrorOccurrence, PickUpTipTipNotAttachedError
 from ..resources import ModelUtils
 from ..state import update_types
 from ..types import PickUpTipWellLocation, DeckPoint
@@ -140,7 +140,12 @@ async def execute(
                 labware_id=labware_id,
                 well_name=well_name,
             )
-        except TipNotAttachedError as e:
+        except PickUpTipTipNotAttachedError as e:
+            state_update_if_false_positive = update_types.StateUpdate()
+            state_update_if_false_positive.update_pipette_tip_state(
+                pipette_id=pipette_id,
+                tip_geometry=e.tip_geometry,
+            )
             state_update.mark_tips_as_used(
                 pipette_id=pipette_id, labware_id=labware_id, well_name=well_name
             )
@@ -157,6 +162,7 @@ async def execute(
                     ],
                 ),
                 state_update=state_update,
+                state_update_if_false_positive=state_update_if_false_positive,
             )
         else:
             state_update.update_pipette_tip_state(
diff --git a/api/src/opentrons/protocol_engine/create_protocol_engine.py b/api/src/opentrons/protocol_engine/create_protocol_engine.py
index dc66591eff2..372972c1f50 100644
--- a/api/src/opentrons/protocol_engine/create_protocol_engine.py
+++ b/api/src/opentrons/protocol_engine/create_protocol_engine.py
@@ -5,12 +5,20 @@
 
 from opentrons.hardware_control import HardwareControlAPI
 from opentrons.hardware_control.types import DoorState
-from opentrons.protocol_engine.error_recovery_policy import ErrorRecoveryPolicy
+from opentrons.protocol_engine.execution.error_recovery_hardware_state_synchronizer import (
+    ErrorRecoveryHardwareStateSynchronizer,
+)
 from opentrons.util.async_helpers import async_context_manager_in_thread
+
 from opentrons_shared_data.robot import load as load_robot
 
+from .actions.action_dispatcher import ActionDispatcher
+from .error_recovery_policy import ErrorRecoveryPolicy
+from .execution.door_watcher import DoorWatcher
+from .execution.hardware_stopper import HardwareStopper
+from .plugins import PluginStarter
 from .protocol_engine import ProtocolEngine
-from .resources import DeckDataProvider, ModuleDataProvider, FileProvider
+from .resources import DeckDataProvider, ModuleDataProvider, FileProvider, ModelUtils
 from .state.config import Config
 from .state.state import StateStore
 from .types import PostRunHardwareState, DeckConfigurationType
@@ -61,10 +69,27 @@ async def create_protocol_engine(
         deck_configuration=deck_configuration,
         notify_publishers=notify_publishers,
     )
+    hardware_state_synchronizer = ErrorRecoveryHardwareStateSynchronizer(
+        hardware_api, state_store
+    )
+    action_dispatcher = ActionDispatcher(state_store)
+    action_dispatcher.add_handler(hardware_state_synchronizer)
+    plugin_starter = PluginStarter(state_store, action_dispatcher)
+    model_utils = ModelUtils()
+    hardware_stopper = HardwareStopper(hardware_api, state_store)
+    door_watcher = DoorWatcher(state_store, hardware_api, action_dispatcher)
+    module_data_provider = ModuleDataProvider()
+    file_provider = file_provider or FileProvider()
 
     return ProtocolEngine(
-        state_store=state_store,
         hardware_api=hardware_api,
+        state_store=state_store,
+        action_dispatcher=action_dispatcher,
+        plugin_starter=plugin_starter,
+        model_utils=model_utils,
+        hardware_stopper=hardware_stopper,
+        door_watcher=door_watcher,
+        module_data_provider=module_data_provider,
         file_provider=file_provider,
     )
 
diff --git a/api/src/opentrons/protocol_engine/error_recovery_policy.py b/api/src/opentrons/protocol_engine/error_recovery_policy.py
index d959651393e..fcc8a2ffef5 100644
--- a/api/src/opentrons/protocol_engine/error_recovery_policy.py
+++ b/api/src/opentrons/protocol_engine/error_recovery_policy.py
@@ -26,10 +26,20 @@ class ErrorRecoveryType(enum.Enum):
     """
 
     WAIT_FOR_RECOVERY = enum.auto()
-    """Stop and wait for the error to be recovered from manually."""
+    """Enter interactive error recovery mode."""
 
-    IGNORE_AND_CONTINUE = enum.auto()
-    """Continue with the run, as if the command never failed."""
+    CONTINUE_WITH_ERROR = enum.auto()
+    """Continue without interruption, carrying on from whatever error state the failed
+    command left the engine in.
+
+    This is like `ProtocolEngine.resume_from_recovery(reconcile_false_positive=False)`.
+    """
+
+    ASSUME_FALSE_POSITIVE_AND_CONTINUE = enum.auto()
+    """Continue without interruption, acting as if the underlying error was a false positive.
+
+    This is like `ProtocolEngine.resume_from_recovery(reconcile_false_positive=True)`.
+    """
 
 
 class ErrorRecoveryPolicy(Protocol):
diff --git a/api/src/opentrons/protocol_engine/errors/__init__.py b/api/src/opentrons/protocol_engine/errors/__init__.py
index 9bbe3aae9b8..b25dfdb2d0e 100644
--- a/api/src/opentrons/protocol_engine/errors/__init__.py
+++ b/api/src/opentrons/protocol_engine/errors/__init__.py
@@ -8,6 +8,7 @@
     InvalidSpecificationForRobotTypeError,
     InvalidLoadPipetteSpecsError,
     TipNotAttachedError,
+    PickUpTipTipNotAttachedError,
     TipAttachedError,
     CommandDoesNotExistError,
     LabwareNotLoadedError,
@@ -89,6 +90,7 @@
     "InvalidSpecificationForRobotTypeError",
     "InvalidLoadPipetteSpecsError",
     "TipNotAttachedError",
+    "PickUpTipTipNotAttachedError",
     "TipAttachedError",
     "CommandDoesNotExistError",
     "LabwareNotLoadedError",
diff --git a/api/src/opentrons/protocol_engine/errors/exceptions.py b/api/src/opentrons/protocol_engine/errors/exceptions.py
index 5656942b338..12f45f4936d 100644
--- a/api/src/opentrons/protocol_engine/errors/exceptions.py
+++ b/api/src/opentrons/protocol_engine/errors/exceptions.py
@@ -1,11 +1,17 @@
 """Protocol engine exceptions."""
 
+from __future__ import annotations
+
 from logging import getLogger
-from typing import Any, Dict, Optional, Union, Iterator, Sequence
+from typing import Any, Dict, Final, Optional, Union, Iterator, Sequence, TYPE_CHECKING
 
 from opentrons_shared_data.errors import ErrorCodes
 from opentrons_shared_data.errors.exceptions import EnumeratedError, PythonException
 
+if TYPE_CHECKING:
+    from opentrons.protocol_engine.types import TipGeometry
+
+
 log = getLogger(__name__)
 
 
@@ -132,6 +138,21 @@ def __init__(
         super().__init__(ErrorCodes.UNEXPECTED_TIP_REMOVAL, message, details, wrapping)
 
 
+class PickUpTipTipNotAttachedError(TipNotAttachedError):
+    """Raised from TipHandler.pick_up_tip().
+
+    This is like TipNotAttachedError except that it carries some extra information
+    about the attempted operation.
+    """
+
+    tip_geometry: Final[TipGeometry]
+    """The tip geometry that would have been on the pipette, had the operation succeeded."""
+
+    def __init__(self, tip_geometry: TipGeometry) -> None:
+        super().__init__()
+        self.tip_geometry = tip_geometry
+
+
 class TipAttachedError(ProtocolEngineError):
     """Raised when a tip shouldn't be attached, but is."""
 
diff --git a/api/src/opentrons/protocol_engine/execution/error_recovery_hardware_state_synchronizer.py b/api/src/opentrons/protocol_engine/execution/error_recovery_hardware_state_synchronizer.py
new file mode 100644
index 00000000000..67d75cfb181
--- /dev/null
+++ b/api/src/opentrons/protocol_engine/execution/error_recovery_hardware_state_synchronizer.py
@@ -0,0 +1,101 @@
+# noqa: D100
+
+
+from opentrons.hardware_control import HardwareControlAPI
+from opentrons.protocol_engine.actions.action_handler import ActionHandler
+from opentrons.protocol_engine.actions.actions import (
+    Action,
+    FailCommandAction,
+    ResumeFromRecoveryAction,
+)
+from opentrons.protocol_engine.commands.command import DefinedErrorData
+from opentrons.protocol_engine.error_recovery_policy import ErrorRecoveryType
+from opentrons.protocol_engine.execution.tip_handler import HardwareTipHandler
+from opentrons.protocol_engine.state import update_types
+from opentrons.protocol_engine.state.state import StateView
+
+
+class ErrorRecoveryHardwareStateSynchronizer(ActionHandler):
+    """A hack to keep the hardware API's state correct through certain error recovery flows.
+
+    BACKGROUND:
+
+    Certain parts of robot state are duplicated between `opentrons.protocol_engine` and
+    `opentrons.hardware_control`. Stuff like "is there a tip attached."
+
+    Normally, Protocol Engine command implementations (`opentrons.protocol_engine.commands`)
+    mutate hardware API state when they execute; and then when they finish executing,
+    the Protocol Engine state stores (`opentrons.protocol_engine.state`) update Protocol
+    Engine state accordingly. So both halves are accounted for. This generally works fine.
+
+    However, we need to go out of our way to support
+    `ProtocolEngine.resume_from_recovery(reconcile_false_positive=True)`.
+    It wants to apply a second set of state updates to "fix things up" with the
+    new knowledge that some error was a false positive. The Protocol Engine half of that
+    is easy for us to apply the normal way, through the state stores; but the
+    hardware API half of that cannot be applied the normal way, from the command
+    implementation, because the command in question is no longer running.
+
+    THE HACK:
+
+    This listens for the same error recovery state updates that the state stores do,
+    figures out what hardware API state mutations ought to go along with them,
+    and then does those mutations.
+
+    The problem is that hardware API state is now mutated from two different places
+    (sometimes the command implementations, and sometimes here), which are bound
+    to grow accidental differences.
+
+    TO FIX:
+
+    Make Protocol Engine's use of the hardware API less stateful. e.g. supply
+    tip geometry every time we call a hardware API movement method, instead of
+    just once when we pick up a tip. Use Protocol Engine state as the single source
+    of truth.
+    """
+
+    def __init__(self, hardware_api: HardwareControlAPI, state_view: StateView) -> None:
+        self._hardware_api = hardware_api
+        self._state_view = state_view
+
+    def handle_action(self, action: Action) -> None:
+        """Modify hardware API state in reaction to a Protocol Engine action."""
+        state_update = _get_state_update(action)
+        if state_update:
+            self._synchronize(state_update)
+
+    def _synchronize(self, state_update: update_types.StateUpdate) -> None:
+        tip_handler = HardwareTipHandler(self._state_view, self._hardware_api)
+
+        if state_update.pipette_tip_state != update_types.NO_CHANGE:
+            pipette_id = state_update.pipette_tip_state.pipette_id
+            tip_geometry = state_update.pipette_tip_state.tip_geometry
+            if tip_geometry is None:
+                tip_handler.remove_tip(pipette_id)
+            else:
+                tip_handler.cache_tip(pipette_id=pipette_id, tip=tip_geometry)
+
+
+def _get_state_update(action: Action) -> update_types.StateUpdate | None:
+    """Get the mutations that we need to do on the hardware API to stay in sync with an engine action.
+
+    The mutations are returned in Protocol Engine terms, as a StateUpdate.
+    They then need to be converted to hardware API terms.
+    """
+    match action:
+        case ResumeFromRecoveryAction(state_update=state_update):
+            return state_update
+
+        case FailCommandAction(
+            error=DefinedErrorData(
+                state_update_if_false_positive=state_update_if_false_positive
+            )
+        ):
+            return (
+                state_update_if_false_positive
+                if action.type == ErrorRecoveryType.ASSUME_FALSE_POSITIVE_AND_CONTINUE
+                else None
+            )
+
+        case _:
+            return None
diff --git a/api/src/opentrons/protocol_engine/execution/hardware_stopper.py b/api/src/opentrons/protocol_engine/execution/hardware_stopper.py
index 24055f6b03b..81d4f10d94d 100644
--- a/api/src/opentrons/protocol_engine/execution/hardware_stopper.py
+++ b/api/src/opentrons/protocol_engine/execution/hardware_stopper.py
@@ -78,9 +78,7 @@ async def _drop_tip(self) -> None:
                 try:
                     if self._state_store.labware.get_fixed_trash_id() == FIXED_TRASH_ID:
                         # OT-2 and Flex 2.15 protocols will default to the Fixed Trash Labware
-                        await self._tip_handler.cache_tip(
-                            pipette_id=pipette_id, tip=tip
-                        )
+                        self._tip_handler.cache_tip(pipette_id=pipette_id, tip=tip)
                         await self._movement_handler.move_to_well(
                             pipette_id=pipette_id,
                             labware_id=FIXED_TRASH_ID,
@@ -92,9 +90,7 @@ async def _drop_tip(self) -> None:
                         )
                     elif self._state_store.config.robot_type == "OT-2 Standard":
                         # API 2.16 and above OT2 protocols use addressable areas
-                        await self._tip_handler.cache_tip(
-                            pipette_id=pipette_id, tip=tip
-                        )
+                        self._tip_handler.cache_tip(pipette_id=pipette_id, tip=tip)
                         await self._movement_handler.move_to_addressable_area(
                             pipette_id=pipette_id,
                             addressable_area_name="fixedTrash",
diff --git a/api/src/opentrons/protocol_engine/execution/tip_handler.py b/api/src/opentrons/protocol_engine/execution/tip_handler.py
index a963dd9abac..dde67ece007 100644
--- a/api/src/opentrons/protocol_engine/execution/tip_handler.py
+++ b/api/src/opentrons/protocol_engine/execution/tip_handler.py
@@ -4,6 +4,9 @@
 
 from opentrons.hardware_control import HardwareControlAPI
 from opentrons.hardware_control.types import FailedTipStateCheck, InstrumentProbeType
+from opentrons.protocol_engine.errors.exceptions import PickUpTipTipNotAttachedError
+from opentrons.types import Mount
+
 from opentrons_shared_data.errors.exceptions import (
     CommandPreconditionViolated,
     CommandParameterLimitViolated,
@@ -70,7 +73,7 @@ async def pick_up_tip(
             Tip geometry of the picked up tip.
 
         Raises:
-            TipNotAttachedError
+            PickUpTipTipNotAttachedError
         """
         ...
 
@@ -83,9 +86,12 @@ async def drop_tip(self, pipette_id: str, home_after: Optional[bool]) -> None:
             TipAttachedError
         """
 
-    async def cache_tip(self, pipette_id: str, tip: TipGeometry) -> None:
+    def cache_tip(self, pipette_id: str, tip: TipGeometry) -> None:
         """Tell the Hardware API that a tip is attached."""
 
+    def remove_tip(self, pipette_id: str) -> None:
+        """Tell the hardware API that no tip is attached."""
+
     async def get_tip_presence(self, pipette_id: str) -> TipPresenceStatus:
         """Get tip presence status on the pipette."""
 
@@ -198,6 +204,11 @@ def __init__(
         self._labware_data_provider = labware_data_provider or LabwareDataProvider()
         self._state_view = state_view
 
+        # WARNING: ErrorRecoveryHardwareStateSynchronizer can currently construct several
+        # instances of this class per run, in addition to the main instance used
+        # for command execution. We're therefore depending on this class being
+        # stateless, so consider that before adding additional attributes here.
+
     async def available_for_nozzle_layout(
         self,
         pipette_id: str,
@@ -223,7 +234,7 @@ async def pick_up_tip(
         well_name: str,
     ) -> TipGeometry:
         """See documentation on abstract base class."""
-        hw_mount = self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
+        hw_mount = self._get_hw_mount(pipette_id)
 
         nominal_tip_geometry = self._state_view.geometry.get_nominal_tip_geometry(
             pipette_id=pipette_id, labware_id=labware_id, well_name=well_name
@@ -234,6 +245,7 @@ async def pick_up_tip(
             labware_definition=self._state_view.labware.get_definition(labware_id),
             nominal_fallback=nominal_tip_geometry.length,
         )
+
         tip_geometry = TipGeometry(
             length=actual_tip_length,
             diameter=nominal_tip_geometry.diameter,
@@ -243,10 +255,12 @@ async def pick_up_tip(
         await self._hardware_api.tip_pickup_moves(
             mount=hw_mount, presses=None, increment=None
         )
-        # Allow TipNotAttachedError to propagate.
-        await self.verify_tip_presence(pipette_id, TipPresenceStatus.PRESENT)
+        try:
+            await self.verify_tip_presence(pipette_id, TipPresenceStatus.PRESENT)
+        except TipNotAttachedError as e:
+            raise PickUpTipTipNotAttachedError(tip_geometry=tip_geometry) from e
 
-        await self.cache_tip(pipette_id, tip_geometry)
+        self.cache_tip(pipette_id, tip_geometry)
 
         await self._hardware_api.prepare_for_aspirate(hw_mount)
 
@@ -254,7 +268,7 @@ async def pick_up_tip(
 
     async def drop_tip(self, pipette_id: str, home_after: Optional[bool]) -> None:
         """See documentation on abstract base class."""
-        hw_mount = self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
+        hw_mount = self._get_hw_mount(pipette_id)
 
         # Let the hardware controller handle defaulting home_after since its behavior
         # differs between machines
@@ -268,12 +282,11 @@ async def drop_tip(self, pipette_id: str, home_after: Optional[bool]) -> None:
         # Allow TipNotAttachedError to propagate.
         await self.verify_tip_presence(pipette_id, TipPresenceStatus.ABSENT)
 
-        self._hardware_api.remove_tip(hw_mount)
-        self._hardware_api.set_current_tiprack_diameter(hw_mount, 0)
+        self.remove_tip(pipette_id)
 
-    async def cache_tip(self, pipette_id: str, tip: TipGeometry) -> None:
+    def cache_tip(self, pipette_id: str, tip: TipGeometry) -> None:
         """See documentation on abstract base class."""
-        hw_mount = self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
+        hw_mount = self._get_hw_mount(pipette_id)
 
         self._hardware_api.cache_tip(mount=hw_mount, tip_length=tip.length)
 
@@ -287,12 +300,18 @@ async def cache_tip(self, pipette_id: str, tip: TipGeometry) -> None:
             tip_volume=tip.volume,
         )
 
+    def remove_tip(self, pipette_id: str) -> None:
+        """See documentation on abstract base class."""
+        hw_mount = self._get_hw_mount(pipette_id)
+        self._hardware_api.remove_tip(hw_mount)
+        self._hardware_api.set_current_tiprack_diameter(hw_mount, 0)
+
     async def get_tip_presence(self, pipette_id: str) -> TipPresenceStatus:
         """See documentation on abstract base class."""
         try:
             ot3api = ensure_ot3_hardware(hardware_api=self._hardware_api)
 
-            hw_mount = self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
+            hw_mount = self._get_hw_mount(pipette_id)
 
             status = await ot3api.get_tip_presence_status(hw_mount)
             return TipPresenceStatus.from_hw_state(status)
@@ -333,7 +352,7 @@ async def verify_tip_presence(
             return
         try:
             ot3api = ensure_ot3_hardware(hardware_api=self._hardware_api)
-            hw_mount = self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
+            hw_mount = self._get_hw_mount(pipette_id)
             await ot3api.verify_tip_presence(
                 hw_mount, expected.to_hw_state(), follow_singular_sensor
             )
@@ -351,6 +370,9 @@ async def verify_tip_presence(
                     wrapping=[PythonException(e)],
                 )
 
+    def _get_hw_mount(self, pipette_id: str) -> Mount:
+        return self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
+
 
 class VirtualTipHandler(TipHandler):
     """Pick up and drop tips, using a virtual pipette."""
@@ -414,13 +436,20 @@ async def drop_tip(
             expected_has_tip=True,
         )
 
-    async def cache_tip(self, pipette_id: str, tip: TipGeometry) -> None:
-        """Add a tip using a virtual pipette.
+    def cache_tip(self, pipette_id: str, tip: TipGeometry) -> None:
+        """See documentation on abstract base class.
 
         This should not be called when using virtual pipettes.
         """
         assert False, "TipHandler.cache_tip should not be used with virtual pipettes"
 
+    def remove_tip(self, pipette_id: str) -> None:
+        """See documentation on abstract base class.
+
+        This should not be called when using virtual pipettes.
+        """
+        assert False, "TipHandler.remove_tip should not be used with virtual pipettes"
+
     async def verify_tip_presence(
         self,
         pipette_id: str,
diff --git a/api/src/opentrons/protocol_engine/protocol_engine.py b/api/src/opentrons/protocol_engine/protocol_engine.py
index d93ab5dd42d..ced32b20cc3 100644
--- a/api/src/opentrons/protocol_engine/protocol_engine.py
+++ b/api/src/opentrons/protocol_engine/protocol_engine.py
@@ -6,7 +6,6 @@
     ResumeFromRecoveryAction,
     SetErrorRecoveryPolicyAction,
 )
-from opentrons.protocol_engine.error_recovery_policy import ErrorRecoveryPolicy
 
 from opentrons.protocols.models import LabwareDefinition
 from opentrons.hardware_control import HardwareControlAPI
@@ -19,6 +18,7 @@
 
 from .errors import ProtocolCommandFailedError, ErrorOccurrence, CommandNotAllowedError
 from .errors.exceptions import EStopActivatedError
+from .error_recovery_policy import ErrorRecoveryPolicy
 from . import commands, slot_standardization
 from .resources import ModelUtils, ModuleDataProvider, FileProvider
 from .types import (
@@ -39,6 +39,7 @@
     HardwareStopper,
 )
 from .state.state import StateStore, StateView
+from .state.update_types import StateUpdate
 from .plugins import AbstractPlugin, PluginStarter
 from .actions import (
     ActionDispatcher,
@@ -88,43 +89,31 @@ def __init__(
         self,
         hardware_api: HardwareControlAPI,
         state_store: StateStore,
-        action_dispatcher: Optional[ActionDispatcher] = None,
-        plugin_starter: Optional[PluginStarter] = None,
+        action_dispatcher: ActionDispatcher,
+        plugin_starter: PluginStarter,
+        model_utils: ModelUtils,
+        hardware_stopper: HardwareStopper,
+        door_watcher: DoorWatcher,
+        module_data_provider: ModuleDataProvider,
+        file_provider: FileProvider,
         queue_worker: Optional[QueueWorker] = None,
-        model_utils: Optional[ModelUtils] = None,
-        hardware_stopper: Optional[HardwareStopper] = None,
-        door_watcher: Optional[DoorWatcher] = None,
-        module_data_provider: Optional[ModuleDataProvider] = None,
-        file_provider: Optional[FileProvider] = None,
     ) -> None:
         """Initialize a ProtocolEngine instance.
 
         Must be called while an event loop is active.
 
-        This constructor does not inject provider implementations.
+        This constructor is only for `ProtocolEngine` unit tests.
         Prefer the `create_protocol_engine()` factory function.
         """
         self._hardware_api = hardware_api
-        self._file_provider = file_provider or FileProvider()
+        self._file_provider = file_provider
         self._state_store = state_store
-        self._model_utils = model_utils or ModelUtils()
-        self._action_dispatcher = action_dispatcher or ActionDispatcher(
-            sink=self._state_store
-        )
-        self._plugin_starter = plugin_starter or PluginStarter(
-            state=self._state_store,
-            action_dispatcher=self._action_dispatcher,
-        )
-        self._hardware_stopper = hardware_stopper or HardwareStopper(
-            hardware_api=hardware_api,
-            state_store=state_store,
-        )
-        self._door_watcher = door_watcher or DoorWatcher(
-            state_store=state_store,
-            hardware_api=hardware_api,
-            action_dispatcher=self._action_dispatcher,
-        )
-        self._module_data_provider = module_data_provider or ModuleDataProvider()
+        self._model_utils = model_utils
+        self._action_dispatcher = action_dispatcher
+        self._plugin_starter = plugin_starter
+        self._hardware_stopper = hardware_stopper
+        self._door_watcher = door_watcher
+        self._module_data_provider = module_data_provider
         self._queue_worker = queue_worker
         if self._queue_worker:
             self._queue_worker.start()
@@ -186,11 +175,35 @@ def request_pause(self) -> None:
         self._action_dispatcher.dispatch(action)
         self._hardware_api.pause(HardwarePauseType.PAUSE)
 
-    def resume_from_recovery(self) -> None:
-        """Resume normal protocol execution after the engine was `AWAITING_RECOVERY`."""
+    def resume_from_recovery(self, reconcile_false_positive: bool) -> None:
+        """Resume normal protocol execution after the engine was `AWAITING_RECOVERY`.
+
+        If `reconcile_false_positive` is `False`, the engine will continue naively from
+        whatever state the error left it in. (Each defined error individually documents
+        exactly how it affects state.) This is appropriate for client-driven error
+        recovery, where the client wants predictable behavior from the engine.
+
+        If `reconcile_false_positive` is `True`, the engine may apply additional fixups
+        to its state to try to get the rest of the run to just work, assuming the error
+        was a false-positive.
+
+        For example, a `tipPhysicallyMissing` error from a `pickUpTip` would normally
+        leave the engine state without a tip on the pipette. If `reconcile_false_positive=True`,
+        the engine will set the pipette to have that missing tip before continuing, so
+        subsequent path planning, aspirates, dispenses, etc. will work as if nothing
+        went wrong.
+        """
+        if reconcile_false_positive:
+            state_update = (
+                self._state_store.commands.get_state_update_for_false_positive()
+            )
+        else:
+            state_update = StateUpdate()  # Empty/no-op.
+
         action = self._state_store.commands.validate_action_allowed(
-            ResumeFromRecoveryAction()
+            ResumeFromRecoveryAction(state_update)
         )
+
         self._action_dispatcher.dispatch(action)
 
     def add_command(
diff --git a/api/src/opentrons/protocol_engine/state/commands.py b/api/src/opentrons/protocol_engine/state/commands.py
index 6723c521892..dc9e0c7ee49 100644
--- a/api/src/opentrons/protocol_engine/state/commands.py
+++ b/api/src/opentrons/protocol_engine/state/commands.py
@@ -25,6 +25,7 @@
     ErrorRecoveryType,
 )
 from opentrons.protocol_engine.notes.notes import CommandNote
+from opentrons.protocol_engine.state import update_types
 
 from ..actions import (
     Action,
@@ -141,6 +142,16 @@ class CommandPointer:
     index: int
 
 
+@dataclass(frozen=True)
+class _RecoveryTargetInfo:
+    """Info about the failed command that we're currently recovering from."""
+
+    command_id: str
+
+    state_update_if_false_positive: update_types.StateUpdate
+    """See `CommandView.get_state_update_if_continued()`."""
+
+
 @dataclass
 class CommandState:
     """State of all protocol engine command resources."""
@@ -205,8 +216,8 @@ class CommandState:
     stable. Eventually, we might want this info to be stored directly on each command.
     """
 
-    recovery_target_command_id: Optional[str]
-    """If we're currently recovering from a command failure, which command it was."""
+    recovery_target: Optional[_RecoveryTargetInfo]
+    """If we're currently recovering from a command failure, info about that command."""
 
     finish_error: Optional[ErrorOccurrence]
     """The error that happened during the post-run finish steps (homing & dropping tips), if any."""
@@ -253,7 +264,7 @@ def __init__(
             finish_error=None,
             failed_command=None,
             command_error_recovery_types={},
-            recovery_target_command_id=None,
+            recovery_target=None,
             run_completed_at=None,
             run_started_at=None,
             latest_protocol_command_hash=None,
@@ -335,14 +346,17 @@ def _handle_succeed_command_action(self, action: SucceedCommandAction) -> None:
     def _handle_fail_command_action(self, action: FailCommandAction) -> None:
         prev_entry = self.state.command_history.get(action.command_id)
 
-        if isinstance(action.error, EnumeratedError):
+        if isinstance(action.error, EnumeratedError):  # The error was undefined.
             public_error_occurrence = ErrorOccurrence.from_failed(
                 id=action.error_id,
                 createdAt=action.failed_at,
                 error=action.error,
             )
-        else:
+            # An empty state update, to no-op.
+            state_update_if_false_positive = update_types.StateUpdate()
+        else:  # The error was defined.
             public_error_occurrence = action.error.public
+            state_update_if_false_positive = action.error.state_update_if_false_positive
 
         self._update_to_failed(
             command_id=action.command_id,
@@ -354,6 +368,19 @@ def _handle_fail_command_action(self, action: FailCommandAction) -> None:
         self._state.failed_command = self._state.command_history.get(action.command_id)
         self._state.failed_command_errors.append(public_error_occurrence)
 
+        if (
+            prev_entry.command.intent in (CommandIntent.PROTOCOL, None)
+            and action.type == ErrorRecoveryType.WAIT_FOR_RECOVERY
+        ):
+            self._state.queue_status = QueueStatus.AWAITING_RECOVERY
+            self._state.recovery_target = _RecoveryTargetInfo(
+                command_id=action.command_id,
+                state_update_if_false_positive=state_update_if_false_positive,
+            )
+            self._state.has_entered_error_recovery = True
+
+        # When one command fails, we generally also cancel the commands that
+        # would have been queued after it.
         other_command_ids_to_fail: List[str]
         if prev_entry.command.intent == CommandIntent.SETUP:
             other_command_ids_to_fail = list(
@@ -373,7 +400,8 @@ def _handle_fail_command_action(self, action: FailCommandAction) -> None:
                 )
             elif (
                 action.type == ErrorRecoveryType.WAIT_FOR_RECOVERY
-                or action.type == ErrorRecoveryType.IGNORE_AND_CONTINUE
+                or action.type == ErrorRecoveryType.CONTINUE_WITH_ERROR
+                or action.type == ErrorRecoveryType.ASSUME_FALSE_POSITIVE_AND_CONTINUE
             ):
                 other_command_ids_to_fail = []
             else:
@@ -390,14 +418,6 @@ def _handle_fail_command_action(self, action: FailCommandAction) -> None:
                 notes=None,
             )
 
-        if (
-            prev_entry.command.intent in (CommandIntent.PROTOCOL, None)
-            and action.type == ErrorRecoveryType.WAIT_FOR_RECOVERY
-        ):
-            self._state.queue_status = QueueStatus.AWAITING_RECOVERY
-            self._state.recovery_target_command_id = action.command_id
-            self._state.has_entered_error_recovery = True
-
     def _handle_play_action(self, action: PlayAction) -> None:
         if not self._state.run_result:
             self._state.run_started_at = (
@@ -425,11 +445,11 @@ def _handle_resume_from_recovery_action(
         self, action: ResumeFromRecoveryAction
     ) -> None:
         self._state.queue_status = QueueStatus.RUNNING
-        self._state.recovery_target_command_id = None
+        self._state.recovery_target = None
 
     def _handle_stop_action(self, action: StopAction) -> None:
         if not self._state.run_result:
-            self._state.recovery_target_command_id = None
+            self._state.recovery_target = None
 
             self._state.queue_status = QueueStatus.PAUSED
             if action.from_estop:
@@ -866,11 +886,11 @@ def get_all_commands_final(self) -> bool:
 
     def get_recovery_target(self) -> Optional[CommandPointer]:
         """Return the command currently undergoing error recovery, if any."""
-        recovery_target_command_id = self._state.recovery_target_command_id
-        if recovery_target_command_id is None:
+        recovery_target = self._state.recovery_target
+        if recovery_target is None:
             return None
         else:
-            entry = self._state.command_history.get(recovery_target_command_id)
+            entry = self._state.command_history.get(recovery_target.command_id)
             return CommandPointer(
                 command_id=entry.command.id,
                 command_key=entry.command.key,
@@ -1083,6 +1103,19 @@ def get_error_recovery_policy(self) -> ErrorRecoveryPolicy:
         """
         return self._state.error_recovery_policy
 
+    def get_state_update_for_false_positive(self) -> update_types.StateUpdate:
+        """Return the state update for if the current recovery target was a false positive.
+
+        If we're currently in error recovery mode, and you have decided that the
+        underlying command error was a false positive, this returns a state update
+        that will undo the error's effects on engine state.
+        See `ProtocolEngine.resume_from_recovery(reconcile_false_positive=True)`.
+        """
+        if self._state.recovery_target is None:
+            return update_types.StateUpdate()  # Empty/no-op.
+        else:
+            return self._state.recovery_target.state_update_if_false_positive
+
     def _may_run_with_door_open(
         self, *, fixit_command: Command | CommandCreate
     ) -> bool:
diff --git a/api/src/opentrons/protocol_engine/state/labware.py b/api/src/opentrons/protocol_engine/state/labware.py
index dad9fe54dd0..7cea4f9765b 100644
--- a/api/src/opentrons/protocol_engine/state/labware.py
+++ b/api/src/opentrons/protocol_engine/state/labware.py
@@ -53,7 +53,7 @@
     Action,
     AddLabwareOffsetAction,
     AddLabwareDefinitionAction,
-    get_state_update,
+    get_state_updates,
 )
 from ._abstract_store import HasState, HandlesActions
 from ._move_types import EdgePathType
@@ -149,8 +149,7 @@ def __init__(
 
     def handle_action(self, action: Action) -> None:
         """Modify state in reaction to an action."""
-        state_update = get_state_update(action)
-        if state_update is not None:
+        for state_update in get_state_updates(action):
             self._add_loaded_labware(state_update)
             self._set_labware_location(state_update)
 
diff --git a/api/src/opentrons/protocol_engine/state/pipettes.py b/api/src/opentrons/protocol_engine/state/pipettes.py
index ced8b6076f7..bb90e067ec6 100644
--- a/api/src/opentrons/protocol_engine/state/pipettes.py
+++ b/api/src/opentrons/protocol_engine/state/pipettes.py
@@ -39,7 +39,7 @@
     FailCommandAction,
     SetPipetteMovementSpeedAction,
     SucceedCommandAction,
-    get_state_update,
+    get_state_updates,
 )
 from ._abstract_store import HasState, HandlesActions
 
@@ -141,8 +141,7 @@ def __init__(self) -> None:
 
     def handle_action(self, action: Action) -> None:
         """Modify state in reaction to an action."""
-        state_update = get_state_update(action)
-        if state_update is not None:
+        for state_update in get_state_updates(action):
             self._set_load_pipette(state_update)
             self._update_current_location(state_update)
             self._update_pipette_config(state_update)
diff --git a/api/src/opentrons/protocol_engine/state/tips.py b/api/src/opentrons/protocol_engine/state/tips.py
index f744b1a01b4..7427c78ac4c 100644
--- a/api/src/opentrons/protocol_engine/state/tips.py
+++ b/api/src/opentrons/protocol_engine/state/tips.py
@@ -6,7 +6,7 @@
 from opentrons.protocol_engine.state import update_types
 
 from ._abstract_store import HasState, HandlesActions
-from ..actions import Action, SucceedCommandAction, ResetTipsAction, get_state_update
+from ..actions import Action, SucceedCommandAction, ResetTipsAction, get_state_updates
 from ..commands import (
     Command,
     LoadLabwareResult,
@@ -63,8 +63,7 @@ def __init__(self) -> None:
 
     def handle_action(self, action: Action) -> None:
         """Modify state in reaction to an action."""
-        state_update = get_state_update(action)
-        if state_update is not None:
+        for state_update in get_state_updates(action):
             self._handle_state_update(state_update)
 
         if isinstance(action, SucceedCommandAction):
diff --git a/api/src/opentrons/protocol_runner/protocol_runner.py b/api/src/opentrons/protocol_runner/protocol_runner.py
index dcf4f224811..aec2aae80df 100644
--- a/api/src/opentrons/protocol_runner/protocol_runner.py
+++ b/api/src/opentrons/protocol_runner/protocol_runner.py
@@ -123,9 +123,9 @@ async def stop(self) -> None:
                 post_run_hardware_state=PostRunHardwareState.STAY_ENGAGED_IN_PLACE,
             )
 
-    def resume_from_recovery(self) -> None:
+    def resume_from_recovery(self, reconcile_false_positive: bool) -> None:
         """See `ProtocolEngine.resume_from_recovery()`."""
-        self._protocol_engine.resume_from_recovery()
+        self._protocol_engine.resume_from_recovery(reconcile_false_positive)
 
     @abstractmethod
     async def run(
diff --git a/api/src/opentrons/protocol_runner/run_orchestrator.py b/api/src/opentrons/protocol_runner/run_orchestrator.py
index 69d9feaf524..dfa66e6a55a 100644
--- a/api/src/opentrons/protocol_runner/run_orchestrator.py
+++ b/api/src/opentrons/protocol_runner/run_orchestrator.py
@@ -205,9 +205,9 @@ async def stop(self) -> None:
                 post_run_hardware_state=PostRunHardwareState.STAY_ENGAGED_IN_PLACE,
             )
 
-    def resume_from_recovery(self) -> None:
+    def resume_from_recovery(self, reconcile_false_positive: bool) -> None:
         """Resume the run from recovery."""
-        self._protocol_engine.resume_from_recovery()
+        self._protocol_engine.resume_from_recovery(reconcile_false_positive)
 
     async def finish(
         self,
diff --git a/api/tests/opentrons/protocol_engine/commands/test_drop_tip.py b/api/tests/opentrons/protocol_engine/commands/test_drop_tip.py
index 4a8e32c05d0..8ba2c9c1d97 100644
--- a/api/tests/opentrons/protocol_engine/commands/test_drop_tip.py
+++ b/api/tests/opentrons/protocol_engine/commands/test_drop_tip.py
@@ -294,4 +294,10 @@ async def test_tip_attached_error(
                 new_deck_point=DeckPoint(x=111, y=222, z=333),
             ),
         ),
+        state_update_if_false_positive=update_types.StateUpdate(
+            pipette_tip_state=update_types.PipetteTipStateUpdate(
+                pipette_id="abc",
+                tip_geometry=None,
+            )
+        ),
     )
diff --git a/api/tests/opentrons/protocol_engine/commands/test_drop_tip_in_place.py b/api/tests/opentrons/protocol_engine/commands/test_drop_tip_in_place.py
index f2061c3d552..fde82626969 100644
--- a/api/tests/opentrons/protocol_engine/commands/test_drop_tip_in_place.py
+++ b/api/tests/opentrons/protocol_engine/commands/test_drop_tip_in_place.py
@@ -91,4 +91,7 @@ async def test_tip_attached_error(
             wrappedErrors=[matchers.Anything()],
         ),
         state_update=StateUpdate(),
+        state_update_if_false_positive=StateUpdate(
+            pipette_tip_state=PipetteTipStateUpdate(pipette_id="abc", tip_geometry=None)
+        ),
     )
diff --git a/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py b/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py
index 3771fe00eb1..2203f514cf4 100644
--- a/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py
+++ b/api/tests/opentrons/protocol_engine/commands/test_pick_up_tip.py
@@ -2,6 +2,7 @@
 from datetime import datetime
 
 from decoy import Decoy, matchers
+from unittest.mock import sentinel
 
 from opentrons.types import MountType, Point
 
@@ -11,7 +12,7 @@
     WellOffset,
     DeckPoint,
 )
-from opentrons.protocol_engine.errors import TipNotAttachedError
+from opentrons.protocol_engine.errors import PickUpTipTipNotAttachedError
 from opentrons.protocol_engine.execution import MovementHandler, TipHandler
 from opentrons.protocol_engine.resources import ModelUtils
 from opentrons.protocol_engine.state import update_types
@@ -140,7 +141,7 @@ async def test_tip_physically_missing_error(
         await tip_handler.pick_up_tip(
             pipette_id=pipette_id, labware_id=labware_id, well_name=well_name
         )
-    ).then_raise(TipNotAttachedError())
+    ).then_raise(PickUpTipTipNotAttachedError(tip_geometry=sentinel.tip_geometry))
     decoy.when(model_utils.generate_id()).then_return(error_id)
     decoy.when(model_utils.get_timestamp()).then_return(error_created_at)
 
@@ -164,4 +165,9 @@ async def test_tip_physically_missing_error(
                 pipette_id="pipette-id", labware_id="labware-id", well_name="well-name"
             ),
         ),
+        state_update_if_false_positive=update_types.StateUpdate(
+            pipette_tip_state=update_types.PipetteTipStateUpdate(
+                pipette_id="pipette-id", tip_geometry=sentinel.tip_geometry
+            )
+        ),
     )
diff --git a/api/tests/opentrons/protocol_engine/execution/test_hardware_stopper.py b/api/tests/opentrons/protocol_engine/execution/test_hardware_stopper.py
index d6c69d0b170..503d681bced 100644
--- a/api/tests/opentrons/protocol_engine/execution/test_hardware_stopper.py
+++ b/api/tests/opentrons/protocol_engine/execution/test_hardware_stopper.py
@@ -158,7 +158,7 @@ async def test_hardware_stopping_sequence_no_tip_drop(
     decoy.verify(await hardware_api.stop(home_after=False), times=1)
 
     decoy.verify(
-        await mock_tip_handler.cache_tip(
+        mock_tip_handler.cache_tip(
             pipette_id="pipette-id",
             tip=TipGeometry(length=1.0, volume=2.0, diameter=3.0),
         ),
@@ -181,7 +181,7 @@ async def test_hardware_stopping_sequence_no_pipette(
     )
 
     decoy.when(
-        await mock_tip_handler.cache_tip(
+        mock_tip_handler.cache_tip(
             pipette_id="pipette-id",
             tip=TipGeometry(length=1.0, volume=2.0, diameter=3.0),
         ),
@@ -271,7 +271,7 @@ async def test_hardware_stopping_sequence_with_fixed_trash(
         await movement.home(
             axes=[MotorAxis.X, MotorAxis.Y, MotorAxis.LEFT_Z, MotorAxis.RIGHT_Z]
         ),
-        await mock_tip_handler.cache_tip(
+        mock_tip_handler.cache_tip(
             pipette_id="pipette-id",
             tip=TipGeometry(length=1.0, volume=2.0, diameter=3.0),
         ),
@@ -320,7 +320,7 @@ async def test_hardware_stopping_sequence_with_OT2_addressable_area(
         await movement.home(
             axes=[MotorAxis.X, MotorAxis.Y, MotorAxis.LEFT_Z, MotorAxis.RIGHT_Z]
         ),
-        await mock_tip_handler.cache_tip(
+        mock_tip_handler.cache_tip(
             pipette_id="pipette-id",
             tip=TipGeometry(length=1.0, volume=2.0, diameter=3.0),
         ),
diff --git a/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py b/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py
index 8ddb8840597..c03a611966c 100644
--- a/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py
+++ b/api/tests/opentrons/protocol_engine/execution/test_tip_handler.py
@@ -266,7 +266,7 @@ async def test_drop_tip(
     )
 
 
-async def test_add_tip(
+def test_add_tip(
     decoy: Decoy,
     mock_state_view: StateView,
     mock_hardware_api: HardwareAPI,
@@ -289,7 +289,7 @@ async def test_add_tip(
         MountType.LEFT
     )
 
-    await subject.cache_tip(pipette_id="pipette-id", tip=tip)
+    subject.cache_tip(pipette_id="pipette-id", tip=tip)
 
     decoy.verify(
         mock_hardware_api.cache_tip(mount=Mount.LEFT, tip_length=50),
@@ -301,6 +301,31 @@ async def test_add_tip(
     )
 
 
+def test_remove_tip(
+    decoy: Decoy,
+    mock_state_view: StateView,
+    mock_hardware_api: HardwareAPI,
+    mock_labware_data_provider: LabwareDataProvider,
+) -> None:
+    """It should remove a tip manually from the hardware API."""
+    subject = HardwareTipHandler(
+        state_view=mock_state_view,
+        hardware_api=mock_hardware_api,
+        labware_data_provider=mock_labware_data_provider,
+    )
+
+    decoy.when(mock_state_view.pipettes.get_mount("pipette-id")).then_return(
+        MountType.LEFT
+    )
+
+    subject.remove_tip(pipette_id="pipette-id")
+
+    decoy.verify(
+        mock_hardware_api.remove_tip(Mount.LEFT),
+        mock_hardware_api.set_current_tiprack_diameter(Mount.LEFT, 0),
+    )
+
+
 @pytest.mark.parametrize(
     argnames=[
         "test_channels",
diff --git a/api/tests/opentrons/protocol_engine/state/test_command_state.py b/api/tests/opentrons/protocol_engine/state/test_command_state.py
index 6f090612a74..9df52541f02 100644
--- a/api/tests/opentrons/protocol_engine/state/test_command_state.py
+++ b/api/tests/opentrons/protocol_engine/state/test_command_state.py
@@ -19,7 +19,7 @@
     PlayAction,
     SetErrorRecoveryPolicyAction,
 )
-from opentrons.protocol_engine.commands.command import CommandIntent
+from opentrons.protocol_engine.commands.command import CommandIntent, DefinedErrorData
 from opentrons.protocol_engine.error_recovery_policy import ErrorRecoveryType
 from opentrons.protocol_engine.errors.error_occurrence import ErrorOccurrence
 from opentrons.protocol_engine.errors.exceptions import (
@@ -32,6 +32,7 @@
     CommandView,
 )
 from opentrons.protocol_engine.state.config import Config
+from opentrons.protocol_engine.state.update_types import StateUpdate
 from opentrons.protocol_engine.types import DeckType, EngineStatus
 
 
@@ -772,7 +773,7 @@ def test_recovery_target_tracking() -> None:
     assert recovery_target.command_id == "c1"
     assert subject_view.get_recovery_in_progress_for_command("c1")
 
-    resume_from_1_recovery = actions.ResumeFromRecoveryAction()
+    resume_from_1_recovery = actions.ResumeFromRecoveryAction(StateUpdate())
     subject.handle_action(resume_from_1_recovery)
 
     # c1 failed recoverably, but we've already completed its recovery.
@@ -808,7 +809,7 @@ def test_recovery_target_tracking() -> None:
     # even though it failed recoverably before.
     assert not subject_view.get_recovery_in_progress_for_command("c1")
 
-    resume_from_2_recovery = actions.ResumeFromRecoveryAction()
+    resume_from_2_recovery = actions.ResumeFromRecoveryAction(StateUpdate())
     subject.handle_action(resume_from_2_recovery)
     queue_3 = actions.QueueCommandAction(
         "c3",
@@ -993,3 +994,57 @@ def test_set_and_get_error_recovery_policy() -> None:
     assert subject_view.get_error_recovery_policy() is initial_policy
     subject.handle_action(SetErrorRecoveryPolicyAction(sentinel.new_policy))
     assert subject_view.get_error_recovery_policy() is new_policy
+
+
+def test_get_state_update_for_false_positive() -> None:
+    """Test storage of false-positive state updates."""
+    subject = CommandStore(
+        config=_make_config(),
+        error_recovery_policy=_placeholder_error_recovery_policy,
+        is_door_open=False,
+    )
+    subject_view = CommandView(subject.state)
+
+    empty_state_update = StateUpdate()
+
+    assert subject_view.get_state_update_for_false_positive() == empty_state_update
+
+    queue = actions.QueueCommandAction(
+        request=commands.CommentCreate(
+            params=commands.CommentParams(message=""), key="command-key-1"
+        ),
+        request_hash=None,
+        created_at=datetime(year=2021, month=1, day=1),
+        command_id="command-id-1",
+    )
+    subject.handle_action(queue)
+    run = actions.RunCommandAction(
+        command_id="command-id-1",
+        started_at=datetime(year=2022, month=2, day=2),
+    )
+    subject.handle_action(run)
+    fail = actions.FailCommandAction(
+        command_id="command-id-1",
+        running_command=subject_view.get("command-id-1"),
+        error_id="error-id",
+        failed_at=datetime(year=2023, month=3, day=3),
+        error=DefinedErrorData(
+            public=sentinel.public,
+            state_update_if_false_positive=sentinel.state_update_if_false_positive,
+        ),
+        type=ErrorRecoveryType.WAIT_FOR_RECOVERY,
+        notes=[],
+    )
+    subject.handle_action(fail)
+
+    assert (
+        subject_view.get_state_update_for_false_positive()
+        == sentinel.state_update_if_false_positive
+    )
+
+    resume_from_recovery = actions.ResumeFromRecoveryAction(
+        state_update=sentinel.some_other_state_update
+    )
+    subject.handle_action(resume_from_recovery)
+
+    assert subject_view.get_state_update_for_false_positive() == empty_state_update
diff --git a/api/tests/opentrons/protocol_engine/state/test_command_store_old.py b/api/tests/opentrons/protocol_engine/state/test_command_store_old.py
index 4b7cf01e87c..5dc3c4a4ee9 100644
--- a/api/tests/opentrons/protocol_engine/state/test_command_store_old.py
+++ b/api/tests/opentrons/protocol_engine/state/test_command_store_old.py
@@ -334,7 +334,7 @@ def test_command_store_handles_pause_action(pause_source: PauseSource) -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
         failed_command_errors=[],
@@ -363,7 +363,7 @@ def test_command_store_handles_play_action(pause_source: PauseSource) -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=datetime(year=2021, month=1, day=1),
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
@@ -398,7 +398,7 @@ def test_command_store_handles_finish_action() -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=datetime(year=2021, month=1, day=1),
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
@@ -453,7 +453,7 @@ def test_command_store_handles_stop_action(
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=datetime(year=2021, month=1, day=1),
         latest_protocol_command_hash=None,
         stopped_by_estop=from_estop,
@@ -491,7 +491,7 @@ def test_command_store_handles_stop_action_when_awaiting_recovery() -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=datetime(year=2021, month=1, day=1),
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
@@ -525,7 +525,7 @@ def test_command_store_cannot_restart_after_should_stop() -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=None,
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
@@ -673,7 +673,7 @@ def test_command_store_wraps_unknown_errors() -> None:
         run_started_at=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
         failed_command_errors=[],
@@ -742,7 +742,7 @@ def __init__(self, message: str) -> None:
         ),
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=None,
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
@@ -778,7 +778,7 @@ def test_command_store_ignores_stop_after_graceful_finish() -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=datetime(year=2021, month=1, day=1),
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
@@ -814,7 +814,7 @@ def test_command_store_ignores_finish_after_non_graceful_stop() -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=datetime(year=2021, month=1, day=1),
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
@@ -850,7 +850,7 @@ def test_handles_hardware_stopped() -> None:
         finish_error=None,
         failed_command=None,
         command_error_recovery_types={},
-        recovery_target_command_id=None,
+        recovery_target=None,
         run_started_at=None,
         latest_protocol_command_hash=None,
         stopped_by_estop=False,
diff --git a/api/tests/opentrons/protocol_engine/state/test_command_view_old.py b/api/tests/opentrons/protocol_engine/state/test_command_view_old.py
index 06318cb8d36..f7b1d6cd31f 100644
--- a/api/tests/opentrons/protocol_engine/state/test_command_view_old.py
+++ b/api/tests/opentrons/protocol_engine/state/test_command_view_old.py
@@ -22,6 +22,9 @@
 
 from opentrons.protocol_engine.error_recovery_policy import ErrorRecoveryType
 from opentrons.protocol_engine.state.commands import (
+    # todo(mm, 2024-10-24): Avoid testing internal implementation details like
+    # _RecoveryTargetInfo. See note above about porting to test_command_state.py.
+    _RecoveryTargetInfo,
     CommandState,
     CommandView,
     CommandSlice,
@@ -38,6 +41,7 @@
 from opentrons_shared_data.errors.codes import ErrorCodes
 
 from opentrons.protocol_engine.state.command_history import CommandHistory
+from opentrons.protocol_engine.state.update_types import StateUpdate
 
 from .command_fixtures import (
     create_queued_command,
@@ -108,7 +112,12 @@ def get_command_view(  # noqa: C901
         finish_error=finish_error,
         failed_command=failed_command,
         command_error_recovery_types=command_error_recovery_types or {},
-        recovery_target_command_id=recovery_target_command_id,
+        recovery_target=_RecoveryTargetInfo(
+            command_id=recovery_target_command_id,
+            state_update_if_false_positive=StateUpdate(),
+        )
+        if recovery_target_command_id is not None
+        else None,
         run_started_at=run_started_at,
         latest_protocol_command_hash=latest_command_hash,
         stopped_by_estop=False,
@@ -592,7 +601,7 @@ class ActionAllowedSpec(NamedTuple):
                 ),
             ),
         ),
-        action=ResumeFromRecoveryAction(),
+        action=ResumeFromRecoveryAction(StateUpdate()),
         expected_error=errors.ResumeFromRecoveryNotAllowedError,
     ),
     ActionAllowedSpec(
diff --git a/api/tests/opentrons/protocol_engine/test_protocol_engine.py b/api/tests/opentrons/protocol_engine/test_protocol_engine.py
index 71e23cfe715..ac83e987153 100644
--- a/api/tests/opentrons/protocol_engine/test_protocol_engine.py
+++ b/api/tests/opentrons/protocol_engine/test_protocol_engine.py
@@ -10,6 +10,7 @@
 from opentrons_shared_data.robot.types import RobotType
 
 from opentrons.protocol_engine.actions.actions import SetErrorRecoveryPolicyAction
+from opentrons.protocol_engine.state.update_types import StateUpdate
 from opentrons.types import DeckSlotName
 from opentrons.hardware_control import HardwareControlAPI, OT2HardwareControlAPI
 from opentrons.hardware_control.modules import MagDeck, TempDeck
@@ -38,7 +39,11 @@
     HardwareStopper,
     DoorWatcher,
 )
-from opentrons.protocol_engine.resources import ModelUtils, ModuleDataProvider
+from opentrons.protocol_engine.resources import (
+    FileProvider,
+    ModelUtils,
+    ModuleDataProvider,
+)
 from opentrons.protocol_engine.state.config import Config
 from opentrons.protocol_engine.state.state import StateStore
 from opentrons.protocol_engine.plugins import AbstractPlugin, PluginStarter
@@ -118,6 +123,12 @@ def module_data_provider(decoy: Decoy) -> ModuleDataProvider:
     return decoy.mock(cls=ModuleDataProvider)
 
 
+@pytest.fixture
+def file_provider(decoy: Decoy) -> FileProvider:
+    """Get a mock FileProvider."""
+    return decoy.mock(cls=FileProvider)
+
+
 @pytest.fixture(autouse=True)
 def _mock_slot_standardization_module(
     decoy: Decoy, monkeypatch: pytest.MonkeyPatch
@@ -148,6 +159,7 @@ def subject(
     hardware_stopper: HardwareStopper,
     door_watcher: DoorWatcher,
     module_data_provider: ModuleDataProvider,
+    file_provider: FileProvider,
 ) -> ProtocolEngine:
     """Get a ProtocolEngine test subject with its dependencies stubbed out."""
     return ProtocolEngine(
@@ -160,6 +172,7 @@ def subject(
         hardware_stopper=hardware_stopper,
         door_watcher=door_watcher,
         module_data_provider=module_data_provider,
+        file_provider=file_provider,
     )
 
 
@@ -613,20 +626,31 @@ def test_pause(
     )
 
 
+@pytest.mark.parametrize("reconcile_false_positive", [True, False])
 def test_resume_from_recovery(
     decoy: Decoy,
     state_store: StateStore,
     action_dispatcher: ActionDispatcher,
     subject: ProtocolEngine,
+    reconcile_false_positive: bool,
 ) -> None:
     """It should dispatch a ResumeFromRecoveryAction."""
-    expected_action = ResumeFromRecoveryAction()
+    decoy.when(state_store.commands.get_state_update_for_false_positive()).then_return(
+        sentinel.state_update_for_false_positive
+    )
+    empty_state_update = StateUpdate()
+
+    expected_action = ResumeFromRecoveryAction(
+        sentinel.state_update_for_false_positive
+        if reconcile_false_positive
+        else empty_state_update
+    )
 
     decoy.when(
         state_store.commands.validate_action_allowed(expected_action)
     ).then_return(expected_action)
 
-    subject.resume_from_recovery()
+    subject.resume_from_recovery(reconcile_false_positive)
 
     decoy.verify(action_dispatcher.dispatch(expected_action))
 
diff --git a/api/tests/opentrons/protocol_runner/test_protocol_runner.py b/api/tests/opentrons/protocol_runner/test_protocol_runner.py
index cd945c33e64..2f06e27c2c2 100644
--- a/api/tests/opentrons/protocol_runner/test_protocol_runner.py
+++ b/api/tests/opentrons/protocol_runner/test_protocol_runner.py
@@ -313,9 +313,16 @@ def test_resume_from_recovery(
     subject: AnyRunner,
 ) -> None:
     """It should call `resume_from_recovery()` on the underlying engine."""
-    subject.resume_from_recovery()
+    subject.resume_from_recovery(
+        reconcile_false_positive=sentinel.reconcile_false_positive
+    )
 
-    decoy.verify(protocol_engine.resume_from_recovery(), times=1)
+    decoy.verify(
+        protocol_engine.resume_from_recovery(
+            reconcile_false_positive=sentinel.reconcile_false_positive
+        ),
+        times=1,
+    )
 
 
 async def test_run_json_runner(
diff --git a/robot-server/robot_server/runs/error_recovery_mapping.py b/robot-server/robot_server/runs/error_recovery_mapping.py
index b548394cd8a..52da8caaad8 100644
--- a/robot-server/robot_server/runs/error_recovery_mapping.py
+++ b/robot-server/robot_server/runs/error_recovery_mapping.py
@@ -101,11 +101,9 @@ def _rule_matches_error(
 def _map_error_recovery_type(reaction_if_match: ReactionIfMatch) -> ErrorRecoveryType:
     match reaction_if_match:
         case ReactionIfMatch.IGNORE_AND_CONTINUE:
-            return ErrorRecoveryType.IGNORE_AND_CONTINUE
+            return ErrorRecoveryType.CONTINUE_WITH_ERROR
         case ReactionIfMatch.ASSUME_FALSE_POSITIVE_AND_CONTINUE:
-            # todo(mm, 2024-10-23): Connect to work in
-            # https://github.com/Opentrons/opentrons/pull/16556.
-            return ErrorRecoveryType.IGNORE_AND_CONTINUE
+            return ErrorRecoveryType.ASSUME_FALSE_POSITIVE_AND_CONTINUE
         case ReactionIfMatch.FAIL_RUN:
             return ErrorRecoveryType.FAIL_RUN
         case ReactionIfMatch.WAIT_FOR_RECOVERY:
diff --git a/robot-server/robot_server/runs/run_controller.py b/robot-server/robot_server/runs/run_controller.py
index 1619cd20a08..41252d4dfc3 100644
--- a/robot-server/robot_server/runs/run_controller.py
+++ b/robot-server/robot_server/runs/run_controller.py
@@ -95,15 +95,22 @@ def create_action(
                 self._task_runner.run(self._run_orchestrator_store.stop)
 
             elif action_type == RunActionType.RESUME_FROM_RECOVERY:
-                self._run_orchestrator_store.resume_from_recovery()
+                log.info(f'Resuming run "{self._run_id}" from error recovery mode.')
+                self._run_orchestrator_store.resume_from_recovery(
+                    reconcile_false_positive=False
+                )
 
             elif (
                 action_type
                 == RunActionType.RESUME_FROM_RECOVERY_ASSUMING_FALSE_POSITIVE
             ):
-                # todo(mm, 2024-10-23): Connect to work in
-                # https://github.com/Opentrons/opentrons/pull/16556.
-                self._run_orchestrator_store.resume_from_recovery()
+                log.info(
+                    f'Resuming run "{self._run_id}" from error recovery mode,'
+                    f" assuming false-positive."
+                )
+                self._run_orchestrator_store.resume_from_recovery(
+                    reconcile_false_positive=True
+                )
 
             else:
                 assert_never(action_type)
diff --git a/robot-server/robot_server/runs/run_orchestrator_store.py b/robot-server/robot_server/runs/run_orchestrator_store.py
index efa97347ae9..bbc070c3b6f 100644
--- a/robot-server/robot_server/runs/run_orchestrator_store.py
+++ b/robot-server/robot_server/runs/run_orchestrator_store.py
@@ -310,9 +310,9 @@ async def stop(self) -> None:
         """Stop the run."""
         await self.run_orchestrator.stop()
 
-    def resume_from_recovery(self) -> None:
+    def resume_from_recovery(self, reconcile_false_positive: bool) -> None:
         """Resume the run from recovery mode."""
-        self.run_orchestrator.resume_from_recovery()
+        self.run_orchestrator.resume_from_recovery(reconcile_false_positive)
 
     async def finish(self, error: Optional[Exception]) -> None:
         """Finish the run."""
diff --git a/robot-server/tests/runs/test_error_recovery_mapping.py b/robot-server/tests/runs/test_error_recovery_mapping.py
index a125d12649d..8b75ff99aad 100644
--- a/robot-server/tests/runs/test_error_recovery_mapping.py
+++ b/robot-server/tests/runs/test_error_recovery_mapping.py
@@ -72,7 +72,7 @@ def test_create_error_recovery_policy_with_rules(
     mock_error_data: CommandDefinedErrorData,
     mock_rule: ErrorRecoveryRule,
 ) -> None:
-    """Should return IGNORE_AND_CONTINUE if that's what we specify as the rule."""
+    """Should return CONTINUE_WITH_ERROR if we specified IGNORE_AND_CONTINUE as the rule."""
     policy = create_error_recovery_policy_from_rules([mock_rule], enabled=True)
     example_config = Config(
         robot_type="OT-3 Standard",
@@ -80,7 +80,7 @@ def test_create_error_recovery_policy_with_rules(
     )
     assert (
         policy(example_config, mock_command, mock_error_data)
-        == ErrorRecoveryType.IGNORE_AND_CONTINUE
+        == ErrorRecoveryType.CONTINUE_WITH_ERROR
     )
 
 
@@ -141,7 +141,7 @@ def test_enabled_boolean(enabled: bool) -> None:
     policy = create_error_recovery_policy_from_rules(rules, enabled)
     result = policy(example_config, command, error_data)
     expected_result = (
-        ErrorRecoveryType.IGNORE_AND_CONTINUE if enabled else ErrorRecoveryType.FAIL_RUN
+        ErrorRecoveryType.CONTINUE_WITH_ERROR if enabled else ErrorRecoveryType.FAIL_RUN
     )
     assert result == expected_result
 
@@ -187,7 +187,7 @@ def test_enabled_on_flex_disabled_on_ot2(
     policy = create_error_recovery_policy_from_rules(rules, enabled=True)
     result = policy(example_config, command, error_data)
     expected_result = (
-        ErrorRecoveryType.IGNORE_AND_CONTINUE
+        ErrorRecoveryType.CONTINUE_WITH_ERROR
         if expect_error_recovery_to_be_enabled
         else ErrorRecoveryType.FAIL_RUN
     )
diff --git a/robot-server/tests/runs/test_run_controller.py b/robot-server/tests/runs/test_run_controller.py
index b069632a4e4..7dbdf827012 100644
--- a/robot-server/tests/runs/test_run_controller.py
+++ b/robot-server/tests/runs/test_run_controller.py
@@ -304,7 +304,9 @@ def test_create_resume_from_recovery_action(
     )
 
     decoy.verify(mock_run_store.insert_action(run_id, result), times=1)
-    decoy.verify(mock_run_orchestrator_store.resume_from_recovery())
+    decoy.verify(
+        mock_run_orchestrator_store.resume_from_recovery(reconcile_false_positive=False)
+    )
 
 
 @pytest.mark.parametrize(