diff --git a/docs/trials_table_mapping.md b/docs/trials_table_mapping.md
index c69f3db..afe3fd7 100644
--- a/docs/trials_table_mapping.md
+++ b/docs/trials_table_mapping.md
@@ -61,8 +61,9 @@ Columns are grouped by the raw source they map from.
 
 | Trials column | Mapping |
 | --- | --- |
-| `auto_waterL` / `auto_waterR` | From `is_auto_reward_right`. `1` on the auto-responded side; `0` on the other side, when there was no auto-response (`None`), or when the trial is missing. |
-| `anti_bias_left_water` / `anti_bias_right_water` | Boolean. `True` when the anti-bias algorithm delivered a water intervention to that side — i.e. `trial.metadata.extra.is_bias_water_intervention` is `True` **and** `is_auto_reward_right` points to that side (`False` → left, `True` → right). The anti-bias water uses the same auto-response channel as ordinary autowater, so the `is_bias_water_intervention` flag is what distinguishes it. `False` otherwise. |
+| `auto_waterL` / `auto_waterR` | **Rewarded** autowater only: `1` when `is_auto_reward_right` points to that side **and** `is_rewarded` is `True`. `0` on the other side, when there was no auto-response (`None`), when the trial delivered autowater but did not pay out, or when the trial is missing. `is_auto_reward_right` is the delivery channel — it says free water was triggered and to which side, not what kind — and the anti-bias water intervention is autowater delivered through it, so anti-bias water counts here too. Note this is narrower than the legacy `dynamic-foraging-task` column of the same name, which was ungated ("Autowater given at Left", straight from `B_AutoWaterTrial`). |
+| `auto_water_offeredL` / `auto_water_offeredR` | Ungated autowater: `1` whenever `is_auto_reward_right` points to that side, whether or not the trial paid out. This is the legacy `dynamic-foraging-task` meaning of `auto_waterL` / `auto_waterR` (`B_AutoWaterTrial`, which that GUI also passes to foraging efficiency as `autowater_offered`), kept so the trial table records every autowater delivery. `auto_waterL` / `auto_waterR` are the reward-keyed subset. |
+| `anti_bias_left_water` / `anti_bias_right_water` | Boolean. `True` when the anti-bias algorithm delivered a water intervention to that side — i.e. `trial.metadata.extra.is_bias_water_intervention` is `True` **and** `is_auto_reward_right` points to that side (`False` → left, `True` → right). The anti-bias water uses the same auto-response channel as ordinary autowater, so the `is_bias_water_intervention` flag is what distinguishes it. `False` otherwise. Deliberately **not** gated on `is_rewarded`: these columns record what the anti-bias algorithm did, and the intervention fires at the go cue regardless of how the animal's own choice resolves, so they can be `True` where `auto_waterL` / `auto_waterR` is `0`. |
 | `anti_bias_lickspout_movement` | Signed horizontal displacement (mm, positive is rightward) the anti-bias algorithm moved the lickspouts on this trial: `trial.lickspout_offset_delta` when `trial.metadata.extra.is_bias_stage_intervention` is `True`, else `0.0`. |
 | `bait_left` / `bait_right` | Boolean. `bait_right` is `True` if `p_reward_right == 1` and `is_auto_reward_right` is `None` or `False`. `bait_left` is `True` if `p_reward_left == 1` and `is_auto_reward_right` is `None` or `True`. |
 | `response_duration` | `response_deadline_duration`. |
@@ -191,3 +192,7 @@ These were mapped during exploration but are no longer in scope:
 | 2026-08-06 | Confirmed and documented that the legacy `delay_*` columns describe the acquisition software's **quiescence period**: `delay_start_time` is the `QuiescentPeriod` timestamp (always equal to the new `quiescent_start_time`) and `delay_duration` / `delay_beta` / `delay_min` / `delay_max` summarize `quiescence_period_duration`. `delay_duration` is the *configured* duration — each lick restarts the quiescent period, so the realized `quiescent_stop_time - quiescent_start_time` can be longer. Column descriptions updated accordingly. |
 | 2026-08-12 | `rewarded_historyL` / `rewarded_historyR` now record **earned** reward only: an auto-reward trial (`is_auto_reward_right` set to either side) is `False` on *both* sides, since `TrialOutcome.is_rewarded` is `True` for autowater too and that water is already reported by `auto_waterL` / `auto_waterR`. This matches the `earned` / `automatic` split used for the NWB reward-delivery annotations. |
 | 2026-08-12 | `min_reward_each_block` is now `0` rather than `NULL` when the trial generator exposes no `min_block_reward` — no per-block minimum is a floor of zero, not an unknown. The column is non-nullable (`float`, default `0`). |
+| 2026-08-17 | Reward-delivery annotations now match each delivery to its trial by the `Response` software-event timestamp rather than the `TrialOutcome` timestamp. `TrialOutcome` fires at the *end* of a trial (after the reward-consumption and ITI periods), so a delivery could land nearer the *previous* trial's outcome and inherit its `is_auto_reward_right`, flipping `earned` and `auto`. The valve opens within milliseconds of the response, so the response anchors the delivery to its own trial. |
+| 2026-08-17 | `auto_waterL` / `auto_waterR` are now gated on `is_rewarded`: autowater on a trial that did not pay out is `0`. Free water is triggered immediately at the go cue and the trial then "continues normally", so `is_rewarded` reports the outcome of the animal's own choice — a separate event from the autowater. Gating keeps these columns equal to the `auto` count in the reward-delivery series, which drops the same deliveries, and makes the retained delivery total equal the metadata mapper's `sum(is_rewarded)`. This is narrower than the legacy ungated column of the same name. |
+| 2026-08-17 | `anti_bias_left_water` / `anti_bias_right_water` are explicitly **not** gated on `is_rewarded`, unlike `auto_water*`: they record what the anti-bias algorithm did, and the intervention fires regardless of the trial's outcome. They are therefore no longer a subset of `auto_water*` — an intervention on a trial that did not pay out appears in the anti-bias column and not in the autowater column. |
+| 2026-08-17 | Added `auto_water_offeredL` / `auto_water_offeredR`: autowater ungated by `is_rewarded`, i.e. the legacy `dynamic-foraging-task` meaning of `auto_waterL` / `auto_waterR` (`B_AutoWaterTrial`). With `auto_water*` now reward-keyed to match the reward-delivery series, these keep every autowater delivery recorded in the trial table, so nothing is lost: `auto_water_offered* - auto_water*` is the autowater delivered on trials that did not pay out. |
diff --git a/src/dynamic_foraging_processing/nwb/acquisition/acquisition_builder.py b/src/dynamic_foraging_processing/nwb/acquisition/acquisition_builder.py
index ebd5270..c46acef 100644
--- a/src/dynamic_foraging_processing/nwb/acquisition/acquisition_builder.py
+++ b/src/dynamic_foraging_processing/nwb/acquisition/acquisition_builder.py
@@ -11,7 +11,7 @@
 )
 from dynamic_foraging_processing.nwb.utils import clean_for_nwb
 from dynamic_foraging_processing.raw_data_loader import RawDataLoader
-from dynamic_foraging_processing.utils.rewards import get_annotated_rewards
+from dynamic_foraging_processing.utils.rewards import get_reward_deliveries
 
 
 class LickSource(t.NamedTuple):
@@ -49,8 +49,8 @@ def __init__(self, loader: RawDataLoader):
         """
         self.loader = loader
 
-    def get_reward_delivery(self) -> pd.DataFrame:
-        """Get the reward delivery stream from the dataset.
+    def get_valve_writes(self) -> pd.DataFrame:
+        """Get the raw valve command stream.
 
         Returns
         -------
@@ -76,6 +76,26 @@ def get_trial_outcomes(self) -> pd.DataFrame:
             self.loader.dataset.at("Behavior").at("SoftwareEvents").at("TrialOutcome").load().data
         )
 
+    def get_response_times(self) -> np.ndarray:
+        """Get the per-trial ``Response`` software-event timestamps.
+
+        The event fires when the animal's choice is registered, within
+        milliseconds of the valve opening, so it anchors a reward delivery to
+        its trial. Only the event timestamp is used; the payload's ``Item1``
+        field nominally carries a response time but is unreliable (it can lag
+        the event by thousands of seconds), so it is ignored.
+
+        Returns
+        -------
+        numpy.ndarray
+            The ``Response`` event timestamps, positionally aligned with the
+            ``TrialOutcome`` stream.
+        """
+        responses = (
+            self.loader.dataset.at("Behavior").at("SoftwareEvents").at("Response").load().data
+        )
+        return responses.index.to_numpy()
+
     def get_manual_water_times(self) -> pd.DataFrame:
         """Get the manual-water software-event stream.
 
@@ -172,6 +192,7 @@ def _reward_delivery_series(
         writes: pd.DataFrame,
         trial_outcomes: pd.DataFrame,
         manual_water: pd.DataFrame,
+        response_times: np.ndarray,
         *,
         port_column: str,
         is_right: bool,
@@ -181,8 +202,10 @@ def _reward_delivery_series(
         """Build one lick port's reward-delivery series with reward annotations.
 
         Only valve-open events (``port_column`` is truthy) are reward
-        deliveries; the ``data`` field annotates each as earned, manual, or
-        auto via :func:`get_annotated_rewards`.
+        deliveries; the ``data`` field annotates each as earned, manual, or auto
+        via :func:`get_reward_deliveries`, which also drops deliveries on trials
+        that did not pay out, so the series reports reward rather than every
+        valve opening.
 
         Parameters
         ----------
@@ -193,6 +216,9 @@ def _reward_delivery_series(
         manual_water : pandas.DataFrame
             The ``GiveManualWaterRight`` stream; the ``data`` column selects the
             side (``True`` right, ``False`` left).
+        response_times : numpy.ndarray
+            ``Response`` event timestamps, one per trial, used to match each
+            delivery to its trial.
         port_column : str
             Supply-port column for this side (``"SupplyPort0"`` left,
             ``"SupplyPort1"`` right).
@@ -211,10 +237,11 @@ def _reward_delivery_series(
         open_writes = writes[writes[port_column].fillna(False).astype(bool)]
         delivery_times = open_writes.index.to_numpy()
         manual_water_times = manual_water.index[manual_water["data"] == is_right].to_numpy()
-        annotations = get_annotated_rewards(
+        delivery_times, annotations = get_reward_deliveries(
             delivery_times,
             trial_outcomes,
             manual_water_times,
+            response_times,
         )
         return AcquisitionSeries(
             name=name,
@@ -248,9 +275,10 @@ def build_acquisition(
         list of AcquisitionSeries or AcquisitionTable
             Acquisition entries to write to the NWB acquisition module.
         """
-        rewards = self.get_reward_delivery()
+        rewards = self.get_valve_writes()
         trial_outcomes = self.get_trial_outcomes()
         manual_water = self.get_manual_water_times()
+        response_times = self.get_response_times()
 
         acquisition_streams = self.loader.get_all_raw_data()
         acqusition_streams_descriptions = self.loader.raw_data_stream_descriptions
@@ -274,6 +302,7 @@ def build_acquisition(
                 rewards,
                 trial_outcomes,
                 manual_water,
+                response_times,
                 port_column="SupplyPort0",
                 is_right=False,
                 name="left_reward_delivery_time",
@@ -285,6 +314,7 @@ def build_acquisition(
                 rewards,
                 trial_outcomes,
                 manual_water,
+                response_times,
                 port_column="SupplyPort1",
                 is_right=True,
                 name="right_reward_delivery_time",
diff --git a/src/dynamic_foraging_processing/processing/_trial_table.py b/src/dynamic_foraging_processing/processing/_trial_table.py
index 724828b..52974d2 100644
--- a/src/dynamic_foraging_processing/processing/_trial_table.py
+++ b/src/dynamic_foraging_processing/processing/_trial_table.py
@@ -25,6 +25,7 @@
 from contraqctor.contract import Dataset
 
 from dynamic_foraging_processing.processing.models import TrialConfig
+from dynamic_foraging_processing.utils.trial_metadata import get_bias_metadata
 
 logger = logging.getLogger(__name__)
 
@@ -353,7 +354,7 @@ def _rewarded_history(
         (``trial.is_auto_reward_right is not None``) is ``False`` on *both*
         sides here — its water is reported by ``auto_waterL``/``auto_waterR``
         instead. This matches the ``earned``/``automatic`` split in
-        :func:`~dynamic_foraging_processing.utils.rewards.get_annotated_rewards`.
+        :func:`~dynamic_foraging_processing.utils.rewards.get_reward_deliveries`.
 
         A trial with no reward or an ignored trial (no choice) likewise counts
         as not rewarded on either side (``False``).
@@ -442,28 +443,48 @@ def _is_baited(trial: Trial, *, is_right: bool) -> bool:
         return trial.p_reward_left == 1 and auto in (None, True)
 
     @staticmethod
-    def _auto_water(trial: Trial, *, is_right: bool) -> int:
-        """Encode autowater for a side from ``is_auto_reward_right``.
+    def _auto_water(trial: Trial, bias_metadata: BlockBasedTrialMetadata, *, is_right: bool) -> int:
+        """Return whether scheduled autowater was delivered to the requested side.
+
+        ``is_auto_reward_right`` is only the delivery *channel* -- it says free
+        water was triggered and to which side (``True`` right, ``False`` left,
+        ``None`` none), not what kind. Scheduled autowater and the anti-bias water
+        intervention share that channel, so the mechanism comes from
+        ``is_autowater`` and the side from the channel -- the mirror of
+        :meth:`_anti_bias_water`. Free water driven by the anti-bias algorithm is
+        ``0`` here and is reported by
+        ``anti_bias_left_water``/``anti_bias_right_water`` instead.
+
+        Like the anti-bias columns, this records what the *task* did and so is not
+        gated on ``is_rewarded``: the water fires at the go cue regardless of how
+        the animal's own choice later resolves. The reward-delivery series is
+        reward-keyed and drops free water on trials that did not pay out, so this
+        column can exceed that series' ``auto`` count.
 
-        Returns ``1`` if the auto response was to the requested side, else ``0``.
-        No auto-response (``is_auto_reward_right`` is ``None``) counts as no
-        autowater (``0``). ``is_right`` is ``True`` for right.
+        Parameters
+        ----------
+        trial : Trial
+            The per-trial task-logic model.
+        bias_metadata : BlockBasedTrialMetadata
+            The trial's extra metadata (see ``_bias_metadata``).
+        is_right : bool
+            ``True`` for the right port, ``False`` for the left port.
+
+        Returns
+        -------
+        int
+            ``1`` when scheduled autowater targeted the requested side, else ``0``.
         """
-        if trial.is_auto_reward_right is None:
+        if not bias_metadata.is_autowater:
             return 0
         return int(trial.is_auto_reward_right is is_right)
 
     @staticmethod
     def _bias_metadata(trial: Trial) -> BlockBasedTrialMetadata:
-        """Return the block-based extra metadata carrying the anti-bias flags.
+        """Return the block-based extra metadata carrying the free-water flags.
 
-        The anti-bias flags (``is_bias_water_intervention``,
-        ``is_bias_stage_intervention``) live on ``trial.metadata.extra``. That
-        field is schema-typed ``Any``, so it deserializes off the stream as a
-        plain ``dict`` rather than a model; a ``BlockBasedTrialMetadata``
-        instance is also accepted. When metadata or extra is missing (e.g. an
-        older session, or a non-block-based generator), the model's all-``False``
-        default is returned so the anti-bias columns are simply inert.
+        Thin wrapper over :func:`get_bias_metadata`, shared with the reward
+        annotation so both classify autowater and anti-bias water identically.
 
         Parameters
         ----------
@@ -476,13 +497,7 @@ def _bias_metadata(trial: Trial) -> BlockBasedTrialMetadata:
             The parsed extra metadata, or an all-``False`` default when absent
             or unrecognized.
         """
-        metadata = trial.metadata
-        extra = metadata.extra if metadata is not None else None
-        if isinstance(extra, BlockBasedTrialMetadata):
-            return extra
-        if isinstance(extra, dict):
-            return BlockBasedTrialMetadata.model_validate(extra)
-        return BlockBasedTrialMetadata()
+        return get_bias_metadata(trial)
 
     @staticmethod
     def _anti_bias_water(
@@ -497,6 +512,13 @@ def _anti_bias_water(
         was a bias-water intervention *and* the auto-response was to the
         requested side.
 
+        This records what the *algorithm* did, so it is not gated on
+        ``is_rewarded``: the intervention fires at the go cue regardless of how
+        the animal's own choice later resolves. It is therefore not a subset of
+        ``auto_waterL``/``auto_waterR``, which count only rewarded autowater --
+        an intervention on a trial that did not pay out appears here and not
+        there.
+
         Parameters
         ----------
         trial : Trial
@@ -897,8 +919,8 @@ def _build_row(
             reward_consumption_duration=trial.reward_consumption_duration,
             ITI_duration=trial.inter_trial_interval_duration,
             delay_duration=trial.quiescence_period_duration,
-            auto_waterL=self._auto_water(trial, is_right=False),
-            auto_waterR=self._auto_water(trial, is_right=True),
+            auto_waterL=self._auto_water(trial, bias_metadata, is_right=False),
+            auto_waterR=self._auto_water(trial, bias_metadata, is_right=True),
             anti_bias_left_water=self._anti_bias_water(trial, bias_metadata, is_right=False),
             anti_bias_right_water=self._anti_bias_water(trial, bias_metadata, is_right=True),
             anti_bias_lickspout_movement=self._anti_bias_lickspout_movement(trial, bias_metadata),
diff --git a/src/dynamic_foraging_processing/processing/models/trial_config.py b/src/dynamic_foraging_processing/processing/models/trial_config.py
index 22e0e96..94b0f47 100644
--- a/src/dynamic_foraging_processing/processing/models/trial_config.py
+++ b/src/dynamic_foraging_processing/processing/models/trial_config.py
@@ -189,20 +189,30 @@ class TrialConfig(BaseModel):
     )
 
     # --- auto_waterL/R (autowater per-side; autoTrain curriculum fields out of scope) ---
-    auto_waterL: int = Field(default=0, description="Autowater given at Left")
-    auto_waterR: int = Field(default=0, description="Autowater given at Right")
+    auto_waterL: int = Field(
+        default=0,
+        description=(
+            "Scheduled autowater at Left: 1 when the trial's free water was scheduled autowater (trial.metadata.extra.is_autowater) and was directed left (is_auto_reward_right is False). is_auto_reward_right is only the delivery channel, so the mechanism comes from the metadata: free water driven by the anti-bias algorithm is 0 here and is reported by anti_bias_left_water instead. Records what the task did, so like the anti-bias columns it is NOT conditioned on is_rewarded; the reward-delivery series is reward-keyed and drops free water on trials that did not pay out, so this column can exceed that series' auto count."
+        ),
+    )
+    auto_waterR: int = Field(
+        default=0,
+        description=(
+            "Scheduled autowater at Right: 1 when the trial's free water was scheduled autowater (trial.metadata.extra.is_autowater) and was directed right (is_auto_reward_right is True). is_auto_reward_right is only the delivery channel, so the mechanism comes from the metadata: free water driven by the anti-bias algorithm is 0 here and is reported by anti_bias_right_water instead. Records what the task did, so like the anti-bias columns it is NOT conditioned on is_rewarded; the reward-delivery series is reward-keyed and drops free water on trials that did not pay out, so this column can exceed that series' auto count."
+        ),
+    )
 
     # --- anti_bias (interventions the anti-bias algorithm applies) ---
     anti_bias_left_water: bool = Field(
         default=False,
         description=(
-            "Whether the anti-bias algorithm delivered a water intervention to the left lickport on this trial."
+            "Whether the anti-bias algorithm delivered a water intervention to the left lickport on this trial. Records what the algorithm did, so unlike auto_waterL this is NOT conditioned on is_rewarded: the intervention fires at the go cue regardless of how the animal's own choice resolves. It can therefore be True where auto_waterL is 0."
         ),
     )
     anti_bias_right_water: bool = Field(
         default=False,
         description=(
-            "Whether the anti-bias algorithm delivered a water intervention to the right lickport on this trial."
+            "Whether the anti-bias algorithm delivered a water intervention to the right lickport on this trial. Records what the algorithm did, so unlike auto_waterR this is NOT conditioned on is_rewarded: the intervention fires at the go cue regardless of how the animal's own choice resolves. It can therefore be True where auto_waterR is 0."
         ),
     )
     anti_bias_lickspout_movement: float = Field(
diff --git a/src/dynamic_foraging_processing/utils/__init__.py b/src/dynamic_foraging_processing/utils/__init__.py
index 1c77e9b..cecf2d5 100644
--- a/src/dynamic_foraging_processing/utils/__init__.py
+++ b/src/dynamic_foraging_processing/utils/__init__.py
@@ -1,6 +1,6 @@
 """Utility helpers for dynamic foraging processing."""
 
-from dynamic_foraging_processing.utils.rewards import get_annotated_rewards
+from dynamic_foraging_processing.utils.rewards import get_reward_deliveries
 from dynamic_foraging_processing.utils.timestamps import find_closest_timestamps
 
-__all__ = ["find_closest_timestamps", "get_annotated_rewards"]
+__all__ = ["find_closest_timestamps", "get_reward_deliveries"]
diff --git a/src/dynamic_foraging_processing/utils/rewards.py b/src/dynamic_foraging_processing/utils/rewards.py
index 02234a8..ad6d59d 100644
--- a/src/dynamic_foraging_processing/utils/rewards.py
+++ b/src/dynamic_foraging_processing/utils/rewards.py
@@ -4,9 +4,10 @@
 
 import numpy as np
 import pandas as pd
-from aind_behavior_dynamic_foraging.task_logic.trial_models import TrialOutcome
+from aind_behavior_dynamic_foraging.task_logic.trial_models import Trial, TrialOutcome
 
 from dynamic_foraging_processing.utils.timestamps import find_closest_timestamps
+from dynamic_foraging_processing.utils.trial_metadata import get_bias_metadata
 
 
 def _parse_outcome(payload: t.Any) -> t.Optional[TrialOutcome]:
@@ -30,12 +31,48 @@ def _parse_outcome(payload: t.Any) -> t.Optional[TrialOutcome]:
     return TrialOutcome.model_validate(payload)
 
 
-def get_annotated_rewards(
+def _free_water_label(trial: t.Optional[Trial]) -> str:
+    """Classify a delivery's trial as ``anti_bias``, ``auto``, or ``earned``.
+
+    ``is_auto_reward_right`` is only the delivery *channel* -- it says free water
+    was triggered and to which side, not what kind -- so the mechanism comes from
+    the block-based metadata: ``is_bias_water_intervention`` for an anti-bias
+    correction, ``is_autowater`` for scheduled autowater. These are the same
+    conditions ``anti_bias_left_water``/``anti_bias_right_water`` and
+    ``auto_waterL``/``auto_waterR`` use in the trials table, so the labels and the
+    columns classify each trial identically.
+
+    A trial with no free water, or free water the metadata flags as neither
+    mechanism, is ``earned``.
+
+    Parameters
+    ----------
+    trial : Trial or None
+        The per-trial task-logic model, or ``None`` when the outcome payload was
+        missing.
+
+    Returns
+    -------
+    str
+        ``"anti_bias"``, ``"auto"``, or ``"earned"``.
+    """
+    if trial is None or trial.is_auto_reward_right is None:
+        return "earned"
+    metadata = get_bias_metadata(trial)
+    if metadata.is_bias_water_intervention:
+        return "anti_bias"
+    if metadata.is_autowater:
+        return "auto"
+    return "earned"
+
+
+def get_reward_deliveries(
     reward_delivery_times: np.ndarray,
     trial_outcome_df: pd.DataFrame,
     manual_water_times: np.ndarray,
-) -> np.ndarray:
-    """Annotate each reward delivery as ``earned``, ``auto``, or ``manual``.
+    response_times: np.ndarray,
+) -> t.Tuple[np.ndarray, np.ndarray]:
+    """Get one lick port's reward deliveries, classified by how the water was given.
 
     Annotates the deliveries of a single lick port. Each delivery is classified
     as follows, with ``manual`` taking precedence because manual water is not
@@ -45,12 +82,41 @@ def get_annotated_rewards(
       ``GiveManualWater`` software event for this port. The software-event
       timestamps are correlated to the reward-delivery timestamps with
       :func:`find_closest_timestamps`.
-    - ``auto`` -- otherwise, when the matching trial auto-responded
-      (``is_auto_reward_right is not None``).
-    - ``earned`` -- otherwise (no matching trial, or no auto-response).
-
-    Deliveries are matched to trials by the ``TrialOutcome`` software-event
-    timestamp: each delivery takes the annotation of the closest trial.
+    - ``anti_bias`` -- otherwise, when the trial's free water came from the
+      anti-bias algorithm (``is_bias_water_intervention``).
+    - ``auto`` -- otherwise, when the trial's free water was scheduled autowater
+      (``is_autowater``).
+    - ``earned`` -- otherwise: water the animal worked for.
+
+    ``is_auto_reward_right`` is only the delivery *channel*, so the mechanism
+    behind free water comes from the block-based metadata (see
+    :func:`get_bias_metadata`) -- the same fields the trials table's
+    ``auto_waterL``/``auto_waterR`` and
+    ``anti_bias_left_water``/``anti_bias_right_water`` read.
+
+    Deliveries on a trial reporting ``is_rewarded=False`` are dropped rather than
+    annotated, so the series reports only water that counted as reward. In
+    practice these are all free water: it is triggered immediately at the go cue
+    and the trial then continues normally, so a trial whose own choice did not
+    pay out still carries the delivery. Manual water is experimenter-driven and
+    is never dropped. The surviving timestamps are returned alongside their
+    annotations so the two stay aligned.
+
+    Note this makes the series reward-keyed rather than a complete record of the
+    hardware's valve openings: free water delivered on an unrewarded trial is
+    real water the animal received, and it is excluded here.
+
+    Deliveries are matched to trials by the ``Response`` software-event
+    timestamp: each delivery takes the annotation of the trial whose response is
+    closest. The response is used rather than the ``TrialOutcome`` timestamp
+    because ``TrialOutcome`` fires at the *end* of a trial, after the
+    reward-consumption and ITI periods, while the valve opens within
+    milliseconds of the response. Matching on trial end lets a delivery land
+    nearer the *previous* trial's outcome and inherit its
+    ``is_auto_reward_right``, flipping ``earned`` and ``auto``.
+
+    ``response_times`` is aligned to ``trial_outcome_df`` positionally: entry
+    ``i`` is the response of the trial in row ``i``.
 
     Parameters
     ----------
@@ -62,31 +128,46 @@ def get_annotated_rewards(
     manual_water_times : numpy.ndarray
         Software-event timestamps of this port's manual water deliveries
         (``GiveManualWaterLeft`` / ``GiveManualWaterRight``).
+    response_times : numpy.ndarray
+        ``Response`` software-event timestamps, one per trial, positionally
+        aligned with the rows of ``trial_outcome_df``.
 
     Returns
     -------
     numpy.ndarray
-        Array of the same shape as ``reward_delivery_times`` whose entries are
-        ``"earned"``, ``"auto"``, or ``"manual"``.
+        The retained reward-delivery timestamps: ``reward_delivery_times`` less
+        the deliveries on unrewarded trials.
+    numpy.ndarray
+        The matching annotations, one per retained timestamp, each ``"earned"``,
+        ``"auto"``, ``"anti_bias"``, or ``"manual"``.
+
+    Raises
+    ------
+    ValueError
+        If ``response_times`` has a different length than ``trial_outcome_df``,
+        since the two are paired by position.
     """
+    response_times = np.asarray(response_times)
+    if response_times.size != len(trial_outcome_df):
+        raise ValueError(
+            f"response_times has {response_times.size} entries but there are "
+            f"{len(trial_outcome_df)} trials; the two are paired by position."
+        )
+
     reward_times = np.asarray(reward_delivery_times)
     if reward_times.size == 0:
-        return np.array([], dtype=object)
+        return reward_times, np.array([], dtype=object)
 
     # Annotate each delivery from its originating trial: query with reward_times so we
     # get one trial position per reward delivery.
-    trial_indices_in_reward_times = find_closest_timestamps(
-        reward_times, trial_outcome_df.index.to_numpy()
-    )
+    trial_indices_in_reward_times = find_closest_timestamps(reward_times, response_times)
 
     annotated_rewards = []
+    is_unrewarded = []
     for trial_index in trial_indices_in_reward_times:
         outcome = _parse_outcome(trial_outcome_df.iloc[trial_index]["data"])
-        trial = outcome.trial if outcome is not None else None
-        if trial is None or trial.is_auto_reward_right is None:
-            annotated_rewards.append("earned")
-        else:
-            annotated_rewards.append("auto")
+        annotated_rewards.append(_free_water_label(outcome.trial if outcome is not None else None))
+        is_unrewarded.append(outcome is not None and not outcome.is_rewarded)
 
     # Object dtype, not the inferred fixed-width string dtype: a run of only "auto"
     # and "earned" entries would be too narrow to hold "manual" and would truncate it.
@@ -97,8 +178,14 @@ def get_annotated_rewards(
     # manual-water software event to its closest reward delivery; the returned
     # positions index into reward_times, i.e. the deliveries that are manual.
     manual_water_times = np.asarray(manual_water_times)
+    manual_mask = np.zeros(reward_times.size, dtype=bool)
     if manual_water_times.size:
         manual_indices_in_reward_times = find_closest_timestamps(manual_water_times, reward_times)
-        annotated_rewards[manual_indices_in_reward_times] = "manual"
-
-    return annotated_rewards
+        manual_mask[manual_indices_in_reward_times] = True
+        annotated_rewards[manual_mask] = "manual"
+
+    # Downstream analysis is keyed on reward, so a delivery whose trial did not pay out
+    # is excluded. Manual water is experimenter-driven, unrelated to the trial's
+    # outcome, and keeps its delivery.
+    keep = ~(np.array(is_unrewarded, dtype=bool) & ~manual_mask)
+    return reward_times[keep], annotated_rewards[keep]
diff --git a/src/dynamic_foraging_processing/utils/trial_metadata.py b/src/dynamic_foraging_processing/utils/trial_metadata.py
new file mode 100644
index 0000000..87ef4d9
--- /dev/null
+++ b/src/dynamic_foraging_processing/utils/trial_metadata.py
@@ -0,0 +1,42 @@
+"""Helpers for reading a trial's block-based extra metadata."""
+
+from aind_behavior_dynamic_foraging.task_logic.trial_generators.block_based_trial_generator import (
+    BlockBasedTrialMetadata,
+)
+from aind_behavior_dynamic_foraging.task_logic.trial_models import Trial
+
+
+def get_bias_metadata(trial: Trial) -> BlockBasedTrialMetadata:
+    """Return the block-based extra metadata naming a trial's free-water mechanism.
+
+    ``trial.is_auto_reward_right`` is only the delivery *channel*: it says free
+    water was triggered and on which side, not what kind. Scheduled autowater and
+    the anti-bias water intervention are told apart here, by ``is_autowater`` and
+    ``is_bias_water_intervention``. (``is_bias_stage_intervention`` marks the
+    anti-bias algorithm's other lever, moving the lickspouts.)
+
+    The field is schema-typed ``Any``, so it deserializes off the stream as a
+    plain ``dict`` rather than a model; a ``BlockBasedTrialMetadata`` instance is
+    also accepted. When metadata or extra is missing (e.g. a non-block-based
+    generator), the model's all-``False`` default is returned, so a trial whose
+    mechanism the data does not record is reported as neither kind rather than
+    guessed at.
+
+    Parameters
+    ----------
+    trial : Trial
+        The per-trial task-logic model.
+
+    Returns
+    -------
+    BlockBasedTrialMetadata
+        The parsed extra metadata, or an all-``False`` default when absent
+        or unrecognized.
+    """
+    metadata = trial.metadata
+    extra = metadata.extra if metadata is not None else None
+    if isinstance(extra, BlockBasedTrialMetadata):
+        return extra
+    if isinstance(extra, dict):
+        return BlockBasedTrialMetadata.model_validate(extra)
+    return BlockBasedTrialMetadata()
diff --git a/tests/test_nwb/test_acquisition/test_acquisition_builder.py b/tests/test_nwb/test_acquisition/test_acquisition_builder.py
index f16915e..73d6dd3 100644
--- a/tests/test_nwb/test_acquisition/test_acquisition_builder.py
+++ b/tests/test_nwb/test_acquisition/test_acquisition_builder.py
@@ -82,6 +82,14 @@ def _make_trial_outcome_frame() -> pd.DataFrame:
     )
 
 
+def _make_response_frame() -> pd.DataFrame:
+    """One ``Response`` event per trial, just before each trial's outcome."""
+    return pd.DataFrame(
+        {"data": [{"Item1": 0.05, "Item2": False}, {"Item1": 0.35, "Item2": True}]},
+        index=pd.Index([0.05, 0.35], name="time"),
+    )
+
+
 def _empty_manual_water_frame() -> pd.DataFrame:
     """Build an empty manual-water stream with the ``data`` side column."""
     return pd.DataFrame({"data": []}, index=pd.Index([], name="time"))
@@ -120,6 +128,7 @@ def _make_dataset(manual_water=None):
                     "SoftwareEvents": _FakeNode(
                         {
                             "TrialOutcome": _FakeStream(_make_trial_outcome_frame()),
+                            "Response": _FakeStream(_make_response_frame()),
                             "GiveManualWaterRight": _FakeStream(manual_water),
                         }
                     ),
@@ -147,11 +156,11 @@ def test_init_stores_loader():
     assert builder.loader is loader
 
 
-def test_get_reward_delivery_filters_to_write_messages():
+def test_get_valve_writes_filters_to_write_messages():
     """Only ``MessageType == 'WRITE'`` rows are returned."""
     builder = AcquisitionBuilder(loader=_make_loader())
 
-    result = builder.get_reward_delivery()
+    result = builder.get_valve_writes()
 
     assert list(result["MessageType"]) == ["WRITE", "WRITE", "WRITE"]
     assert list(result.index) == [0.1, 0.3, 0.5]
@@ -177,6 +186,7 @@ def test_get_manual_water_times_returns_empty_when_absent():
                     "SoftwareEvents": _FakeNode(
                         {
                             "TrialOutcome": _FakeStream(_make_trial_outcome_frame()),
+                            "Response": _FakeStream(_make_response_frame()),
                         }
                     ),
                 }
@@ -224,6 +234,7 @@ def test_get_lick_times_returns_empty_when_absent():
                     "SoftwareEvents": _FakeNode(
                         {
                             "TrialOutcome": _FakeStream(_make_trial_outcome_frame()),
+                            "Response": _FakeStream(_make_response_frame()),
                             "GiveManualWaterRight": _FakeStream(_empty_manual_water_frame()),
                         }
                     ),
diff --git a/tests/test_processing/test_trial_table.py b/tests/test_processing/test_trial_table.py
index b213ba1..f422b8b 100644
--- a/tests/test_processing/test_trial_table.py
+++ b/tests/test_processing/test_trial_table.py
@@ -690,16 +690,70 @@ def test_rewarded_history_false_on_every_auto_reward_trial():
 
 def test_auto_water_encodes_side_from_auto_response():
     """A non-null auto response encodes ``1`` on its side and ``0`` on the other."""
-    trial = TrialOutcome.model_validate(
+    outcome = TrialOutcome.model_validate(
         _outcome(1.0, 1.0, is_right_choice=True, is_rewarded=True, auto=True)
-    ).trial
-    assert TrialTableBuilder._auto_water(trial, is_right=True) == 1
-    assert TrialTableBuilder._auto_water(trial, is_right=False) == 0
+    )
+    assert TrialTableBuilder._auto_water(outcome.trial, outcome, is_right=True) == 1
+    assert TrialTableBuilder._auto_water(outcome.trial, outcome, is_right=False) == 0
     # No auto-response counts as no autowater (0).
     no_auto = TrialOutcome.model_validate(
         _outcome(1.0, 1.0, is_right_choice=True, is_rewarded=True, auto=None)
-    ).trial
-    assert TrialTableBuilder._auto_water(no_auto, is_right=True) == 0
+    )
+    assert TrialTableBuilder._auto_water(no_auto.trial, no_auto, is_right=True) == 0
+
+
+def test_auto_water_excludes_unrewarded_trials():
+    """Autowater on a trial that did not pay out is ``0``.
+
+    Free water fires at the go cue whether or not the animal's own choice later
+    pays out, so this keeps the column matching the reward-keyed reward-delivery
+    series, which drops those deliveries.
+    """
+    unrewarded = TrialOutcome.model_validate(
+        _outcome(1.0, 1.0, is_right_choice=False, is_rewarded=False, auto=True)
+    )
+    assert TrialTableBuilder._auto_water(unrewarded.trial, unrewarded, is_right=True) == 0
+
+
+def test_auto_water_offered_is_ungated():
+    """``auto_water_offered*`` records every autowater, paid out or not.
+
+    This is the legacy ``dynamic-foraging-task`` meaning of ``auto_water*``, kept
+    so the trial table loses no delivery now that ``auto_water*`` is reward-keyed.
+    """
+    unrewarded = TrialOutcome.model_validate(
+        _outcome(1.0, 1.0, is_right_choice=False, is_rewarded=False, auto=True)
+    )
+    assert TrialTableBuilder._auto_water_offered(unrewarded.trial, is_right=True) == 1
+    assert TrialTableBuilder._auto_water_offered(unrewarded.trial, is_right=False) == 0
+    # The reward-keyed column is 0 on the very same trial.
+    assert TrialTableBuilder._auto_water(unrewarded.trial, unrewarded, is_right=True) == 0
+
+    # No autowater at all is 0 on both sides.
+    no_auto = TrialOutcome.model_validate(
+        _outcome(1.0, 1.0, is_right_choice=True, is_rewarded=True, auto=None)
+    )
+    assert TrialTableBuilder._auto_water_offered(no_auto.trial, is_right=True) == 0
+
+
+def test_auto_water_counts_anti_bias_water():
+    """Anti-bias water is autowater delivered through the auto-response channel.
+
+    The anti-bias intervention shares ``is_auto_reward_right`` with scheduled
+    autowater, so it counts here as well; ``anti_bias_right_water`` marks which
+    of these the algorithm drove.
+    """
+    bias_water = TrialOutcome.model_validate(
+        _outcome(
+            1.0,
+            1.0,
+            is_right_choice=True,
+            is_rewarded=True,
+            auto=True,
+            is_bias_water_intervention=True,
+        )
+    )
+    assert TrialTableBuilder._auto_water(bias_water.trial, bias_water, is_right=True) == 1
 
 
 def test_bias_metadata_parses_dict_model_and_default():
@@ -750,18 +804,42 @@ def test_anti_bias_water_gated_on_intervention_flag_and_side():
             auto=True,
             is_bias_water_intervention=True,
         )
-    ).trial
-    meta = TrialTableBuilder._bias_metadata(right)
-    assert TrialTableBuilder._anti_bias_water(right, meta, is_right=True) is True
-    assert TrialTableBuilder._anti_bias_water(right, meta, is_right=False) is False
+    )
+    meta = TrialTableBuilder._bias_metadata(right.trial)
+    assert TrialTableBuilder._anti_bias_water(right.trial, meta, is_right=True) is True
+    assert TrialTableBuilder._anti_bias_water(right.trial, meta, is_right=False) is False
 
     # Auto-response to the left without the bias flag is ordinary autowater, not
     # an anti-bias intervention.
     autowater = TrialOutcome.model_validate(
         _outcome(1.0, 1.0, is_right_choice=False, is_rewarded=True, auto=False)
-    ).trial
-    auto_meta = TrialTableBuilder._bias_metadata(autowater)
-    assert TrialTableBuilder._anti_bias_water(autowater, auto_meta, is_right=False) is False
+    )
+    auto_meta = TrialTableBuilder._bias_metadata(autowater.trial)
+    assert TrialTableBuilder._anti_bias_water(autowater.trial, auto_meta, is_right=False) is False
+
+
+def test_anti_bias_water_includes_unrewarded_trials():
+    """An intervention on a trial that did not pay out still counts.
+
+    The column records what the anti-bias algorithm did, and the intervention
+    fires at the go cue regardless of how the animal's own choice resolves. This
+    is deliberately *not* gated on ``is_rewarded``, so it is not a subset of
+    ``auto_waterL``/``auto_waterR``.
+    """
+    unrewarded = TrialOutcome.model_validate(
+        _outcome(
+            1.0,
+            1.0,
+            is_right_choice=False,
+            is_rewarded=False,
+            auto=True,
+            is_bias_water_intervention=True,
+        )
+    )
+    meta = TrialTableBuilder._bias_metadata(unrewarded.trial)
+    assert TrialTableBuilder._anti_bias_water(unrewarded.trial, meta, is_right=True) is True
+    # The same trial contributes no autowater, which *is* reward-gated.
+    assert TrialTableBuilder._auto_water(unrewarded.trial, unrewarded, is_right=True) == 0
 
 
 def test_anti_bias_lickspout_movement_gated_on_stage_flag():
diff --git a/tests/test_utils/test_rewards.py b/tests/test_utils/test_rewards.py
index 2be8269..545d6c6 100644
--- a/tests/test_utils/test_rewards.py
+++ b/tests/test_utils/test_rewards.py
@@ -4,12 +4,13 @@
 
 import numpy as np
 import pandas as pd
+import pytest
 from aind_behavior_dynamic_foraging.task_logic.trial_models import TrialOutcome
 
-from dynamic_foraging_processing.utils.rewards import get_annotated_rewards
+from dynamic_foraging_processing.utils.rewards import get_reward_deliveries
 
 
-def _outcome_payload(auto=None) -> dict:
+def _outcome_payload(auto=None, is_rewarded: bool = True) -> dict:
     """Return a serialized ``TrialOutcome`` payload with the given auto-response."""
     return {
         "trial": {
@@ -22,103 +23,226 @@ def _outcome_payload(auto=None) -> dict:
             "is_auto_reward_right": auto,
         },
         "is_right_choice": True,
-        "is_rewarded": True,
+        "is_rewarded": is_rewarded,
     }
 
 
-def _trial_outcome_df(trial_times: np.ndarray, autos=None) -> pd.DataFrame:
+def _trial_outcome_df(trial_times: np.ndarray, autos=None, rewarded=None) -> pd.DataFrame:
     """Build a trial outcome DataFrame with one row per entry of ``trial_times``."""
     autos = autos if autos is not None else [None] * len(trial_times)
+    rewarded = rewarded if rewarded is not None else [True] * len(trial_times)
     return pd.DataFrame(
-        {"data": [_outcome_payload(auto) for auto in autos]},
+        {"data": [_outcome_payload(a, r) for a, r in zip(autos, rewarded)]},
         index=pd.Index(trial_times, name="time"),
     )
 
 
-def test_get_annotated_rewards_marks_default_trials_as_earned():
+def test_get_reward_deliveries_marks_default_trials_as_earned():
     """Trials with no auto-response setting and no manual water are ``earned``."""
     reward_times = np.array([0.15, 0.42, 0.95])
-    trial_outcome_df = _trial_outcome_df(np.array([0.1, 0.4, 0.9]))
+    response_times = np.array([0.1, 0.4, 0.9])
+    trial_outcome_df = _trial_outcome_df(np.array([1.1, 1.4, 1.9]))
 
-    annotations = get_annotated_rewards(reward_times, trial_outcome_df, np.array([]))
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, np.array([]), response_times
+    )
 
+    np.testing.assert_array_equal(times, reward_times)
     np.testing.assert_array_equal(annotations, np.array(["earned", "earned", "earned"]))
 
 
-def test_get_annotated_rewards_marks_auto_response_trials_as_auto():
+def test_get_reward_deliveries_marks_auto_response_trials_as_auto():
     """Trials with ``is_auto_reward_right`` set (either side) are ``auto``."""
     reward_times = np.array([0.15, 0.42])
-    trial_outcome_df = _trial_outcome_df(np.array([0.1, 0.4]), autos=[True, False])
+    response_times = np.array([0.1, 0.4])
+    trial_outcome_df = _trial_outcome_df(np.array([1.1, 1.4]), autos=[True, False])
 
-    annotations = get_annotated_rewards(reward_times, trial_outcome_df, np.array([]))
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, np.array([]), response_times
+    )
 
+    np.testing.assert_array_equal(times, reward_times)
     np.testing.assert_array_equal(annotations, np.array(["auto", "auto"]))
 
 
-def test_get_annotated_rewards_matches_closest_trial_outcome_time():
-    """Each delivery takes the annotation of the closest ``TrialOutcome`` event."""
-    # Both deliveries sit nearest the second (auto) trial, so both are auto even
-    # though the first trial is earned.
+def test_get_reward_deliveries_marks_anti_bias_water_as_auto():
+    """Anti-bias water is autowater, so it annotates as ``auto``.
+
+    The anti-bias intervention is delivered through the same
+    ``is_auto_reward_right`` channel as scheduled autowater, so the annotation
+    does not distinguish them; ``anti_bias_left_water``/``anti_bias_right_water``
+    in the trials table mark which deliveries the algorithm drove.
+    """
+    reward_times = np.array([0.15])
+    response_times = np.array([0.1])
+    payload = _outcome_payload(True)
+    payload["trial"]["metadata"] = {"extra": {"is_bias_water_intervention": True}}
+    trial_outcome_df = pd.DataFrame({"data": [payload]}, index=pd.Index([1.1], name="time"))
+
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, np.array([]), response_times
+    )
+
+    np.testing.assert_array_equal(times, reward_times)
+    np.testing.assert_array_equal(annotations, np.array(["auto"]))
+
+
+def test_get_reward_deliveries_matches_closest_response_time():
+    """Each delivery takes the annotation of the trial whose response is closest.
+
+    The trial-outcome timestamps deliberately disagree with the response times:
+    matching on the outcome would pick the first (earned) trial, so this pins the
+    match to the ``Response`` stream.
+    """
     reward_times = np.array([0.95, 1.05])
-    trial_outcome_df = _trial_outcome_df(np.array([0.1, 1.0]), autos=[None, True])
+    response_times = np.array([0.1, 1.0])
+    # Outcome events fire at the end of each trial, far from the deliveries.
+    trial_outcome_df = _trial_outcome_df(np.array([0.9, 5.0]), autos=[None, True])
 
-    annotations = get_annotated_rewards(reward_times, trial_outcome_df, np.array([]))
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, np.array([]), response_times
+    )
 
+    np.testing.assert_array_equal(times, reward_times)
     np.testing.assert_array_equal(annotations, np.array(["auto", "auto"]))
 
 
-def test_get_annotated_rewards_marks_manual_water_as_manual():
+def test_get_reward_deliveries_drops_auto_water_on_unrewarded_trials():
+    """Autowater on a trial reporting ``is_rewarded=False`` is dropped, not annotated.
+
+    The water is delivered at the go cue and the trial then continues normally,
+    so a trial whose own choice did not pay out still carries the delivery. The
+    series is reward-keyed, so those deliveries are excluded.
+    """
+    reward_times = np.array([0.15, 0.42, 0.95])
+    response_times = np.array([0.1, 0.4, 0.9])
+    trial_outcome_df = _trial_outcome_df(
+        np.array([1.1, 1.4, 1.9]),
+        autos=[None, True, True],
+        rewarded=[True, False, True],
+    )
+
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, np.array([]), response_times
+    )
+
+    np.testing.assert_array_equal(times, np.array([0.15, 0.95]))
+    np.testing.assert_array_equal(annotations, np.array(["earned", "auto"]))
+
+
+def test_get_reward_deliveries_drops_any_delivery_on_an_unrewarded_trial():
+    """The drop rule is ``is_rewarded=False``, not autowater specifically.
+
+    Autowater is the only case seen in practice, but the condition is the trial's
+    reward outcome, so any delivery on a trial that did not pay out is excluded
+    regardless of what triggered it.
+    """
+    reward_times = np.array([0.15])
+    response_times = np.array([0.1])
+    trial_outcome_df = _trial_outcome_df(np.array([1.1]), autos=[None], rewarded=[False])
+
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, np.array([]), response_times
+    )
+
+    assert times.size == 0
+    assert annotations.size == 0
+
+
+def test_get_reward_deliveries_marks_manual_water_as_manual():
     """Deliveries closest to a manual-water event are annotated as ``manual``."""
     reward_times = np.array([0.15, 0.42, 0.95])
-    trial_outcome_df = _trial_outcome_df(np.array([0.1, 0.4, 0.9]))
+    response_times = np.array([0.1, 0.4, 0.9])
+    trial_outcome_df = _trial_outcome_df(np.array([1.1, 1.4, 1.9]))
     # Software event near the second delivery (0.42).
     manual_water_times = np.array([0.43])
 
-    annotations = get_annotated_rewards(reward_times, trial_outcome_df, manual_water_times)
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, manual_water_times, response_times
+    )
 
+    np.testing.assert_array_equal(times, reward_times)
     np.testing.assert_array_equal(annotations, np.array(["earned", "manual", "earned"]))
 
 
-def test_get_annotated_rewards_manual_takes_precedence_over_auto():
+def test_get_reward_deliveries_manual_takes_precedence_over_auto():
     """A manual delivery is ``manual`` even when the trial has auto-response set."""
     reward_times = np.array([0.15, 0.42])
-    trial_outcome_df = _trial_outcome_df(np.array([0.1, 0.4]), autos=[None, True])
+    response_times = np.array([0.1, 0.4])
+    trial_outcome_df = _trial_outcome_df(np.array([1.1, 1.4]), autos=[None, True])
     manual_water_times = np.array([0.42])
 
-    annotations = get_annotated_rewards(reward_times, trial_outcome_df, manual_water_times)
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, manual_water_times, response_times
+    )
 
+    np.testing.assert_array_equal(times, reward_times)
     np.testing.assert_array_equal(annotations, np.array(["earned", "manual"]))
 
 
-def test_get_annotated_rewards_empty_deliveries_returns_empty():
-    """No reward deliveries yields an empty annotation array."""
+def test_get_reward_deliveries_manual_water_survives_the_auto_drop():
+    """Manual water on an unrewarded auto trial is kept, not dropped."""
+    reward_times = np.array([0.15, 0.42])
+    response_times = np.array([0.1, 0.4])
+    trial_outcome_df = _trial_outcome_df(
+        np.array([1.1, 1.4]), autos=[None, True], rewarded=[True, False]
+    )
+    manual_water_times = np.array([0.42])
+
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, manual_water_times, response_times
+    )
+
+    np.testing.assert_array_equal(times, reward_times)
+    np.testing.assert_array_equal(annotations, np.array(["earned", "manual"]))
+
+
+def test_get_reward_deliveries_empty_deliveries_returns_empty():
+    """No reward deliveries yields empty timestamp and annotation arrays."""
     trial_outcome_df = _trial_outcome_df(np.array([0.0]))
 
-    result = get_annotated_rewards(np.array([]), trial_outcome_df, np.array([]))
+    times, annotations = get_reward_deliveries(
+        np.array([]), trial_outcome_df, np.array([]), np.array([0.0])
+    )
 
-    assert isinstance(result, np.ndarray)
-    assert result.size == 0
+    assert isinstance(annotations, np.ndarray)
+    assert times.size == 0
+    assert annotations.size == 0
 
 
-def test_get_annotated_rewards_accepts_json_and_model_payloads():
+def test_get_reward_deliveries_accepts_json_and_model_payloads():
     """``data`` payloads may be JSON strings or already-parsed ``TrialOutcome``."""
     reward_times = np.array([0.15, 0.42])
+    response_times = np.array([0.1, 0.4])
     payload = _outcome_payload(True)
     trial_outcome_df = pd.DataFrame(
         {"data": [json.dumps(payload), TrialOutcome.model_validate(payload)]},
-        index=pd.Index([0.1, 0.4], name="time"),
+        index=pd.Index([1.1, 1.4], name="time"),
     )
 
-    annotations = get_annotated_rewards(reward_times, trial_outcome_df, np.array([]))
+    times, annotations = get_reward_deliveries(
+        reward_times, trial_outcome_df, np.array([]), response_times
+    )
 
+    np.testing.assert_array_equal(times, reward_times)
     np.testing.assert_array_equal(annotations, np.array(["auto", "auto"]))
 
 
-def test_get_annotated_rewards_returns_ndarray():
-    """The return value is a ``numpy.ndarray``."""
-    reward_times = np.array([0.1])
-    trial_outcome_df = _trial_outcome_df(np.array([0.0]))
+def test_get_reward_deliveries_rejects_misaligned_response_times():
+    """``response_times`` must have one entry per trial; they pair by position."""
+    trial_outcome_df = _trial_outcome_df(np.array([1.1, 1.4]))
+
+    with pytest.raises(ValueError, match="paired by position"):
+        get_reward_deliveries(np.array([0.15]), trial_outcome_df, np.array([]), np.array([0.1]))
+
 
-    result = get_annotated_rewards(reward_times, trial_outcome_df, np.array([]))
+def test_get_reward_deliveries_returns_ndarray():
+    """Both return values are :class:`numpy.ndarray`."""
+    trial_outcome_df = _trial_outcome_df(np.array([1.0]))
+
+    times, annotations = get_reward_deliveries(
+        np.array([0.1]), trial_outcome_df, np.array([]), np.array([0.0])
+    )
 
-    assert isinstance(result, np.ndarray)
+    assert isinstance(times, np.ndarray)
+    assert isinstance(annotations, np.ndarray)