diff --git a/api/src/opentrons/hardware_control/backends/ot3controller.py b/api/src/opentrons/hardware_control/backends/ot3controller.py
index f710eb405ac..7bbc6d5bc25 100644
--- a/api/src/opentrons/hardware_control/backends/ot3controller.py
+++ b/api/src/opentrons/hardware_control/backends/ot3controller.py
@@ -24,6 +24,7 @@
     KeysView,
     Union,
     Mapping,
+    Literal,
 )
 from opentrons.config.types import OT3Config, GantryLoad
 from opentrons.config import gripper_config
@@ -167,6 +168,7 @@
     liquid_probe,
     check_overpressure,
     grab_pressure,
+    move_plunger_while_tracking_z,
 )
 from opentrons_hardware.hardware_control.rear_panel_settings import (
     get_door_state,
@@ -720,6 +722,32 @@ def _build_move_gear_axis_runner(
             True,
         )
 
+    @requires_update
+    @requires_estop
+    async def aspirate_while_tracking(
+        self,
+        mount: OT3Mount,
+        distance: float,
+        speed: float,
+        direction: Union[Literal[1], Literal[-1]],
+        duration: float,
+    ) -> None:
+        head_node = axis_to_node(Axis.by_mount(mount))
+        tool = sensor_node_for_pipette(OT3Mount(mount.value))
+        async with self._monitor_overpressure([tool]):
+            positions = await move_plunger_while_tracking_z(
+                messenger=self._messenger,
+                tool=tool,
+                head_node=head_node,
+                distance=distance,
+                speed=speed,
+                direction=direction,
+                duration=duration,
+            )
+        for node, point in positions.items():
+            self._position.update({node: point.motor_position})
+            self._encoder_position.update({node: point.encoder_position})
+
     @requires_update
     @requires_estop
     async def move(
diff --git a/api/src/opentrons/hardware_control/ot3api.py b/api/src/opentrons/hardware_control/ot3api.py
index 7bb5e05f47b..31dfc004b99 100644
--- a/api/src/opentrons/hardware_control/ot3api.py
+++ b/api/src/opentrons/hardware_control/ot3api.py
@@ -2748,7 +2748,7 @@ async def liquid_probe(  # noqa: C901
         if not probe_settings:
             probe_settings = deepcopy(self.config.liquid_sense)
 
-        # We need to significatly slow down the 96 channel liquid probe
+        # We need to significantly slow down the 96 channel liquid probe
         if self.gantry_load == GantryLoad.HIGH_THROUGHPUT:
             max_plunger_speed = self.config.motion_settings.max_speed_discontinuity[
                 GantryLoad.HIGH_THROUGHPUT
@@ -2961,6 +2961,45 @@ async def capacitive_sweep(
 
     AMKey = TypeVar("AMKey")
 
+    async def aspirate_while_tracking(
+        self,
+        mount: Union[top_types.Mount, OT3Mount],
+        distance: float,
+        rate: float,
+        volume: Optional[float] = None,
+    ) -> None:
+        """
+        Aspirate a volume of liquid (in microliters/uL) using this pipette."""
+        realmount = OT3Mount.from_mount(mount)
+        aspirate_spec = self._pipette_handler.plan_check_aspirate(
+            realmount, volume, rate
+        )
+        if not aspirate_spec:
+            return
+
+        try:
+            await self._backend.set_active_current(
+                {aspirate_spec.axis: aspirate_spec.current}
+            )
+            async with self.restore_system_constrants():
+                await self.set_system_constraints_for_plunger_acceleration(
+                    realmount, aspirate_spec.acceleration
+                )
+                await self._backend.aspirate_while_tracking(
+                    mount=mount,
+                    distance=distance,
+                    speed=rate,
+                    direction=-1,
+                    # have to actually determine duration here
+                    duration=0.0,
+                )
+        except Exception:
+            self._log.exception("Aspirate failed")
+            aspirate_spec.instr.set_current_volume(0)
+            raise
+        else:
+            aspirate_spec.instr.add_current_volume(aspirate_spec.volume)
+
     @property
     def attached_subsystems(self) -> Dict[SubSystem, SubSystemState]:
         """Get a view of the state of the currently-attached subsystems."""
diff --git a/api/src/opentrons/protocol_api/core/engine/instrument.py b/api/src/opentrons/protocol_api/core/engine/instrument.py
index 010f3110fdb..67c68b37bfa 100644
--- a/api/src/opentrons/protocol_api/core/engine/instrument.py
+++ b/api/src/opentrons/protocol_api/core/engine/instrument.py
@@ -135,6 +135,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = False,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
@@ -192,6 +193,7 @@ def aspirate(
                     wellLocation=well_location,
                     volume=volume,
                     flowRate=flow_rate,
+                    is_tracking=is_tracking if is_tracking else False,
                 )
             )
 
diff --git a/api/src/opentrons/protocol_api/core/instrument.py b/api/src/opentrons/protocol_api/core/instrument.py
index bc1ec3669df..b512251967b 100644
--- a/api/src/opentrons/protocol_api/core/instrument.py
+++ b/api/src/opentrons/protocol_api/core/instrument.py
@@ -42,6 +42,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = False,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
diff --git a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
index d2d25051d49..bb5223558cb 100644
--- a/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
+++ b/api/src/opentrons/protocol_api/core/legacy/legacy_instrument_core.py
@@ -85,6 +85,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = None,
     ) -> None:
         """Aspirate a given volume of liquid from the specified location.
         Args:
diff --git a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
index ec194874528..0edb3af7cc1 100644
--- a/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
+++ b/api/src/opentrons/protocol_api/core/legacy_simulator/legacy_instrument_core.py
@@ -96,6 +96,7 @@ def aspirate(
         flow_rate: float,
         in_place: bool,
         is_meniscus: Optional[bool] = None,
+        is_tracking: Optional[bool] = None,
     ) -> None:
         if self.get_current_volume() == 0:
             # Make sure we're at the top of the labware and clear of any
diff --git a/api/src/opentrons/protocol_api/instrument_context.py b/api/src/opentrons/protocol_api/instrument_context.py
index 9c6338270c7..937b79210c2 100644
--- a/api/src/opentrons/protocol_api/instrument_context.py
+++ b/api/src/opentrons/protocol_api/instrument_context.py
@@ -292,6 +292,137 @@ def aspirate(
 
         return self
 
+    ########
+
+    @requires_version(2, 0)
+    def aspirate_while_tracking(
+        self,
+        volume: Optional[float] = None,
+        location: Optional[Union[types.Location, labware.Well]] = None,
+        rate: float = 1.0,
+    ) -> InstrumentContext:
+        """
+        Draw liquid into a pipette tip.
+
+        See :ref:`new-aspirate` for more details and examples.
+
+        :param volume: The volume to aspirate, measured in µL. If unspecified,
+                    defaults to the maximum volume for the pipette and its currently
+                    attached tip.
+
+                    If ``aspirate`` is called with a volume of precisely 0, its behavior
+                    depends on the API level of the protocol. On API levels below 2.16,
+                    it will behave the same as a volume of ``None``/unspecified: aspirate
+                    until the pipette is full. On API levels at or above 2.16, no liquid
+                    will be aspirated.
+        :type volume: int or float
+        :param location: Tells the robot where to aspirate from. The location can be
+                         a :py:class:`.Well` or a :py:class:`.Location`.
+
+                            - If the location is a ``Well``, the robot will aspirate at
+                              or above the bottom center of the well. The distance (in mm)
+                              from the well bottom is specified by
+                              :py:obj:`well_bottom_clearance.aspirate
+                              <well_bottom_clearance>`.
+
+                            - If the location is a ``Location`` (e.g., the result of
+                              :py:meth:`.Well.top` or :py:meth:`.Well.bottom`), the robot
+                              will aspirate from that specified position.
+
+                            - If the ``location`` is unspecified, the robot will
+                              aspirate from its current position.
+        :param rate: A multiplier for the default flow rate of the pipette. Calculated
+                     as ``rate`` multiplied by :py:attr:`flow_rate.aspirate
+                     <flow_rate>`. If not specified, defaults to 1.0. See
+                     :ref:`new-plunger-flow-rates`.
+        :type rate: float
+        :returns: This instance.
+
+        .. note::
+
+            If ``aspirate`` is called with a single, unnamed argument, it will treat
+            that argument as ``volume``. If you want to call ``aspirate`` with only
+            ``location``, specify it as a keyword argument:
+            ``pipette.aspirate(location=plate['A1'])``
+
+        """
+        _log.debug(
+            "aspirate {} from {} at {}".format(
+                volume, location if location else "current position", rate
+            )
+        )
+
+        move_to_location: types.Location
+        well: Optional[labware.Well] = None
+        is_meniscus: Optional[bool] = None
+        last_location = self._get_last_location_by_api_version()
+        try:
+            target = validation.validate_location(
+                location=location, last_location=last_location
+            )
+        except validation.NoLocationError as e:
+            raise RuntimeError(
+                "If aspirate is called without an explicit location, another"
+                " method that moves to a location (such as move_to or "
+                "dispense) must previously have been called so the robot "
+                "knows where it is."
+            ) from e
+
+        if isinstance(target, (TrashBin, WasteChute)):
+            raise ValueError(
+                "Trash Bin and Waste Chute are not acceptable location parameters for Aspirate commands."
+            )
+        move_to_location, well, is_meniscus = self._handle_aspirate_target(
+            target=target
+        )
+        if self.api_version >= APIVersion(2, 11):
+            instrument.validate_takes_liquid(
+                location=move_to_location,
+                reject_module=self.api_version >= APIVersion(2, 13),
+                reject_adapter=self.api_version >= APIVersion(2, 15),
+            )
+
+        if self.api_version >= APIVersion(2, 16):
+            c_vol = self._core.get_available_volume() if volume is None else volume
+        else:
+            c_vol = self._core.get_available_volume() if not volume else volume
+        flow_rate = self._core.get_aspirate_flow_rate(rate)
+
+        if (
+            self.api_version >= APIVersion(2, 20)
+            and well is not None
+            and self.liquid_presence_detection
+            and self._core.nozzle_configuration_valid_for_lld()
+            and self._core.get_current_volume() == 0
+        ):
+            self._raise_if_pressure_not_supported_by_pipette()
+            self.require_liquid_presence(well=well)
+
+        with publisher.publish_context(
+            broker=self.broker,
+            command=cmds.aspirate(
+                instrument=self,
+                volume=c_vol,
+                location=move_to_location,
+                flow_rate=flow_rate,
+                rate=rate,
+            ),
+        ):
+            self._core.aspirate(
+                location=move_to_location,
+                well_core=well._core if well is not None else None,
+                volume=c_vol,
+                rate=rate,
+                flow_rate=flow_rate,
+                in_place=target.in_place,
+                is_meniscus=is_meniscus,
+                is_tracking=True,
+            )
+
+        return self
+
+    ########
+
     @requires_version(2, 0)
     def dispense(  # noqa: C901
         self,
diff --git a/api/src/opentrons/protocol_engine/commands/aspirate.py b/api/src/opentrons/protocol_engine/commands/aspirate.py
index fa84afbde8c..c7e06e32ee3 100644
--- a/api/src/opentrons/protocol_engine/commands/aspirate.py
+++ b/api/src/opentrons/protocol_engine/commands/aspirate.py
@@ -11,7 +11,9 @@
     FlowRateMixin,
     BaseLiquidHandlingResult,
     aspirate_in_place,
+    aspirate_while_tracking,
     prepare_for_aspirate,
+    IsTrackingMixin,
 )
 from .movement_common import (
     LiquidHandlingWellLocationMixin,
@@ -47,7 +49,11 @@
 
 
 class AspirateParams(
-    PipetteIdMixin, AspirateVolumeMixin, FlowRateMixin, LiquidHandlingWellLocationMixin
+    PipetteIdMixin,
+    AspirateVolumeMixin,
+    FlowRateMixin,
+    LiquidHandlingWellLocationMixin,
+    IsTrackingMixin,
 ):
     """Parameters required to aspirate from a specific well."""
 
@@ -158,6 +164,7 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
             well_location=well_location,
             current_well=current_well,
             operation_volume=-params.volume,
+            is_tracking=params.is_tracking,
         )
         state_update.append(move_result.state_update)
         if isinstance(move_result, DefinedErrorData):
@@ -165,21 +172,38 @@ async def execute(self, params: AspirateParams) -> _ExecuteReturn:
                 public=move_result.public, state_update=state_update
             )
 
-        aspirate_result = await aspirate_in_place(
-            pipette_id=pipette_id,
-            volume=params.volume,
-            flow_rate=params.flowRate,
-            location_if_error={
-                "retryLocation": (
-                    move_result.public.position.x,
-                    move_result.public.position.y,
-                    move_result.public.position.z,
-                )
-            },
-            command_note_adder=self._command_note_adder,
-            pipetting=self._pipetting,
-            model_utils=self._model_utils,
-        )
+        if params.is_tracking:
+            aspirate_result = await aspirate_while_tracking(
+                pipette_id=pipette_id,
+                volume=params.volume,
+                flow_rate=params.flowRate,
+                location_if_error={
+                    "retryLocation": (
+                        move_result.public.position.x,
+                        move_result.public.position.y,
+                        move_result.public.position.z,
+                    )
+                },
+                command_note_adder=self._command_note_adder,
+                pipetting=self._pipetting,
+                model_utils=self._model_utils,
+            )
+        else:
+            aspirate_result = await aspirate_in_place(
+                pipette_id=pipette_id,
+                volume=params.volume,
+                flow_rate=params.flowRate,
+                location_if_error={
+                    "retryLocation": (
+                        move_result.public.position.x,
+                        move_result.public.position.y,
+                        move_result.public.position.z,
+                    )
+                },
+                command_note_adder=self._command_note_adder,
+                pipetting=self._pipetting,
+                model_utils=self._model_utils,
+            )
         state_update.append(aspirate_result.state_update)
         if isinstance(aspirate_result, DefinedErrorData):
             state_update.set_liquid_operated(
diff --git a/api/src/opentrons/protocol_engine/commands/movement_common.py b/api/src/opentrons/protocol_engine/commands/movement_common.py
index ca12d2d1ad8..5eb0fb5af4d 100644
--- a/api/src/opentrons/protocol_engine/commands/movement_common.py
+++ b/api/src/opentrons/protocol_engine/commands/movement_common.py
@@ -145,6 +145,7 @@ async def move_to_well(
     minimum_z_height: Optional[float] = None,
     speed: Optional[float] = None,
     operation_volume: Optional[float] = None,
+    is_tracking: Optional[bool] = False,
 ) -> MoveToWellOperationReturn:
     """Execute a move to well microoperation."""
     try:
@@ -158,6 +159,7 @@ async def move_to_well(
             minimum_z_height=minimum_z_height,
             speed=speed,
             operation_volume=operation_volume,
+            is_tracking=is_tracking,
         )
     except StallOrCollisionDetectedError as e:
         return DefinedErrorData(
diff --git a/api/src/opentrons/protocol_engine/commands/pipetting_common.py b/api/src/opentrons/protocol_engine/commands/pipetting_common.py
index 0292b51eee1..b10af9bed27 100644
--- a/api/src/opentrons/protocol_engine/commands/pipetting_common.py
+++ b/api/src/opentrons/protocol_engine/commands/pipetting_common.py
@@ -61,6 +61,15 @@ class FlowRateMixin(BaseModel):
     )
 
 
+class IsTrackingMixin(BaseModel):
+    """Mixin for the 'is_tracking' field of aspirate commands."""
+
+    is_tracking: bool = Field(
+        False,
+        description="Whether or not the pipette should move with the liquid while aspirating.",
+    )
+
+
 class BaseLiquidHandlingResult(BaseModel):
     """Base properties of a liquid handling result."""
 
@@ -211,6 +220,51 @@ async def aspirate_in_place(
         )
 
 
+async def aspirate_while_tracking(
+    pipette_id: str,
+    volume: float,
+    flow_rate: float,
+    location_if_error: ErrorLocationInfo,
+    command_note_adder: CommandNoteAdder,
+    pipetting: PipettingHandler,
+    model_utils: ModelUtils,
+) -> SuccessData[BaseLiquidHandlingResult] | DefinedErrorData[OverpressureError]:
+    """Execute an aspirate in place microoperation."""
+    try:
+        volume_aspirated = await pipetting.aspirate_while_tracking(
+            pipette_id=pipette_id,
+            volume=volume,
+            flow_rate=flow_rate,
+            command_note_adder=command_note_adder,
+        )
+    except PipetteOverpressureError as e:
+        return DefinedErrorData(
+            public=OverpressureError(
+                id=model_utils.generate_id(),
+                createdAt=model_utils.get_timestamp(),
+                wrappedErrors=[
+                    ErrorOccurrence.from_failed(
+                        id=model_utils.generate_id(),
+                        createdAt=model_utils.get_timestamp(),
+                        error=e,
+                    )
+                ],
+                errorInfo=location_if_error,
+            ),
+            state_update=StateUpdate().set_fluid_unknown(pipette_id=pipette_id),
+        )
+    else:
+        return SuccessData(
+            public=BaseLiquidHandlingResult(
+                volume=volume_aspirated,
+            ),
+            state_update=StateUpdate().set_fluid_aspirated(
+                pipette_id=pipette_id,
+                fluid=AspiratedFluid(kind=FluidKind.LIQUID, volume=volume_aspirated),
+            ),
+        )
+
+
 async def dispense_in_place(
     pipette_id: str,
     volume: float,
diff --git a/api/src/opentrons/protocol_engine/execution/movement.py b/api/src/opentrons/protocol_engine/execution/movement.py
index be8bbbb8de2..983fa0af3f4 100644
--- a/api/src/opentrons/protocol_engine/execution/movement.py
+++ b/api/src/opentrons/protocol_engine/execution/movement.py
@@ -74,6 +74,7 @@ async def move_to_well(
         minimum_z_height: Optional[float] = None,
         speed: Optional[float] = None,
         operation_volume: Optional[float] = None,
+        is_tracking: Optional[bool] = False,
     ) -> Point:
         """Move to a specific well."""
         self._state_store.labware.raise_if_labware_inaccessible_by_pipette(
diff --git a/api/src/opentrons/protocol_engine/execution/pipetting.py b/api/src/opentrons/protocol_engine/execution/pipetting.py
index 10d613e4dcf..4d088bb63d1 100644
--- a/api/src/opentrons/protocol_engine/execution/pipetting.py
+++ b/api/src/opentrons/protocol_engine/execution/pipetting.py
@@ -1,5 +1,5 @@
 """Pipetting command handling."""
-from typing import Optional, Iterator
+from typing import Optional, Iterator, Tuple
 from typing_extensions import Protocol as TypingProtocol
 from contextlib import contextmanager
 
@@ -45,6 +45,15 @@ async def aspirate_in_place(
     ) -> float:
         """Set flow-rate and aspirate."""
 
+    async def aspirate_while_tracking(
+        self,
+        pipette_id: str,
+        volume: float,
+        flow_rate: float,
+        command_note_adder: CommandNoteAdder,
+    ) -> float:
+        """Set flow-rate and aspirate while tracking."""
+
     async def dispense_in_place(
         self,
         pipette_id: str,
@@ -99,7 +108,25 @@ async def prepare_for_aspirate(self, pipette_id: str) -> None:
         hw_mount = self._state_view.pipettes.get_mount(pipette_id).to_hw_mount()
         await self._hardware_api.prepare_for_aspirate(mount=hw_mount)
 
-    async def aspirate_in_place(
+    def get_hw_aspirate_params(
+        self,
+        pipette_id: str,
+        volume: float,
+        command_note_adder: CommandNoteAdder,
+    ) -> Tuple[HardwarePipette, float]:
+        _adjusted_volume = _validate_aspirate_volume(
+            state_view=self._state_view,
+            pipette_id=pipette_id,
+            aspirate_volume=volume,
+            command_note_adder=command_note_adder,
+        )
+        _hw_pipette = self._state_view.pipettes.get_hardware_pipette(
+            pipette_id=pipette_id,
+            attached_pipettes=self._hardware_api.attached_instruments,
+        )
+        return _hw_pipette, _adjusted_volume
+
+    async def aspirate_while_tracking(
         self,
         pipette_id: str,
         volume: float,
@@ -112,15 +139,31 @@ async def aspirate_in_place(
             PipetteOverpressureError, propagated as-is from the hardware controller.
         """
         # get mount and config data from state and hardware controller
-        adjusted_volume = _validate_aspirate_volume(
-            state_view=self._state_view,
-            pipette_id=pipette_id,
-            aspirate_volume=volume,
-            command_note_adder=command_note_adder,
+        hw_pipette, adjusted_volume = self.get_hw_aspirate_params(
+            pipette_id, volume, command_note_adder
         )
-        hw_pipette = self._state_view.pipettes.get_hardware_pipette(
-            pipette_id=pipette_id,
-            attached_pipettes=self._hardware_api.attached_instruments,
+        with self._set_flow_rate(pipette=hw_pipette, aspirate_flow_rate=flow_rate):
+            await self._hardware_api.aspirate_while_tracking(
+                mount=hw_pipette.mount, volume=adjusted_volume
+            )
+
+        return adjusted_volume
+
+    async def aspirate_in_place(
+        self,
+        pipette_id: str,
+        volume: float,
+        flow_rate: float,
+        command_note_adder: CommandNoteAdder,
+    ) -> float:
+        """Set flow-rate and aspirate.
+
+        Raises:
+            PipetteOverpressureError, propagated as-is from the hardware controller.
+        """
+        # get mount and config data from state and hardware controller
+        hw_pipette, adjusted_volume = self.get_hw_aspirate_params(
+            pipette_id, volume, command_note_adder
         )
         with self._set_flow_rate(pipette=hw_pipette, aspirate_flow_rate=flow_rate):
             await self._hardware_api.aspirate(
diff --git a/api/src/opentrons/protocol_engine/state/geometry.py b/api/src/opentrons/protocol_engine/state/geometry.py
index ed915530b90..33ece5438c1 100644
--- a/api/src/opentrons/protocol_engine/state/geometry.py
+++ b/api/src/opentrons/protocol_engine/state/geometry.py
@@ -473,6 +473,7 @@ def get_well_position(
         well_location: Optional[WellLocations] = None,
         operation_volume: Optional[float] = None,
         pipette_id: Optional[str] = None,
+        is_tracking: Optional[bool] = False,
     ) -> Point:
         """Given relative well location in a labware, get absolute position."""
         labware_pos = self.get_labware_position(labware_id)
@@ -488,6 +489,7 @@ def get_well_position(
                 well_location=well_location,
                 well_depth=well_depth,
                 operation_volume=operation_volume,
+                is_tracking=is_tracking,
             )
             offset = offset.copy(update={"z": offset.z + offset_adjustment})
             self.validate_well_position(
@@ -1411,6 +1413,7 @@ def get_well_offset_adjustment(
         well_location: WellLocations,
         well_depth: float,
         operation_volume: Optional[float] = None,
+        is_tracking: Optional[bool] = False,
     ) -> float:
         """Return a z-axis distance that accounts for well handling height and operation volume.
 
@@ -1423,6 +1426,8 @@ def get_well_offset_adjustment(
             well_location=well_location,
             well_depth=well_depth,
         )
+        if is_tracking:
+            return initial_handling_height
         if isinstance(well_location, PickUpTipWellLocation):
             volume = 0.0
         elif isinstance(well_location.volumeOffset, float):
diff --git a/api/src/opentrons/protocol_engine/state/motion.py b/api/src/opentrons/protocol_engine/state/motion.py
index 0863c42a0c1..4f41474f05b 100644
--- a/api/src/opentrons/protocol_engine/state/motion.py
+++ b/api/src/opentrons/protocol_engine/state/motion.py
@@ -98,6 +98,7 @@ def get_movement_waypoints_to_well(
         force_direct: bool = False,
         minimum_z_height: Optional[float] = None,
         operation_volume: Optional[float] = None,
+        is_tracking: Optional[bool] = False,
     ) -> List[motion_planning.Waypoint]:
         """Calculate waypoints to a destination that's specified as a well."""
         location = current_well or self._pipettes.get_current_location()
@@ -114,6 +115,7 @@ def get_movement_waypoints_to_well(
             well_location=well_location,
             operation_volume=operation_volume,
             pipette_id=pipette_id,
+            is_tracking=is_tracking,
         )
 
         move_type = _move_types.get_move_type_to_well(
diff --git a/hardware/opentrons_hardware/hardware_control/tool_sensors.py b/hardware/opentrons_hardware/hardware_control/tool_sensors.py
index e9475de8b95..512b6a2cb6f 100644
--- a/hardware/opentrons_hardware/hardware_control/tool_sensors.py
+++ b/hardware/opentrons_hardware/hardware_control/tool_sensors.py
@@ -258,6 +258,32 @@ async def finalize_logs(
     for listener in listeners.values():
         await listener.wait_for_complete()
 
+async def move_plunger_while_tracking_z(
+    messenger: CanMessenger,
+    tool: PipetteProbeTarget,
+    head_node: NodeId,
+    distance: float,
+    speed: float,
+    direction: Union[Literal[1], Literal[-1]],
+    duration: float,
+) -> Dict[NodeId, MotorPositionStatus]:
+    liquid_action_step = create_step(
+        distance={
+            tool: float64(abs(distance) * direction),
+            head_node: float64(abs(distance) * direction)
+        },
+        velocity={
+            tool: float64(abs(speed) * direction),
+            head_node: float64(abs(speed) * direction)
+        },
+        acceleration={},
+        duration=float64(duration),
+        present_nodes=[tool],
+    )
+    runner = MoveGroupRunner(move_groups=[[liquid_action_step]])
+    positions = await runner.run(can_messenger=messenger)
+    return positions
+
 
 async def liquid_probe(
     messenger: CanMessenger,