From ce890cd06749206b7a5afe97c8eb61b894127225 Mon Sep 17 00:00:00 2001
From: Logiquo <fanyongda2012@gmail.com>
Date: Fri, 30 Oct 2026 11:13:15 +0000
Subject: [PATCH] Move the YAIB tasks and per-step labels to
 SequenceProcessorV2
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- TimeseriesLabelProcessor and its binary and regression subclasses are now
  restricted SequenceProcessorV2 fields: one series level with scalar
  values and strictly increasing times. A batch is packed as (values [ΣN, 1],
  time [ΣN], batch offsets [B + 1]); there is no grid or mask.
- The timeseries losses average over the label records, and
  Trainer.inference splits the packed labels into one array per sample.
  _resolve_mode picks the most specific registered processor class, so the
  label subclasses resolve to their own modes.
- RNN and MultimodalRNN reject per-step (timeseries_*) labels.
- The five YAIB MIMIC-IV tasks use a sequence_v2 series for the dynamic
  features (forward fill, then mean, then zero imputation, z-score
  normalization and missing indicators) and per-hour label records.
- Tests use a minimal per-step model (tests/core/label_record_model_utils.py)
  for the loss, Trainer and metrics.
---
 ...cessors.TimeseriesBinaryLabelProcessor.rst |   2 +-
 ...th.processors.TimeseriesLabelProcessor.rst |   2 +-
 ...ors.TimeseriesRegressionLabelProcessor.rst |   2 +-
 pyhealth/models/base_model.py                 |  80 ++++---
 pyhealth/models/rnn.py                        |  54 ++---
 pyhealth/processors/timeseries_processor.py   | 132 +++++-------
 pyhealth/tasks/yaib_mimic4.py                 | 103 +++++----
 pyhealth/trainer.py                           |  20 +-
 tests/core/label_record_model_utils.py        |  32 +++
 tests/core/test_per_step_labels.py            | 115 ++++++----
 tests/core/test_timeseries_modes.py           | 201 ++++++++----------
 tests/core/test_yaib_mimic4.py                | 103 +++++++--
 12 files changed, 452 insertions(+), 394 deletions(-)
 create mode 100644 tests/core/label_record_model_utils.py

diff --git a/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst
index 378971fb3..578bdca3e 100644
--- a/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst
+++ b/docs/api/processors/pyhealth.processors.TimeseriesBinaryLabelProcessor.rst
@@ -1,7 +1,7 @@
 pyhealth.processors.TimeseriesBinaryLabelProcessor
 ==================================================
 
-Binary (0/1) label per time step (``"timeseries_binary"``).
+Binary (0/1) labels at time points (``"timeseries_binary"``).
 
 .. autoclass:: pyhealth.processors.TimeseriesBinaryLabelProcessor
     :members:
diff --git a/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst
index 009a0850c..b1d302ef2 100644
--- a/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst
+++ b/docs/api/processors/pyhealth.processors.TimeseriesLabelProcessor.rst
@@ -1,7 +1,7 @@
 pyhealth.processors.TimeseriesLabelProcessor
 ============================================
 
-Base processor for labels given per time step, returning ``(value, mask)`` on the time grid of the input time series.
+Base processor for labels given at time points: a restricted ``sequence_v2`` field with one ``series`` level of scalar values, packed per batch.
 
 .. autoclass:: pyhealth.processors.TimeseriesLabelProcessor
     :members:
diff --git a/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst b/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst
index 0aa0cd59a..30ace8a1e 100644
--- a/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst
+++ b/docs/api/processors/pyhealth.processors.TimeseriesRegressionLabelProcessor.rst
@@ -1,7 +1,7 @@
 pyhealth.processors.TimeseriesRegressionLabelProcessor
 ======================================================
 
-Real-valued label per time step (``"timeseries_regression"``).
+Real-valued labels at time points (``"timeseries_regression"``).
 
 .. autoclass:: pyhealth.processors.TimeseriesRegressionLabelProcessor
     :members:
diff --git a/pyhealth/models/base_model.py b/pyhealth/models/base_model.py
index 40ca7a007..8c76693d8 100644
--- a/pyhealth/models/base_model.py
+++ b/pyhealth/models/base_model.py
@@ -27,37 +27,28 @@
 )
 
 
-def _masked_step_loss(
+def _label_record_loss(
     loss_fn: Callable,
     logits: torch.Tensor,
-    target: tuple[torch.Tensor, torch.Tensor],
+    target: tuple[torch.Tensor, ...],
 ) -> torch.Tensor:
-    """Averages a per-step loss over the time steps that have a label.
+    """Averages a loss over the label records of a batch.
 
     Args:
-        loss_fn: Elementwise loss accepting ``reduction="none"``.
-        logits: Per-step logits of shape (batch, steps, 1).
-        target: Batched ``(value, mask)`` of shape (batch, label steps), where
-            padding and unlabeled steps have mask 0. Values at those steps are
-            ignored and may be NaN.
+        loss_fn: The loss, e.g. ``F.binary_cross_entropy_with_logits``.
+        logits: One prediction per label record, ``[ΣN, 1]``.
+        target: The packed batch of a
+            :class:`~pyhealth.processors.TimeseriesLabelProcessor`:
+            ``(values [ΣN, 1], time [ΣN], batch offsets [B + 1])``.
     """
-    # Temporary workaround: per-step labels arrive as a (value, mask) tuple so
-    # the existing single-tensor label API stays unchanged. A more unified
-    # handling of padded/masked label tensors across models is a follow-up.
-    value, mask = target
-    if value.shape[1] > logits.shape[1]:
+    values = target[0]
+    if logits.shape != values.shape:
         raise ValueError(
-            f"The label has {value.shape[1]} time steps but the input only "
-            f"{logits.shape[1]}; label and input must share the time grid."
+            f"Expected one prediction per label record, {tuple(values.shape)}, "
+            f"got {tuple(logits.shape)}. Per-step labels need a model that "
+            "predicts at each label's time."
         )
-    logits = logits[:, : value.shape[1], 0]
-    labeled = mask > 0
-    # Replace unlabeled values before the loss so a NaN there reaches neither
-    # the loss nor its gradient, then drop those steps from the sum.
-    value = torch.where(labeled, value, torch.zeros_like(value))
-    loss = loss_fn(logits, value, reduction="none")
-    loss = torch.where(labeled, loss, torch.zeros_like(loss))
-    return loss.sum() / labeled.sum().clamp(min=1)
+    return loss_fn(logits, values)
 
 
 class BaseModel(ABC, nn.Module):
@@ -193,13 +184,14 @@ def _resolve_mode(self, schema_entry: Any) -> str:
         if isinstance(schema_entry, str):
             return schema_entry.lower()
 
-        # Get class reference
+        # The most specific registered class: a subclass of a registered
+        # processor (e.g. the timeseries labels, which are SequenceProcessorV2s)
+        # resolves to its own name before its base's.
         cls = schema_entry if inspect.isclass(schema_entry) else schema_entry.__class__
-        for name, registered_cls in PROCESSOR_REGISTRY.items():
-            if cls is registered_cls or issubclass(
-                cls, registered_cls
-            ):  # allow subclassing
-                return name.lower()
+        for klass in cls.__mro__:
+            for name, registered_cls in PROCESSOR_REGISTRY.items():
+                if klass is registered_cls:
+                    return name.lower()
         raise ValueError(
             f"Cannot resolve mode from output_schema entry {schema_entry}. Use a supported string"
         )
@@ -313,16 +305,14 @@ def get_loss_function(self) -> Callable:
             - multiclass: `F.cross_entropy`
             - multilabel: `F.binary_cross_entropy_with_logits`
             - regression: `F.mse_loss`
-            - timeseries_binary: `F.binary_cross_entropy_with_logits` per
-              time step, averaged over the labeled steps
-            - timeseries_regression: `F.mse_loss` per time step, averaged
-              over the labeled steps
+            - timeseries_binary: `F.binary_cross_entropy_with_logits` over
+              the label records
+            - timeseries_regression: `F.mse_loss` over the label records
 
-        The timeseries losses take per-step logits of shape (batch, steps, 1)
-        and the batched ``(value, mask)`` label, each of shape (batch, steps),
-        produced by an output processor whose ``process()`` returns
-        ``(value, mask)`` per sample with mask 0 for steps without a label.
-        Values at masked steps are ignored and may be NaN.
+        The timeseries losses take one logit per label record, ``[ΣN, 1]``,
+        and the packed batch of a
+        :class:`~pyhealth.processors.TimeseriesLabelProcessor`, ``(values
+        [ΣN, 1], time [ΣN], batch offsets [B + 1])``.
 
         For binary and multilabel labels, a weight set with
         :meth:`set_pos_weight` is passed as ``pos_weight``.
@@ -353,9 +343,11 @@ def weighted_bce(input, target, **kwargs):
         elif mode == "regression":
             return F.mse_loss
         elif mode == "timeseries_binary":
-            return functools.partial(_masked_step_loss, F.binary_cross_entropy_with_logits)
+            return functools.partial(
+                _label_record_loss, F.binary_cross_entropy_with_logits
+            )
         elif mode == "timeseries_regression":
-            return functools.partial(_masked_step_loss, F.mse_loss)
+            return functools.partial(_label_record_loss, F.mse_loss)
         else:
             raise ValueError(f"Invalid mode: {mode}")
 
@@ -373,10 +365,10 @@ def prepare_y_prob(self, logits: torch.Tensor) -> torch.Tensor:
             - multilabel: a tensor of shape (batch_size, num_labels) with values
                 in [0, 1], which is obtained with `torch.sigmoid()`
             - regression: a tensor of shape (batch_size, 1) with raw logits
-            - timeseries_binary: a tensor of shape (batch_size, steps, 1)
-                with values in [0, 1] per time step, from `torch.sigmoid()`
-            - timeseries_regression: a tensor of shape (batch_size, steps, 1)
-                with raw logits per time step
+            - timeseries_binary: a tensor of shape (label records, 1) with
+                values in [0, 1], from `torch.sigmoid()`
+            - timeseries_regression: a tensor of shape (label records, 1)
+                with raw logits
 
         Args:
             logits (torch.Tensor): The predicted logit tensor.
diff --git a/pyhealth/models/rnn.py b/pyhealth/models/rnn.py
index 8e728b622..dc2c0a0b5 100644
--- a/pyhealth/models/rnn.py
+++ b/pyhealth/models/rnn.py
@@ -1,4 +1,4 @@
-from typing import Dict, Optional, Tuple
+from typing import Optional
 
 import torch
 import torch.nn as nn
@@ -21,6 +21,9 @@
 
 from .embedding import EmbeddingModel
 
+# Label modes with one prediction per label record (TimeseriesLabelProcessor).
+_PER_STEP_MODES = ("timeseries_binary", "timeseries_regression")
+
 
 class RNNLayer(nn.Module):
     """Recurrent neural network layer.
@@ -86,7 +89,7 @@ def forward(
         self,
         x: torch.Tensor,
         mask: Optional[torch.Tensor] = None,
-    ) -> Tuple[torch.Tensor, torch.Tensor]:
+    ) -> tuple[torch.Tensor, torch.Tensor]:
         """Forward propagation.
 
         Args:
@@ -225,12 +228,9 @@ def __init__(
             raise ValueError("hidden_size is determined by hidden_dim")
         assert len(self.label_keys) == 1, "Only one label key is supported if RNN is initialized"
         self.label_key = self.label_keys[0]
-        # Per-step (timeseries) labels predict at every input step.
-        self._per_step = self.mode in ("timeseries_binary", "timeseries_regression")
-        if self._per_step and kwargs.get("bidirectional", False):
+        if self.mode in _PER_STEP_MODES:
             raise ValueError(
-                "A per-step label needs a causal model; bidirectional=True would "
-                "use future time steps."
+                "RNN does not support per-step (timeseries_*) labels."
             )
 
         self.embedding_model = EmbeddingModel(dataset, embedding_dim)
@@ -243,7 +243,7 @@ def __init__(
         output_size = self.get_output_size()
         self.fc = nn.Linear(len(self.feature_keys) * self.hidden_dim, output_size)
 
-    def forward(self, **kwargs) -> Dict[str, torch.Tensor]:
+    def forward(self, **kwargs) -> dict[str, torch.Tensor]:
         """Forward propagation.
 
         The label `kwargs[self.label_key]` is a list of labels for each patient.
@@ -253,7 +253,7 @@ def forward(self, **kwargs) -> Dict[str, torch.Tensor]:
                 all the feature keys and the label key.
 
         Returns:
-            Dict[str, torch.Tensor]: A dictionary with the following keys:
+            dict[str, torch.Tensor]: A dictionary with the following keys:
                 - loss: a scalar tensor representing the loss.
                 - y_prob: a tensor representing the predicted probabilities.
                 - y_true: a tensor representing the true labels.
@@ -309,30 +309,14 @@ def forward(self, **kwargs) -> Dict[str, torch.Tensor]:
                 else:
                     mask = (torch.abs(x).sum(dim=-1) != 0).int()
 
-            outputs, x = self.rnn[feature_key](x, mask)
-            # Per-step labels use the output at every step, else the last one.
-            patient_emb.append(outputs if self._per_step else x)
-
-        if self._per_step:
-            # Features without a time axis (one step) are repeated over time.
-            steps = max(emb.shape[1] for emb in patient_emb)
-            for i, emb in enumerate(patient_emb):
-                if emb.shape[1] == 1:
-                    patient_emb[i] = emb.expand(-1, steps, -1)
-                elif emb.shape[1] != steps:
-                    raise ValueError(
-                        "Per-step labels need all time series features on the "
-                        "same time grid."
-                    )
-        patient_emb = torch.cat(patient_emb, dim=-1)
-        # (patient, label_size), or (patient, steps, label_size) per step
+            _, x = self.rnn[feature_key](x, mask)
+            patient_emb.append(x)
+
+        patient_emb = torch.cat(patient_emb, dim=1)
+        # (patient, label_size)
         logits = self.fc(patient_emb)
         # obtain y_true, loss, y_prob
-        if self._per_step:
-            # (value, mask) per step; the loss averages over labeled steps.
-            y_true = tuple(t.to(self.device) for t in kwargs[self.label_key])
-        else:
-            y_true = kwargs[self.label_key].to(self.device)
+        y_true = kwargs[self.label_key].to(self.device)
         loss = self.get_loss_function()(logits, y_true)
         y_prob = self.prepare_y_prob(logits)
         results = {"loss": loss, "y_prob": y_prob, "y_true": y_true, "logit": logits}
@@ -443,6 +427,10 @@ def __init__(
 
         assert len(self.label_keys) == 1, "Only one label key is supported"
         self.label_key = self.label_keys[0]
+        if self.mode in _PER_STEP_MODES:
+            raise ValueError(
+                "MultimodalRNN does not support per-step (timeseries_*) labels."
+            )
 
         self.embedding_model = EmbeddingModel(dataset, embedding_dim)
 
@@ -496,7 +484,7 @@ def _is_sequential_processor(self, processor) -> bool:
             TimeseriesProcessor,
         ))
 
-    def forward(self, **kwargs) -> Dict[str, torch.Tensor]:
+    def forward(self, **kwargs) -> dict[str, torch.Tensor]:
         """Forward propagation handling mixed modalities.
 
         The label `kwargs[self.label_key]` is a list of labels for each patient.
@@ -506,7 +494,7 @@ def forward(self, **kwargs) -> Dict[str, torch.Tensor]:
                 all the feature keys and the label key.
 
         Returns:
-            Dict[str, torch.Tensor]: A dictionary with the following keys:
+            dict[str, torch.Tensor]: A dictionary with the following keys:
                 - loss: a scalar tensor representing the loss.
                 - y_prob: a tensor representing the predicted probabilities.
                 - y_true: a tensor representing the true labels.
diff --git a/pyhealth/processors/timeseries_processor.py b/pyhealth/processors/timeseries_processor.py
index ef5a34ddd..c38e49cd9 100644
--- a/pyhealth/processors/timeseries_processor.py
+++ b/pyhealth/processors/timeseries_processor.py
@@ -6,11 +6,11 @@
 
 from . import register_processor
 from .base_processor import FeatureProcessor
+from .sequence_processor_v2 import Records, SequenceProcessorV2
 from .utils import (
     NORMALIZATIONS,
     Normalization,
     fit_normalization,
-    grid_steps,
     sample_on_grid,
 )
 
@@ -191,24 +191,23 @@ def __repr__(self):
         )
 
 
-class TimeseriesLabelProcessor(FeatureProcessor):
-    """Base processor for labels given per time step.
+class TimeseriesLabelProcessor(SequenceProcessorV2):
+    """Base processor for labels given at time points (a restricted V2 field).
 
-    The label is a time series ``(timestamps, values)``, placed on the same
-    kind of uniform grid as :class:`TimeseriesProcessor`: a label at time
-    ``t`` goes to step ``floor((t - t0) / sampling_rate)``, where ``t0`` is the
-    first label timestamp. Steps that no label falls into, or whose label is
-    NaN, are masked out and never imputed. Two labels in the same step raise
-    a ``ValueError`` rather than one silently replacing the other.
+    A :class:`SequenceProcessorV2` with one fixed ``series`` level of scalar
+    values, so a label is a :class:`Records`: ``{"time": [...], "values":
+    [[v], ...]}`` (a flat ``[v, ...]`` also works). Times are in the unit of
+    the inputs' top level and must be strictly increasing: two labels at one
+    time raise. A missing label is left out rather than given as NaN, which
+    raises. There is no grid, resampling, imputation or mask.
 
-    The processor does not see the input time series, so the caller must give
-    the labels the same first timestamp and ``sampling_rate`` as the input for
-    each label to line up with one input step. A label series longer than its
-    input is rejected by the model's loss.
+    A model predicts one value per label, at its time, from the input records
+    at or before it.
 
     Output:
-        - ``(value, mask)``: two float tensors of shape (S,), where ``mask``
-          is 1 for steps with a label. Padding added when batching has mask 0.
+        - ``(values, time)``: ``[N, 1]`` values and ``[N]`` times. A batch
+          (:meth:`collate`) is packed: ``[ΣN, 1]`` values, ``[ΣN]`` times
+          and the batch offsets ``[B + 1]``.
 
     The registered names ``"timeseries_binary"`` and
     ``"timeseries_regression"`` are model modes of their own, so modules that
@@ -216,97 +215,64 @@ class TimeseriesLabelProcessor(FeatureProcessor):
     interpretability) do not treat these labels as ``"binary"`` or
     ``"regression"``.
 
-    Args:
-        sampling_rate: Interval of the sampled time steps.
-
     Examples:
-        >>> from datetime import datetime, timedelta
-        >>> from pyhealth.processors import TimeseriesBinaryLabelProcessor
-        >>> t0 = datetime(2020, 1, 1)
-        >>> times = [t0, t0 + timedelta(hours=2)]
-        >>> TimeseriesBinaryLabelProcessor().process((times, [0, 1]))
-        (tensor([0., 0., 1.]), tensor([1., 0., 1.]))
+        >>> from pyhealth.processors import TimeseriesRegressionLabelProcessor
+        >>> TimeseriesRegressionLabelProcessor().process({"time": [0.0, 3.0], "values": [2.5, 4.0]})
+        (tensor([[2.5000],
+                [4.0000]]), tensor([0., 3.]))
     """
 
-    def __init__(self, sampling_rate: timedelta = timedelta(hours=1)):
-        self.sampling_rate = sampling_rate
-
-    def process(
-        self, value: tuple[list[datetime], np.ndarray]
-    ) -> tuple[torch.Tensor, torch.Tensor]:
-        timestamps, labels = value
-        if len(timestamps) == 0:
-            raise ValueError("Timestamps list is empty.")
-        steps, counts = np.unique(
-            grid_steps(timestamps, self.sampling_rate), return_counts=True
-        )
-        if (counts > 1).any():
-            step = int(steps[counts > 1][0])
-            raise ValueError(
-                f"{int(counts.max())} labels fall into time step {step}; "
-                "give at most one label per step of the sampling rate."
-            )
-        sampled = sample_on_grid(timestamps, labels, self.sampling_rate)
-        if sampled.shape[1] != 1:
-            raise ValueError("Expected one label per time step.")
-        sampled = sampled[:, 0]
-        mask = ~np.isnan(sampled)
-        self._check(sampled[mask])
-        return (
-            torch.tensor(np.where(mask, sampled, 0.0), dtype=torch.float),
-            torch.tensor(mask, dtype=torch.float),
-        )
-
-    def _check(self, labels: np.ndarray) -> None:
+    def __init__(self) -> None:
+        super().__init__(levels=[{"ordering": "series", "values": [1]}])
+        # No tokens and no statistics: there is nothing to fit.
+        self._fitted = True
+
+    def process(self, value: Records) -> tuple[torch.Tensor, ...]:
+        if isinstance(value, dict) and "time" in value:
+            time = np.asarray(value["time"], dtype=np.float64)
+            if (np.diff(time) <= 0).any():
+                raise ValueError(
+                    "Label times must be strictly increasing: give at most one "
+                    "label per time."
+                )
+        output = super().process(value)
+        self._check(output[0])
+        return output
+
+    def _check(self, values: torch.Tensor) -> None:
         pass
 
     def size(self) -> int:
+        """One output per label."""
         return 1
 
-    def is_token(self) -> bool:
-        return False
-
-    def schema(self) -> tuple[str, ...]:
-        return ("value", "mask")
-
-    def dim(self) -> tuple[int, ...]:
-        return (1, 1)
-
-    def spatial(self) -> tuple[bool, ...]:
-        return (True,)
-
-    def __repr__(self):
-        return f"{type(self).__name__}(sampling_rate={self.sampling_rate})"
+    def __repr__(self) -> str:
+        return f"{type(self).__name__}()"
 
 
 @register_processor("timeseries_binary")
 class TimeseriesBinaryLabelProcessor(TimeseriesLabelProcessor):
-    """Binary (0/1) label per time step, see :class:`TimeseriesLabelProcessor`.
+    """Binary (0/1) labels at time points, see :class:`TimeseriesLabelProcessor`.
 
     Examples:
-        >>> from datetime import datetime, timedelta
         >>> from pyhealth.processors import TimeseriesBinaryLabelProcessor
-        >>> t0 = datetime(2020, 1, 1)
-        >>> times = [t0 + timedelta(hours=h) for h in range(3)]
-        >>> TimeseriesBinaryLabelProcessor().process((times, [0, float("nan"), 1]))
-        (tensor([0., 0., 1.]), tensor([1., 0., 1.]))
+        >>> TimeseriesBinaryLabelProcessor().process({"time": [0.0, 1.0], "values": [0, 1]})
+        (tensor([[0.],
+                [1.]]), tensor([0., 1.]))
     """
 
-    def _check(self, labels: np.ndarray) -> None:
-        if not np.isin(labels, (0.0, 1.0)).all():
+    def _check(self, values: torch.Tensor) -> None:
+        if not torch.isin(values, torch.tensor([0.0, 1.0])).all():
             raise ValueError("Binary labels must be 0 or 1.")
 
 
 @register_processor("timeseries_regression")
 class TimeseriesRegressionLabelProcessor(TimeseriesLabelProcessor):
-    """Real-valued label per time step, see :class:`TimeseriesLabelProcessor`.
+    """Real-valued labels at time points, see :class:`TimeseriesLabelProcessor`.
 
     Examples:
-        >>> from datetime import datetime, timedelta
         >>> from pyhealth.processors import TimeseriesRegressionLabelProcessor
-        >>> t0 = datetime(2020, 1, 1)
-        >>> times = [t0 + timedelta(hours=h) for h in range(3)]
-        >>> TimeseriesRegressionLabelProcessor().process((times, [2.5, float("nan"), 0.5]))
-        (tensor([2.5000, 0.0000, 0.5000]), tensor([1., 0., 1.]))
+        >>> TimeseriesRegressionLabelProcessor().process({"time": [0.0, 2.0], "values": [2.5, 0.5]})
+        (tensor([[2.5000],
+                [0.5000]]), tensor([0., 2.]))
     """
-
diff --git a/pyhealth/tasks/yaib_mimic4.py b/pyhealth/tasks/yaib_mimic4.py
index 749e867f1..d5d830ae0 100644
--- a/pyhealth/tasks/yaib_mimic4.py
+++ b/pyhealth/tasks/yaib_mimic4.py
@@ -223,17 +223,21 @@ class _YAIBMIMIC4Task(BaseTask):
     measurement (or the last stay, for measurements after all stays), which is
     how ricu maps admission-level data onto ICU stays.
 
-    Each sample holds ``dynamic`` (hourly timestamps from ICU admission and a
-    ``[T, 48]`` array with ``NaN`` for bins without measurement) and
+    Each sample holds ``dynamic``, a ``sequence_v2`` series of the hourly
+    bins (``{"time": [0, 1, ..., T], "values": [T + 1, 48]}``, time in hours
+    since ICU admission, ``NaN`` for bins without measurement), and
     ``static`` (``[age, sex, height, weight]``, sex encoded as 1 for male and
-    0 for female, ``NaN`` when unavailable). Missing values are left to the
-    processors, which apply YAIB's preprocessing: the dynamic features are
-    standardized, get missing-value indicators and are forward-filled, with
-    values before the first measurement set to 0 (the training mean); the
-    static features get missing-value indicators, are standardized and
-    zero-filled. The statistics are learned when the processors are fitted,
-    so use ``set_task(task, split=...)`` to learn them on the training
-    patients only. Unlike YAIB, which label-encodes sex as a category, sex is
+    0 for female, ``NaN`` when unavailable). Per-hour labels are
+    ``timeseries_*`` labels on the same hours (``{"time": [...], "values":
+    [...]}``), so a model predicts at hour ``h`` from the bins up to ``h``.
+    Missing values are left to the processors, which apply YAIB's
+    preprocessing: the dynamic features are forward-filled, with values
+    before the first measurement set to the training mean (0 once
+    standardized), standardized, and get a missing-value mask; the static
+    features get missing-value indicators, are standardized and zero-filled.
+    The statistics are learned when the processors are fitted, so use
+    ``set_task(task, split=...)`` to learn them on the training patients
+    only. Unlike YAIB, which label-encodes sex as a category, sex is
     standardized like the other static features.
 
     Note on ``o2sat`` (deliberate deviation from YAIB): YAIB pins ricu 0.6.0,
@@ -256,9 +260,27 @@ class _YAIBMIMIC4Task(BaseTask):
     """
 
     input_schema: ClassVar[dict[str, Any]] = {
+        # Forward fill, then the training mean (0 once standardized) before
+        # the first measurement, then 0 for a feature never seen in training
+        # (normalization leaves it as is): YAIB's preprocessing, with the
+        # missing-value indicators as the mask.
         "dynamic": (
-            "timeseries",
-            {"normalization": "zscore", "missing_indicators": True},
+            "sequence_v2",
+            {
+                "levels": [
+                    {
+                        "ordering": "series",
+                        "values": [len(_DYNAMIC_FEATURES)],
+                        "impute": [
+                            {"method": "forward_fill"},
+                            {"method": "mean", "by_dims": ()},  # per feature
+                            {"method": "zero"},
+                        ],
+                        "normalize": {"method": "zscore"},
+                        "missing_indicator": True,
+                    }
+                ]
+            },
         ),
         "static": (
             "tensor",
@@ -492,10 +514,8 @@ def _base_cohort(
         hourly = hourly.filter(pl.col("stay_id").is_in(keep))
         return stays, hourly, {stay_id: static[stay_id] for stay_id in keep}
 
-    def _dynamic(
-        self, hourly: pl.DataFrame, stay_id: int, intime: datetime, end: int
-    ) -> tuple[list[datetime], np.ndarray]:
-        """Builds the ``[end + 1, 48]`` hourly grid for bins ``0..end``."""
+    def _dynamic(self, hourly: pl.DataFrame, stay_id: int, end: int) -> dict[str, Any]:
+        """Builds the series of hourly bins ``0..end``: ``[end + 1, 48]`` values."""
         values = np.full((end + 1, len(self.DYNAMIC_FEATURES)), np.nan)
         rows = hourly.filter(
             (pl.col("stay_id") == stay_id)
@@ -505,8 +525,12 @@ def _dynamic(
         values[rows["hour"].to_numpy(), rows["feature"].to_numpy()] = (
             rows["value"].to_numpy()
         )
-        timestamps = [intime + timedelta(hours=hour) for hour in range(end + 1)]
-        return timestamps, values
+        return {"time": [float(hour) for hour in range(end + 1)], "values": values}
+
+    @staticmethod
+    def _labels(time: list[float], labels: list[float]) -> dict[str, Any]:
+        """A per-hour label series on the hours of ``dynamic``."""
+        return {"time": time, "values": [[label] for label in labels]}
 
     def _icu_deaths(self, patient: Any, stays: pl.DataFrame) -> dict[int, datetime]:
         """Returns the ICU death time per ICU stay.
@@ -682,7 +706,7 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
                     "patient_id": patient.patient_id,
                     "stay_id": stay_id,
                     "dynamic": self._dynamic(
-                        hourly, stay_id, intime, min(end, self.OBSERVATION_HOURS)
+                        hourly, stay_id, min(end, self.OBSERVATION_HOURS)
                     ),
                     "static": static[stay_id],
                     "mortality": int(death is not None),
@@ -737,8 +761,8 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
         )
 
         samples = []
-        for stay_id, intime, los_hours, end in stays.select(
-            "stay_id", "intime", "los_hours", "end"
+        for stay_id, los_hours, end in stays.select(
+            "stay_id", "los_hours", "end"
         ).iter_rows():
             if los_hours < self.MIN_STAY_HOURS or stay_id not in labels:
                 continue
@@ -747,7 +771,7 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
                     "patient_id": patient.patient_id,
                     "stay_id": stay_id,
                     "dynamic": self._dynamic(
-                        hourly, stay_id, intime, min(end, self.OBSERVATION_HOURS)
+                        hourly, stay_id, min(end, self.OBSERVATION_HOURS)
                     ),
                     "static": static[stay_id],
                     "creatinine": labels[stay_id],
@@ -766,7 +790,7 @@ class YAIBLengthOfStayMIMIC4(_YAIBMIMIC4Task):
     ``min(168, end - h)``. The cohort is the YAIB base cohort without further
     exclusions.
 
-    The label is a per-hour time series on the same grid as ``dynamic``
+    The label is a per-hour series on the hours of ``dynamic``
     (``timeseries_regression``), so models predict at every hour.
 
     Requires the ``labevents``, ``chartevents`` and ``outputevents`` tables.
@@ -793,20 +817,19 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
         stays, hourly, static = self._base_cohort(patient, all_stays)
 
         samples = []
-        for stay_id, intime, end in stays.select("stay_id", "intime", "end").iter_rows():
-            timestamps, values = self._dynamic(
-                hourly, stay_id, intime, min(end, self.MAX_HOURS)
-            )
+        for stay_id, end in stays.select("stay_id", "end").iter_rows():
+            dynamic = self._dynamic(hourly, stay_id, min(end, self.MAX_HOURS))
             remaining = [
-                float(min(self.MAX_HOURS, end - hour)) for hour in range(len(timestamps))
+                float(min(self.MAX_HOURS, end - hour))
+                for hour in range(len(dynamic["time"]))
             ]
             samples.append(
                 {
                     "patient_id": patient.patient_id,
                     "stay_id": stay_id,
-                    "dynamic": (timestamps, values),
+                    "dynamic": dynamic,
                     "static": static[stay_id],
-                    "los": (timestamps, remaining),
+                    "los": self._labels(dynamic["time"], remaining),
                 }
             )
         return samples
@@ -930,7 +953,7 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
         stays, hourly, static = self._base_cohort(patient, all_stays)
 
         samples = []
-        for stay_id, intime, end in stays.select("stay_id", "intime", "end").iter_rows():
+        for stay_id, end in stays.select("stay_id", "end").iter_rows():
             onset, baseline = self._aki_onset(hourly, stay_id, static[stay_id][3])
             if onset is not None and onset < self.MIN_ONSET_HOURS:
                 continue
@@ -939,17 +962,17 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
             last = min(end, self.MAX_HOURS)
             if onset is not None:
                 last = min(last, onset + self.HORIZON)
-            timestamps, values = self._dynamic(hourly, stay_id, intime, last)
+            dynamic = self._dynamic(hourly, stay_id, last)
             labels = self._outcome_window(
-                onset, len(timestamps), self.HORIZON, self.HORIZON
+                onset, len(dynamic["time"]), self.HORIZON, self.HORIZON
             )
             samples.append(
                 {
                     "patient_id": patient.patient_id,
                     "stay_id": stay_id,
-                    "dynamic": (timestamps, values),
+                    "dynamic": dynamic,
                     "static": static[stay_id],
-                    "aki": (timestamps, labels),
+                    "aki": self._labels(dynamic["time"], labels),
                 }
             )
         return samples
@@ -1554,7 +1577,7 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
         sample_hours = self._sample_hours(patient, all_stays)
 
         samples = []
-        for stay_id, intime, end in stays.select("stay_id", "intime", "end").iter_rows():
+        for stay_id, end in stays.select("stay_id", "end").iter_rows():
             sofa_hours, sofa = self._sofa(
                 inputs.get(stay_id, {}),
                 vasopressors.get(stay_id, {}),
@@ -1570,17 +1593,17 @@ def __call__(self, patient: Any) -> list[dict[str, Any]]:
             last = min(end, self.MAX_HOURS)
             if onset is not None:
                 last = min(last, onset + self.HORIZON)
-            timestamps, values = self._dynamic(hourly, stay_id, intime, last)
+            dynamic = self._dynamic(hourly, stay_id, last)
             labels = self._outcome_window(
-                onset, len(timestamps), self.HORIZON, self.HORIZON
+                onset, len(dynamic["time"]), self.HORIZON, self.HORIZON
             )
             samples.append(
                 {
                     "patient_id": patient.patient_id,
                     "stay_id": stay_id,
-                    "dynamic": (timestamps, values),
+                    "dynamic": dynamic,
                     "static": static[stay_id],
-                    "sepsis": (timestamps, labels),
+                    "sepsis": self._labels(dynamic["time"], labels),
                 }
             )
         return samples
diff --git a/pyhealth/trainer.py b/pyhealth/trainer.py
index 3726e2792..8329c37ff 100644
--- a/pyhealth/trainer.py
+++ b/pyhealth/trainer.py
@@ -2,6 +2,7 @@
 import logging
 import os
 from datetime import datetime
+from itertools import pairwise
 from typing import Callable, Dict, List, Optional, Type
 
 import numpy as np
@@ -358,7 +359,7 @@ def inference(self, dataloader, additional_outputs=None,
             y_true_all: List of true labels.
             y_prob_all: List of predicted probabilities.
                 For the per-step ``timeseries_*`` modes, both are lists with
-                one array per sample, over its labeled time steps.
+                one array per sample, over its label records.
             loss_mean: Mean loss over batches.
             additional_outputs (only if requested): Dict of additional results.
             patient_ids (only if requested): List of patient ids in the same order as y_true_all/y_prob_all.
@@ -377,16 +378,13 @@ def inference(self, dataloader, additional_outputs=None,
                 loss = output["loss"]
                 loss_all.append(loss.item())
                 if per_step:
-                    # Temporary workaround to avoid breaking the existing API:
-                    # samples differ in length, so keep each sample's labeled
-                    # steps as a separate array. A more unified handling of
-                    # padded/masked outputs is a follow-up.
-                    value, mask = (t.cpu().numpy() for t in output["y_true"])
-                    y_prob = output["y_prob"].cpu().numpy()[:, : value.shape[1], 0]
-                    for i in range(len(value)):
-                        labeled = mask[i] > 0
-                        y_true_all.append(value[i][labeled])
-                        y_prob_all.append(y_prob[i][labeled])
+                    # Packed label records (values, time, batch offsets): one
+                    # array per sample, as the timeseries metrics expect.
+                    values, _, offsets = (t.cpu().numpy() for t in output["y_true"])
+                    y_prob = output["y_prob"].cpu().numpy()[:, 0]
+                    for lo, hi in pairwise(offsets):
+                        y_true_all.append(values[lo:hi, 0])
+                        y_prob_all.append(y_prob[lo:hi])
                 else:
                     y_true_all.append(output["y_true"].cpu().numpy())
                     y_prob_all.append(output["y_prob"].cpu().numpy())
diff --git a/tests/core/label_record_model_utils.py b/tests/core/label_record_model_utils.py
new file mode 100644
index 000000000..f7fb0ead7
--- /dev/null
+++ b/tests/core/label_record_model_utils.py
@@ -0,0 +1,32 @@
+"""A minimal per-step model for tests of the timeseries label plumbing."""
+
+from typing import Any
+
+from torch import nn
+
+from pyhealth.datasets import SampleDataset
+from pyhealth.models import BaseModel
+
+
+class LabelRecordModel(BaseModel):
+    """Predicts each label record from its time with one linear layer.
+
+    It ignores the input features, so tests of the label processors, the loss,
+    the Trainer and the metrics do not depend on a particular architecture.
+    """
+
+    def __init__(self, dataset: SampleDataset):
+        super().__init__(dataset)
+        self.label_key = self.label_keys[0]
+        self.fc = nn.Linear(1, 1)
+
+    def forward(self, **batch: Any) -> dict[str, Any]:
+        # Packed label records: (values [ΣN, 1], time [ΣN], offsets [B + 1]).
+        label = tuple(t.to(self.device) for t in batch[self.label_key])
+        logits = self.fc(label[1].unsqueeze(1))
+        return {
+            "loss": self.get_loss_function()(logits, label),
+            "y_prob": self.prepare_y_prob(logits),
+            "y_true": label,
+            "logit": logits,
+        }
diff --git a/tests/core/test_per_step_labels.py b/tests/core/test_per_step_labels.py
index 4268be3b0..feafd2da3 100644
--- a/tests/core/test_per_step_labels.py
+++ b/tests/core/test_per_step_labels.py
@@ -1,35 +1,35 @@
 import unittest
-from datetime import datetime, timedelta
+from typing import Any
 
 import numpy as np
 import torch
 
 from pyhealth.datasets import create_sample_dataset, get_dataloader
-from pyhealth.models import RNN
 from pyhealth.processors import (
+    LevelConfig,
+    SequenceProcessorV2,
     TimeseriesBinaryLabelProcessor,
+    TimeseriesLabelProcessor,
     TimeseriesRegressionLabelProcessor,
 )
 from pyhealth.trainer import Trainer
+from tests.core.label_record_model_utils import LabelRecordModel
 
-T0 = datetime(2020, 1, 1)
 NAN = float("nan")
+DYNAMIC: list[LevelConfig] = [{"ordering": "series", "values": [2]}]
 
 
-def _times(n):
-    return [T0 + timedelta(hours=h) for h in range(n)]
-
-
-def _samples(label_fn):
+def _samples(label_fn) -> list[dict[str, Any]]:
     rng = np.random.default_rng(0)
     samples = []
     for i, n in enumerate([5, 3, 6, 4]):
+        hours = [float(h) for h in range(n)]
         samples.append(
             {
                 "patient_id": f"p{i}",
-                "dynamic": (_times(n), rng.normal(size=(n, 2))),
+                "dynamic": {"time": hours, "values": rng.normal(size=(n, 2))},
                 "static": [float(i), 1.0],
-                "label": (_times(n), [label_fn(i, h) for h in range(n)]),
+                "label": {"time": hours, "values": [label_fn(i, h) for h in range(n)]},
             }
         )
     return samples
@@ -38,47 +38,78 @@ def _samples(label_fn):
 def _dataset(output="timeseries_binary", label_fn=lambda i, h: float((i + h) % 2)):
     return create_sample_dataset(
         samples=_samples(label_fn),
-        input_schema={"dynamic": "timeseries", "static": "tensor"},
+        input_schema={
+            "dynamic": ("sequence_v2", {"levels": DYNAMIC}),
+            "static": "tensor",
+        },
         output_schema={"label": output},
         dataset_name="per_step",
     )
 
 
 class TestTimeseriesLabelProcessors(unittest.TestCase):
-    def test_value_and_mask(self):
-        value, mask = TimeseriesBinaryLabelProcessor().process(
-            (_times(4), [0, NAN, 1, 1])
+    def test_are_restricted_sequence_v2_fields(self):
+        processor = TimeseriesBinaryLabelProcessor()
+        self.assertIsInstance(processor, SequenceProcessorV2)
+        self.assertIsInstance(processor, TimeseriesLabelProcessor)
+        self.assertEqual(processor.schema(), ((0, "values"), (0, "time")))
+        self.assertEqual(processor.size(), 1)
+        self.assertFalse(processor.learns_statistics)
+        self.assertEqual(repr(processor), "TimeseriesBinaryLabelProcessor()")
+
+    def test_values_and_time(self):
+        values, time = TimeseriesRegressionLabelProcessor().process(
+            {"time": [0.0, 3.0, 4.5], "values": [[2.5], [4.0], [1.0]]}
+        )
+        self.assertEqual(values.tolist(), [[2.5], [4.0], [1.0]])
+        self.assertEqual(time.tolist(), [0.0, 3.0, 4.5])
+
+    def test_needs_no_fit(self):
+        self.assertEqual(
+            TimeseriesBinaryLabelProcessor()
+            .process({"time": [1.0], "values": [1]})[0]
+            .tolist(),
+            [[1.0]],
         )
-        np.testing.assert_array_equal(value.numpy(), [0, 0, 1, 1])
-        np.testing.assert_array_equal(mask.numpy(), [1, 0, 1, 1])
-
-    def test_grid_follows_sampling_rate(self):
-        times = [T0, T0 + timedelta(hours=3)]
-        value, mask = TimeseriesRegressionLabelProcessor().process((times, [2.5, 4.0]))
-        np.testing.assert_array_equal(value.numpy(), [2.5, 0, 0, 4.0])
-        np.testing.assert_array_equal(mask.numpy(), [1, 0, 0, 1])
-
-    def test_binary_rejects_other_values(self):
-        with self.assertRaises(ValueError):
-            TimeseriesBinaryLabelProcessor().process((_times(2), [0, 2]))
-
-    def test_timestamps_between_steps_are_floored(self):
-        times = [T0, T0 + timedelta(minutes=90)]
-        value, mask = TimeseriesRegressionLabelProcessor().process((times, [1.0, 2.0]))
-        np.testing.assert_array_equal(value.numpy(), [1.0, 2.0])
-        np.testing.assert_array_equal(mask.numpy(), [1, 1])
 
-    def test_two_labels_in_one_step_are_rejected(self):
-        times = [T0, T0 + timedelta(minutes=20), T0 + timedelta(hours=1)]
-        with self.assertRaisesRegex(ValueError, "time step 0"):
-            TimeseriesBinaryLabelProcessor().process((times, [0, 1, 0]))
+    def test_rejects(self):
+        bad: list[tuple[Any, Any]] = [
+            (TimeseriesBinaryLabelProcessor(), {"time": [0.0, 1.0], "values": [0, 2]}),
+            (
+                TimeseriesBinaryLabelProcessor(),
+                {"time": [0.0, 0.0], "values": [0, 1]},
+            ),  # one time twice
+            (
+                TimeseriesRegressionLabelProcessor(),
+                {"time": [1.0, 0.0], "values": [1.0, 2.0]},
+            ),
+            (
+                TimeseriesRegressionLabelProcessor(),
+                {"time": [0.0, 1.0], "values": [1.0, NAN]},
+            ),
+            (TimeseriesRegressionLabelProcessor(), {"values": [1.0]}),  # no time
+            (
+                TimeseriesRegressionLabelProcessor(),
+                {"time": [0.0], "values": [[1.0, 2.0]]},
+            ),
+        ]
+        for processor, records in bad:
+            with self.subTest(records=records), self.assertRaises(ValueError):
+                processor.process(records)
+
+    def test_batches_are_packed(self):
+        dataset = _dataset()
+        batch = next(iter(get_dataloader(dataset, batch_size=4)))
+        values, time, offsets = batch["label"]
+        self.assertEqual(values.shape, (18, 1))
+        self.assertEqual(time.shape, (18,))
+        self.assertEqual(offsets.tolist(), [0, 5, 8, 14, 18])
 
 
 class TestPerStepEndToEnd(unittest.TestCase):
-    """The label processors with a model and the Trainer.
+    """The label processors with a per-step model and the Trainer.
 
-    Model-level behaviour (masked loss, causality, bidirectional guard) is
-    covered with a mock label processor in test_timeseries_modes.py.
+    The loss over label records is covered in test_timeseries_modes.py.
     """
 
     def test_label_processors_train_and_evaluate(self):
@@ -87,18 +118,20 @@ def test_label_processors_train_and_evaluate(self):
             ("timeseries_regression", "mae"),
         ):
             dataset = _dataset(mode)
-            model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8)
+            model = LabelRecordModel(dataset)
             self.assertEqual(model.mode, mode)
             batch = next(iter(get_dataloader(dataset, batch_size=4)))
             out = model(**batch)
-            self.assertEqual(tuple(out["y_prob"].shape), (4, 6, 1))
+            self.assertEqual(tuple(out["y_prob"].shape), (18, 1))
             self.assertTrue(torch.isfinite(out["loss"]))
 
             trainer = Trainer(model=model, metrics=[metric], enable_logging=False)
             loader = get_dataloader(dataset, batch_size=3)
+            trainer.train(train_dataloader=loader, epochs=1)
             y_true, _, _ = trainer.inference(loader)
             self.assertEqual(sorted(len(y) for y in y_true), [3, 4, 5, 6])
             self.assertIn(metric, trainer.evaluate(loader))
 
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/core/test_timeseries_modes.py b/tests/core/test_timeseries_modes.py
index bb82858ca..c9d9b340a 100644
--- a/tests/core/test_timeseries_modes.py
+++ b/tests/core/test_timeseries_modes.py
@@ -9,157 +9,123 @@
     timeseries_binary_metrics_fn,
     timeseries_regression_metrics_fn,
 )
-from pyhealth.models import RNN
-from pyhealth.processors.base_processor import FeatureProcessor
+from pyhealth.models import RNN, MultimodalRNN
+from pyhealth.processors import (
+    LevelConfig,
+    TimeseriesBinaryLabelProcessor,
+    TimeseriesLabelProcessor,
+    TimeseriesRegressionLabelProcessor,
+)
 from pyhealth.trainer import Trainer
+from tests.core.label_record_model_utils import LabelRecordModel
 
-T0 = datetime(2020, 1, 1)
-
-
-class _StepLabelProcessor(FeatureProcessor):
-    """Minimal per-step label processor: ``(values, mask)`` lists to tensors."""
-
-    def process(self, value):
-        values, mask = value
-        return (
-            torch.tensor(values, dtype=torch.float),
-            torch.tensor(mask, dtype=torch.float),
-        )
-
-    def size(self):
-        return 1
-
-    def is_token(self):
-        return False
+DYNAMIC: list[LevelConfig] = [{"ordering": "series", "values": [2]}]
 
-    def schema(self):
-        return ("value", "mask")
 
-    def dim(self):
-        return (1, 1)
-
-    def spatial(self):
-        return (True,)
-
-
-def _dataset(mode="timeseries_binary", unlabeled=(), unlabeled_value=None):
+def _dataset(
+    mode: str | type[TimeseriesLabelProcessor] = "timeseries_binary",
+    labeled=lambda i, h: True,
+):
+    """Hourly dynamic series; labels at the hours ``labeled`` keeps."""
     rng = np.random.default_rng(0)
     samples = []
     for i, steps in enumerate([5, 3, 6, 4]):
-        values = [float((i + h) % 2) for h in range(steps)]
-        mask = [0.0 if (i, h) in unlabeled else 1.0 for h in range(steps)]
-        if unlabeled_value is not None:
-            values = [unlabeled_value if m == 0 else v for v, m in zip(values, mask)]
+        hours = [float(h) for h in range(steps)]
+        kept = [h for h in range(steps) if labeled(i, h)]
         samples.append(
             {
                 "patient_id": f"p{i}",
-                "dynamic": (
-                    [T0 + timedelta(hours=h) for h in range(steps)],
-                    rng.normal(size=(steps, 2)),
-                ),
+                "dynamic": {"time": hours, "values": rng.normal(size=(steps, 2))},
                 "static": [float(i), 1.0],
-                "label": (values, mask),
+                "label": {
+                    "time": [float(h) for h in kept],
+                    "values": [float((i + h) % 2) for h in kept],
+                },
             }
         )
     return create_sample_dataset(
         samples=samples,
-        input_schema={"dynamic": "timeseries", "static": "tensor"},
+        input_schema={
+            "dynamic": ("sequence_v2", {"levels": DYNAMIC}),
+            "static": "tensor",
+        },
         output_schema={"label": mode},
-        output_processors={"label": _StepLabelProcessor()},
         dataset_name="timeseries_modes",
     )
 
 
+def _model(dataset) -> LabelRecordModel:
+    model = LabelRecordModel(dataset)
+    model.eval()
+    return model
+
+
 class TestTimeseriesModes(unittest.TestCase):
     def test_modes_are_separate_from_binary_and_regression(self):
         for mode in ("timeseries_binary", "timeseries_regression"):
-            self.assertEqual(RNN(dataset=_dataset(mode)).mode, mode)
-
-    def test_loss_ignores_unlabeled_and_padded_steps(self):
-        dataset = _dataset(unlabeled={(1, 0)})
-        model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8).eval()
+            self.assertEqual(_model(_dataset(mode)).mode, mode)
+
+    def test_label_processor_class_resolves_to_its_own_mode(self):
+        # The label processors subclass SequenceProcessorV2; the most specific
+        # registered class decides the mode.
+        for processor, mode in (
+            (TimeseriesBinaryLabelProcessor, "timeseries_binary"),
+            (TimeseriesRegressionLabelProcessor, "timeseries_regression"),
+        ):
+            with self.subTest(mode=mode):
+                self.assertEqual(_model(_dataset(processor)).mode, mode)
+
+    def test_one_prediction_per_label_record(self):
+        # Sample 1 has no label at hour 0: it is left out, not masked.
+        dataset = _dataset(labeled=lambda i, h: (i, h) != (1, 0))
+        model = _model(dataset)
         batch = next(iter(get_dataloader(dataset, batch_size=4)))
         with torch.no_grad():
             out = model(**batch)
-        value, mask = out["y_true"]
-        self.assertEqual(tuple(out["y_prob"].shape), (4, 6, 1))
-        self.assertEqual(int(mask.sum()), 17)  # 5 + 3 + 6 + 4 steps, minus one
-
-        logits = out["logit"].clone()
-        logits[mask == 0] = 1e3
-        loss = model.get_loss_function()(logits, (value, mask))
+        values, time, offsets = out["y_true"]
+        self.assertEqual(tuple(out["logit"].shape), (17, 1))  # 5 + 2 + 6 + 4
+        self.assertEqual(offsets.tolist(), [0, 5, 7, 13, 17])
+        self.assertEqual(time[5:7].tolist(), [1.0, 2.0])  # sample 1 from hour 1
+        loss = torch.nn.functional.binary_cross_entropy_with_logits(
+            out["logit"], values
+        )
         torch.testing.assert_close(loss, out["loss"])
 
-    def test_predictions_do_not_see_future_steps(self):
-        dataset = _dataset()
-        model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8).eval()
-        batch = next(iter(get_dataloader(dataset, batch_size=1)))
-        with torch.no_grad():
-            before = model(**batch)["y_prob"][0, :, 0]
-            batch["dynamic"] = batch["dynamic"].clone()
-            batch["dynamic"][:, -1] += 100.0
-            after = model(**batch)["y_prob"][0, :, 0]
-        torch.testing.assert_close(before[:-1], after[:-1])
-        self.assertFalse(torch.allclose(before[-1], after[-1]))
-
-    def test_nan_on_unlabeled_step_is_ignored(self):
-        for mode in ("timeseries_binary", "timeseries_regression"):
-            nan = _dataset(mode, unlabeled={(1, 0)}, unlabeled_value=float("nan"))
-            zero = _dataset(mode, unlabeled={(1, 0)}, unlabeled_value=0.0)
-            torch.manual_seed(0)
-            model = RNN(dataset=nan, hidden_dim=8, embedding_dim=8)
-            batch = next(iter(get_dataloader(nan, batch_size=4)))
-            out = model(**batch)
-            self.assertTrue(torch.isfinite(out["loss"]))
-            out["loss"].backward()
-            for parameter in model.parameters():
-                if parameter.grad is not None:
-                    self.assertTrue(torch.isfinite(parameter.grad).all())
-
-            # Same loss as with any other value on the unlabeled step.
-            loss_fn = model.get_loss_function()
-            logits = out["logit"].detach()
-            zero_batch = next(iter(get_dataloader(zero, batch_size=4)))
-            torch.testing.assert_close(
-                loss_fn(logits, batch["label"]), loss_fn(logits, zero_batch["label"])
-            )
-
-    def test_label_longer_than_input_is_rejected(self):
-        model = RNN(dataset=_dataset(), hidden_dim=8, embedding_dim=8)
-        logits = torch.zeros(1, 3, 1)
-        label = (torch.zeros(1, 4), torch.ones(1, 4))
-        with self.assertRaises(ValueError):
-            model.get_loss_function()(logits, label)
-
-    def test_time_series_inputs_must_share_the_time_grid(self):
+    def test_loss_needs_one_prediction_per_label(self):
+        model = _model(_dataset())
+        label = (torch.zeros(4, 1), torch.arange(4.0), torch.tensor([0, 4]))
+        with self.assertRaisesRegex(ValueError, "one prediction per label record"):
+            model.get_loss_function()(torch.zeros(1, 3, 1), label)
+
+    def test_v1_rnns_reject_per_step_labels(self):
+        rng = np.random.default_rng(0)
+        t0 = datetime(2020, 1, 1)
         samples = [
             {
-                "patient_id": "p0",
-                "a": ([T0 + timedelta(hours=h) for h in range(3)], np.zeros((3, 1))),
-                "b": ([T0 + timedelta(hours=h) for h in range(5)], np.zeros((5, 1))),
-                "label": ([0.0] * 3, [1.0] * 3),
+                "patient_id": f"p{i}",
+                "dynamic": (
+                    [t0 + timedelta(hours=h) for h in range(3)],
+                    rng.normal(size=(3, 2)),
+                ),
+                "label": {"time": [0.0, 1.0, 2.0], "values": [0.0, 1.0, float(i)]},
             }
+            for i in range(2)
         ]
         dataset = create_sample_dataset(
-            samples=samples,
-            input_schema={"a": "timeseries", "b": "timeseries"},
-            output_schema={"label": "timeseries_binary"},
-            output_processors={"label": _StepLabelProcessor()},
-            dataset_name="timeseries_grids",
+            samples, {"dynamic": "timeseries"}, {"label": "timeseries_binary"}
         )
-        model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8)
-        batch = next(iter(get_dataloader(dataset, batch_size=1)))
-        with self.assertRaises(ValueError):
-            model(**batch)
-
-    def test_bidirectional_is_rejected(self):
-        with self.assertRaises(ValueError):
-            RNN(dataset=_dataset(), bidirectional=True)
-
-    def test_trainer_scores_labeled_steps_per_sample(self):
+        for model in (RNN, MultimodalRNN):
+            with (
+                self.subTest(model=model.__name__),
+                self.assertRaisesRegex(ValueError, "does not support per-step"),
+            ):
+                model(dataset=dataset)
+
+    def test_trainer_scores_label_records_per_sample(self):
         dataset = _dataset()
-        model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8)
-        trainer = Trainer(model=model, metrics=["roc_auc"], enable_logging=False)
+        trainer = Trainer(
+            model=_model(dataset), metrics=["roc_auc"], enable_logging=False
+        )
         loader = get_dataloader(dataset, batch_size=3)
         y_true, y_prob, _ = trainer.inference(loader)
         self.assertEqual(sorted(len(y) for y in y_true), [3, 4, 5, 6])
@@ -167,8 +133,7 @@ def test_trainer_scores_labeled_steps_per_sample(self):
         self.assertIn("roc_auc", trainer.evaluate(loader))
 
         dataset = _dataset("timeseries_regression")
-        model = RNN(dataset=dataset, hidden_dim=8, embedding_dim=8)
-        trainer = Trainer(model=model, metrics=["mae"], enable_logging=False)
+        trainer = Trainer(model=_model(dataset), metrics=["mae"], enable_logging=False)
         self.assertIn("mae", trainer.evaluate(get_dataloader(dataset, batch_size=3)))
 
 
diff --git a/tests/core/test_yaib_mimic4.py b/tests/core/test_yaib_mimic4.py
index 8cddb5ae2..06454c95a 100644
--- a/tests/core/test_yaib_mimic4.py
+++ b/tests/core/test_yaib_mimic4.py
@@ -3,11 +3,15 @@
 
 import numpy as np
 import polars as pl
+import torch
 
 from pyhealth.data import Patient
+from pyhealth.datasets import create_sample_dataset, get_dataloader
+from pyhealth.processors import LevelConfig, SequenceProcessorV2, TimeseriesProcessor
 from pyhealth.tasks import (
     YAIBAKIMIMIC4,
     YAIBKidneyFunctionMIMIC4,
+    YAIBLengthOfStayMIMIC4,
     YAIBMortalityMIMIC4,
     YAIBSepsisMIMIC4,
 )
@@ -20,6 +24,11 @@
 CREA = YAIBMortalityMIMIC4.DYNAMIC_FEATURES.index("crea")
 
 
+def _labels(records):
+    """The label values of a ``timeseries_*`` label series."""
+    return [value[0] for value in records["values"]]
+
+
 def _event(event_type, timestamp, **attrs):
     row = {"event_type": event_type, "timestamp": timestamp}
     for key, value in attrs.items():
@@ -131,10 +140,10 @@ def test_survivor_features(self):
         self.assertEqual(sample["stay_id"], 10)
         self.assertEqual(sample["mortality"], 0)
 
-        timestamps, values = sample["dynamic"]
+        values = sample["dynamic"]["values"]
         self.assertEqual(values.shape, (25, len(self.task.DYNAMIC_FEATURES)))
-        self.assertEqual(timestamps[0], intime)
-        self.assertEqual(timestamps[-1], intime + timedelta(hours=24))
+        # Hours since ICU admission.
+        self.assertEqual(sample["dynamic"]["time"], [float(h) for h in range(25)])
         np.testing.assert_allclose(values[:, HR], 80 + np.arange(25))
         self.assertAlmostEqual(values[3, TEMP], 37.0)
         self.assertEqual(values[4, URINE], 150)
@@ -244,7 +253,7 @@ def test_labs_are_attributed_to_stays_of_the_admission(self):
             extra=extra,
         )
         crea = {
-            s["stay_id"]: np.nonzero(np.isfinite(s["dynamic"][1][:, CREA]))[0]
+            s["stay_id"]: np.nonzero(np.isfinite(s["dynamic"]["values"][:, CREA]))[0]
             for s in self.task(patient)
         }
         np.testing.assert_array_equal(crea[10], [2])
@@ -263,7 +272,7 @@ def test_o2sat_includes_arterial_saturation(self):
             admissions=[_admission()],
             extra=extra,
         )
-        _, values = self.task(patient)[0]["dynamic"]
+        values = self.task(patient)[0]["dynamic"]["values"]
         o2sat = values[:, YAIBMortalityMIMIC4.DYNAMIC_FEATURES.index("o2sat")]
         self.assertEqual(o2sat[1], 90)
         self.assertEqual(o2sat[2], 94)
@@ -318,7 +327,7 @@ def test_label_is_median_of_hourly_medians_in_hours_25_to_48(self):
         samples = self.task(patient)
         self.assertEqual(len(samples), 1)
         self.assertEqual(samples[0]["creatinine"], 4.0)
-        _, values = samples[0]["dynamic"]
+        values = samples[0]["dynamic"]["values"]
         self.assertEqual(values.shape[0], 25)
         self.assertEqual(values[10, CREA], 9.0)
 
@@ -349,6 +358,29 @@ def _weight(value, stay_id=10):
                   hadm_id=1, itemid=226512, valuenum=value, valueuom="kg")
 
 
+class TestYAIBLengthOfStayMIMIC4(unittest.TestCase):
+    def setUp(self):
+        self.task = YAIBLengthOfStayMIMIC4()
+
+    def test_remaining_hours_at_every_hour(self):
+        patient = _patient(
+            stays=[(10, 1, T0, 20.5, range(20))], admissions=[_admission()]
+        )
+        sample = self.task(patient)[0]
+        self.assertEqual(sample["dynamic"]["time"], [float(h) for h in range(21)])
+        self.assertEqual(sample["los"]["time"], sample["dynamic"]["time"])
+        self.assertEqual(_labels(sample["los"]), [float(20 - h) for h in range(21)])
+
+    def test_capped_at_seven_days(self):
+        patient = _patient(
+            stays=[(10, 1, T0, 200, range(200))], admissions=[_admission(hours=210)]
+        )
+        los = self.task(patient)[0]["los"]
+        self.assertEqual(len(los["time"]), 169)  # observed for 7 days
+        self.assertEqual(_labels(los)[0], 168.0)
+        self.assertEqual(_labels(los)[-1], 32.0)  # 200 - 168
+
+
 class TestYAIBAKIMIMIC4(unittest.TestCase):
     def setUp(self):
         self.task = YAIBAKIMIMIC4()
@@ -368,17 +400,17 @@ def test_creatinine_rise_labels_six_hours_around_onset(self):
         ]
         samples = self._run(extra)
         self.assertEqual(len(samples), 1)
-        timestamps, labels = samples[0]["aki"]
-        # Observed until 6 hours after the onset at hour 20.
-        self.assertEqual(len(timestamps), 27)
-        self.assertEqual(samples[0]["dynamic"][1].shape[0], 27)
-        self.assertEqual(labels, [0.0] * 14 + [1.0] * 13)
+        aki = samples[0]["aki"]
+        # Observed until 6 hours after the onset at hour 20, on the same hours.
+        self.assertEqual(aki["time"], samples[0]["dynamic"]["time"])
+        self.assertEqual(len(aki["time"]), 27)
+        self.assertEqual(_labels(aki), [0.0] * 14 + [1.0] * 13)
 
     def test_stay_without_aki_is_negative_for_seven_days(self):
         samples = self._run([_crea(T0 + timedelta(hours=2), 1.0)], hours=200)
-        timestamps, labels = samples[0]["aki"]
-        self.assertEqual(len(timestamps), 169)
-        self.assertEqual(sum(labels), 0)
+        aki = samples[0]["aki"]
+        self.assertEqual(len(aki["time"]), 169)
+        self.assertEqual(sum(_labels(aki)), 0)
 
     def test_onset_within_six_hours_is_excluded(self):
         extra = [
@@ -398,12 +430,12 @@ def test_urine_stage_divides_by_weight(self):
         # 20 mL every hour: 6-hour sum of rates 6 * 20 / 2 = 60 mL/h, below
         # 0.5 per kg only when divided by a weight above 120 kg.
         urine = [_urine(T0 + timedelta(hours=h, minutes=30), 20) for h in range(40)]
-        _, labels = self._run(urine + [_weight(130)])[0]["aki"]
+        labels = _labels(self._run(urine + [_weight(130)])[0]["aki"])
         self.assertGreater(sum(labels), 0)
-        _, labels = self._run(urine + [_weight(100)])[0]["aki"]
+        labels = _labels(self._run(urine + [_weight(100)])[0]["aki"])
         self.assertEqual(sum(labels), 0)
         # Without any weight the sums are not divided, as in YAIB.
-        _, labels = self._run(urine)[0]["aki"]
+        labels = _labels(self._run(urine)[0]["aki"])
         self.assertEqual(sum(labels), 0)
 
 
@@ -457,9 +489,9 @@ def test_sepsis_onset_labels_six_hours_around_onset(self):
         )
         samples = self.task(patient)
         self.assertEqual(len(samples), 1)
-        timestamps, labels = samples[0]["sepsis"]
-        self.assertEqual(len(timestamps), 29)  # until 6 hours after hour 22
-        self.assertEqual(labels, [0.0] * 16 + [1.0] * 13)
+        sepsis = samples[0]["sepsis"]
+        self.assertEqual(len(sepsis["time"]), 29)  # until 6 hours after hour 22
+        self.assertEqual(_labels(sepsis), [0.0] * 16 + [1.0] * 13)
 
     def test_intermittent_antibiotics_are_not_suspected_infection(self):
         extra = [_platelets(T0 + timedelta(hours=h), 200) for h in range(0, 22, 3)]
@@ -471,7 +503,7 @@ def test_intermittent_antibiotics_are_not_suspected_infection(self):
             admissions=[_admission(hours=130)],
             extra=extra,
         )
-        _, labels = self.task(patient)[0]["sepsis"]
+        labels = _labels(self.task(patient)[0]["sepsis"])
         self.assertEqual(sum(labels), 0)
 
     def test_suspected_infection_both_directions(self):
@@ -560,5 +592,34 @@ def test_urine24_needs_twelve_hours(self):
         self.assertEqual(urine[29], 50.0 * 25 * 24 / 25)
 
 
+class TestYAIBProcessing(unittest.TestCase):
+    """The tasks' schemas: processors on task samples."""
+
+    def test_dynamic_matches_the_previous_timeseries_preprocessing(self):
+        # The sequence_v2 configuration reproduces TimeseriesProcessor with
+        # z-score, forward fill (0 before the first value) and indicators.
+        rng = np.random.default_rng(0)
+        grids = []
+        for steps in (5, 9, 3):
+            values = rng.normal(loc=2.0, scale=3.0, size=(steps, 4))
+            values[rng.random(values.shape) < 0.4] = np.nan
+            grids.append(values)
+        grids[0][:, 3] = np.nan  # a feature never seen in one sample
+        yaib: LevelConfig = YAIBMortalityMIMIC4.input_schema["dynamic"][1]["levels"][0]
+        levels: list[LevelConfig] = [{**yaib, "values": [4]}]  # 4 features here
+        v2 = SequenceProcessorV2(levels=levels)
+        v2.fit([{"x": {"time": list(range(len(g))), "values": g}} for g in grids], "x")
+        def hourly(grid):  # the previous input format: datetimes and values
+            return [T0 + timedelta(hours=h) for h in range(len(grid))], grid
+
+        v1 = TimeseriesProcessor(normalization="zscore", missing_indicators=True)
+        v1.fit([{"x": hourly(g)} for g in grids], "x")
+        for grid in grids:
+            values, mask, _ = v2.process({"time": list(range(len(grid))), "values": grid})
+            old = v1.process(hourly(grid))
+            torch.testing.assert_close(values, old[:, :4])
+            torch.testing.assert_close(mask.float(), old[:, 4:])
+
+
 if __name__ == "__main__":
     unittest.main()