diff --git a/tzrec/main.py b/tzrec/main.py
index 651c7d682..e0b43b329 100644
--- a/tzrec/main.py
+++ b/tzrec/main.py
@@ -515,6 +515,12 @@ def run_eval(step: int, epoch: int) -> None:
             if lr.by_epoch:
                 lr.step()
 
+    # One-shot end-of-loop hook (default no-op; e.g. SidRqkmeans fits its FAISS
+    # codebook here). SID models run with periodic checkpointing disabled
+    # (save_checkpoints_steps/epochs = 0), so the tail final=True save below is
+    # the only checkpoint and persists whatever on_train_end produced.
+    _model.on_train_end()
+
     _log_train(
         i_step,
         losses,
diff --git a/tzrec/metrics/relative_l1.py b/tzrec/metrics/relative_l1.py
new file mode 100644
index 000000000..685307608
--- /dev/null
+++ b/tzrec/metrics/relative_l1.py
@@ -0,0 +1,58 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import torch
+from torchmetrics import Metric
+
+
+class RelativeL1(Metric):
+    """Mean symmetric relative-L1 error ``|t - p| / (max(|t|, |p|) + eps)``.
+
+    A bounded reconstruction-error metric (0 = exact, → 1 = unrelated). It is a
+    verbatim port of OpenOneRec's residual-K-Means ``calc_loss`` and is
+    deliberately **not** ``torchmetrics.MeanAbsolutePercentageError``, which uses
+    the asymmetric ``|t - p| / |t|`` denominator. Aggregation is element-wise
+    (count-weighted), so the reported value is the mean over all elements seen.
+    """
+
+    higher_is_better = False
+    is_differentiable = True
+
+    def __init__(self, epsilon: float = 1e-4, **kwargs) -> None:
+        super().__init__(**kwargs)
+        self.epsilon = epsilon
+        # float64 sum / long count: float32 loses integer precision past 2**24
+        # (~32K rows of a 512-dim embedding) under element-wise aggregation.
+        self.add_state(
+            "sum_rel",
+            default=torch.tensor(0.0, dtype=torch.float64),
+            dist_reduce_fx="sum",
+        )
+        self.add_state(
+            "count", default=torch.tensor(0, dtype=torch.long), dist_reduce_fx="sum"
+        )
+
+    def update(self, preds: torch.Tensor, target: torch.Tensor) -> None:
+        """Accumulate the relative-L1 error between ``preds`` and ``target``.
+
+        Args:
+            preds (Tensor): reconstruction, shape (B, D).
+            target (Tensor): ground-truth embedding, shape (B, D).
+        """
+        rel = torch.abs(target - preds) / (
+            torch.maximum(torch.abs(target), torch.abs(preds)) + self.epsilon
+        )
+        self.sum_rel += rel.sum().double()
+        self.count += rel.numel()
+
+    def compute(self) -> torch.Tensor:
+        """Mean relative-L1 over all elements (NaN before any update)."""
+        return self.sum_rel / self.count
diff --git a/tzrec/metrics/relative_l1_test.py b/tzrec/metrics/relative_l1_test.py
new file mode 100644
index 000000000..0f89c2ccd
--- /dev/null
+++ b/tzrec/metrics/relative_l1_test.py
@@ -0,0 +1,49 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import torch
+
+from tzrec.metrics.relative_l1 import RelativeL1
+
+
+class RelativeL1Test(unittest.TestCase):
+    def test_zero_on_identity(self) -> None:
+        metric = RelativeL1()
+        x = torch.randn(8, 4)
+        metric.update(x, x.clone())
+        self.assertAlmostEqual(metric.compute().item(), 0.0, places=6)
+
+    def test_matches_formula(self) -> None:
+        metric = RelativeL1(epsilon=1e-4)
+        p = torch.tensor([[1.0, 0.0]])
+        t = torch.tensor([[0.0, 2.0]])
+        # |t-p|/(max(|t|,|p|)+eps): [1/(1+eps), 2/(2+eps)], mean of the two.
+        expected = (1.0 / (1.0 + 1e-4) + 2.0 / (2.0 + 1e-4)) / 2
+        metric.update(p, t)
+        self.assertAlmostEqual(metric.compute().item(), expected, places=5)
+
+    def test_count_weighted_across_updates(self) -> None:
+        """Aggregation is element-wise, not a mean of per-batch means."""
+        metric = RelativeL1()
+        metric.update(torch.zeros(1, 4), torch.ones(1, 4))  # 4 elems, rel ~1
+        metric.update(torch.ones(3, 4), torch.ones(3, 4))  # 12 elems, rel 0
+        # Element-weighted: 4 nonzero over 16 elems -> ~0.25, NOT (1+0)/2 = 0.5.
+        per = 1.0 / (1.0 + 1e-4)  # rel of a 0-vs-1 element (with epsilon)
+        self.assertAlmostEqual(metric.compute().item(), 4 * per / 16, places=6)
+
+    def test_nan_before_update(self) -> None:
+        self.assertTrue(torch.isnan(RelativeL1().compute()))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tzrec/models/model.py b/tzrec/models/model.py
index 40da5335a..26ec63dbc 100644
--- a/tzrec/models/model.py
+++ b/tzrec/models/model.py
@@ -150,6 +150,15 @@ def compute_train_metric(self) -> Dict[str, torch.Tensor]:
             metric_results[metric_name] = metric.compute()
         return metric_results
 
+    def on_train_end(self) -> None:
+        """Hook fired once after the train_eval loop exits.
+
+        Default no-op; override for one-shot end-of-loop work (e.g.
+        :class:`SidRqkmeans` fits its FAISS codebook here). The tail
+        ``final=True`` checkpoint persists whatever it produced.
+        """
+        return
+
     def sparse_parameters(
         self,
     ) -> Tuple[Iterable[nn.Parameter], Iterable[nn.Parameter]]:
diff --git a/tzrec/models/sid_model.py b/tzrec/models/sid_model.py
index 973fcf99f..8db468799 100644
--- a/tzrec/models/sid_model.py
+++ b/tzrec/models/sid_model.py
@@ -18,6 +18,7 @@
 
 from tzrec.datasets.utils import BASE_DATA_GROUP, Batch
 from tzrec.features.feature import BaseFeature
+from tzrec.metrics.relative_l1 import RelativeL1
 from tzrec.metrics.unique_ratio import UniqueRatio
 from tzrec.models.model import BaseModel
 from tzrec.protos.model_pb2 import ModelConfig
@@ -40,9 +41,9 @@ class BaseSidModel(BaseModel):
 
     Subclasses build their quantizer in ``__init__`` (after calling
     ``super().__init__``) and implement :meth:`predict` and :meth:`loss`.
-    They extend :meth:`init_metric` (via ``super()``) and implement
-    :meth:`update_metric` to populate the registered metrics
-    (:meth:`update_train_metric` defaults to a no-op).
+    :meth:`predict` exposes the reconstruction under ``predictions["x_hat"]``
+    (only when meaningful) so the shared :meth:`update_metric` can score it.
+    (:meth:`update_train_metric` defaults to a no-op.)
 
     Args:
         model_config (ModelConfig): an instance of ModelConfig.
@@ -69,8 +70,17 @@ def __init__(
         self._input_dim = cfg.input_dim
         self._normalize_residuals = cfg.normalize_residuals
 
-        assert cfg.codebook, "codebook must be set, e.g. [256, 256, 256]"
+        if not cfg.codebook:
+            raise ValueError("codebook must be set, e.g. [256, 256, 256]")
         self._n_embed_list = list(cfg.codebook)
+        # Fail fast: a zero codebook entry / input_dim==0 only errors opaquely
+        # deep inside faiss, after the whole training pass.
+        if any(k < 1 for k in self._n_embed_list):
+            raise ValueError(
+                f"every codebook entry must be >= 1, got {self._n_embed_list}"
+            )
+        if self._input_dim < 1:
+            raise ValueError(f"input_dim must be >= 1, got {self._input_dim}")
         self._n_layers = len(self._n_embed_list)
 
     def _extract_feature(
@@ -99,14 +109,48 @@ def init_loss(self) -> None:
     def init_metric(self) -> None:
         """Initialize the eval metrics shared by all SID models.
 
-        ``mse``: reconstruction error (input vs. quantized / decoded).
-        ``unique_sid_ratio``: mean per-batch unique-SID ratio (distinct rows /
-        batch size; a batch-size-sensitive diversity proxy, not global
-        coverage). Subclasses call ``super().init_metric()`` then add extras.
+        - ``mse``: reconstruction error (input vs. quantized / decoded).
+        - ``rel_loss``: symmetric relative-L1 reconstruction error
+          (:class:`~tzrec.metrics.relative_l1.RelativeL1`); meaningful only with
+          ``normalize_residuals=False`` (else the reconstruction and the input
+          live on different scales).
+        - ``unique_sid_ratio``: mean per-batch unique-SID ratio (distinct rows /
+          batch size; a batch-size-sensitive diversity proxy, not global
+          coverage).
+
+        Subclasses that add extras call ``super().init_metric()`` first.
         """
         self._metric_modules["mse"] = torchmetrics.MeanSquaredError()
+        self._metric_modules["rel_loss"] = RelativeL1()
         self._metric_modules["unique_sid_ratio"] = UniqueRatio()
 
+    def update_metric(
+        self,
+        predictions: Dict[str, torch.Tensor],
+        batch: Batch,
+        losses: Optional[Dict[str, torch.Tensor]] = None,
+    ) -> None:
+        """Update eval metrics from the reconstruction + the re-extracted input.
+
+        ``predictions["x_hat"]`` is the model's reconstruction of the input
+        embedding (the centroid sum for RQ-KMeans, the decoder output for
+        RQ-VAE). Subclasses expose it only when it is meaningful, so a
+        not-yet-fitted model omits it and this logs nothing. The target
+        embedding is re-extracted from ``batch`` (it is an input, not an output).
+
+        Args:
+            predictions (dict): a dict of predicted result.
+            batch (Batch): input batch data.
+            losses (dict, optional): a dict of loss.
+        """
+        if "x_hat" not in predictions:
+            return
+        recon = predictions["x_hat"]
+        embedding = self._extract_feature(batch)
+        self._metric_modules["mse"].update(recon, embedding)
+        self._metric_modules["rel_loss"].update(recon, embedding)
+        self._metric_modules["unique_sid_ratio"].update(predictions["codes"])
+
     def update_train_metric(
         self,
         predictions: Dict[str, torch.Tensor],
diff --git a/tzrec/models/sid_rqkmeans.py b/tzrec/models/sid_rqkmeans.py
new file mode 100644
index 000000000..59b05af41
--- /dev/null
+++ b/tzrec/models/sid_rqkmeans.py
@@ -0,0 +1,202 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""SidRqkmeans: SID generation model using residual K-Means.
+
+Training is FAISS-only: ``predict`` collects embeddings into a CPU
+buffer; the actual FAISS fit is triggered ONCE after the train_eval
+loop ends, via the :meth:`BaseModel.on_train_end` lifecycle hook
+(``tzrec.main`` calls ``_model.on_train_end()`` unconditionally).
+"""
+
+from typing import Any, Dict, List, Optional
+
+import torch
+import torch.distributed as dist
+from torch import nn
+
+from tzrec.datasets.utils import Batch
+from tzrec.features.feature import BaseFeature
+from tzrec.models.sid_model import BaseSidModel
+from tzrec.modules.sid.kmeans_quantize import ReservoirSampler
+from tzrec.modules.sid.residual_kmeans_quantizer import (
+    ResidualKMeansQuantizer,
+)
+from tzrec.protos.model_pb2 import ModelConfig
+from tzrec.utils.logging_util import logger
+
+
+class SidRqkmeans(BaseSidModel):
+    """SID generation model using residual K-Means (FAISS-only).
+
+    No gradient-based training. The codebook is built once at the end
+    of the train_eval loop via a single FAISS K-Means pass over the
+    embeddings collected during training.
+
+    Args:
+        model_config (ModelConfig): an instance of ModelConfig.
+        features (list): list of features.
+        labels (list): list of label names.
+        sample_weights (list): sample weight names.
+    """
+
+    def __init__(
+        self,
+        model_config: ModelConfig,
+        features: List[BaseFeature],
+        labels: List[str],
+        sample_weights: Optional[List[str]] = None,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(model_config, features, labels, sample_weights, **kwargs)
+
+        # CPU-only: v1 restricts the whole model (train + inference) to the
+        # host. Refuse to run when CUDA is visible.
+        if torch.cuda.is_available():
+            raise RuntimeError(
+                "SidRqkmeans is CPU-only, but a CUDA device is visible. "
+                'Run with CUDA_VISIBLE_DEVICES="-1" (or on a CPU-only host).'
+            )
+
+        # Single-process only: the fit runs over one process's local reservoir,
+        # with no cross-rank gather. Fail fast before the (wasted) train pass.
+        if dist.is_available() and dist.is_initialized() and dist.get_world_size() > 1:
+            raise RuntimeError(
+                "SidRqkmeans supports single-process training only "
+                f"(world_size=1); got world_size={dist.get_world_size()}. "
+                "Launch with --nproc-per-node=1."
+            )
+
+        cfg = self._model_config
+
+        # Typed faiss kwargs: only the explicitly-set fields are forwarded, so
+        # unset ones fall back to faiss's own defaults (no float->int coercion).
+        self._faiss_kwargs = {
+            f.name: v for f, v in cfg.faiss_kmeans_kwargs.ListFields()
+        }
+
+        self._quantizer = ResidualKMeansQuantizer(
+            embed_dim=self._input_dim,
+            n_layers=self._n_layers,
+            n_embed=self._n_embed_list,
+            normalize_residuals=self._normalize_residuals,
+            faiss_kmeans_kwargs=self._faiss_kwargs,
+        )
+
+        # Bounded host reservoir for the end-of-loop fit: cap at
+        # ``train_sample_size`` (when >0) else the fit's subsample size, rather
+        # than buffer the whole corpus.
+        target = self._model_config.train_sample_size
+        cap = target if target > 0 else self._quantizer.default_fit_sample_size()
+        # Fail fast: a cap below the largest codebook would only fail deep in
+        # train_offline, after the whole training pass.
+        max_k = max(self._n_embed_list)
+        if cap < max_k:
+            raise RuntimeError(
+                f"reservoir cap ({cap}) < largest codebook size ({max_k}); set "
+                f"train_sample_size >= {max_k} (or 0 for the default)."
+            )
+        self._reservoir = ReservoirSampler(cap, self._input_dim)
+
+        # KMeans has no learnable params; a dummy keeps the optimizer/DDP happy.
+        self._dummy_param = nn.Parameter(torch.zeros(1), requires_grad=True)
+
+    def predict(self, batch: Batch) -> Dict[str, torch.Tensor]:
+        """Predict the model.
+
+        Training: buffer embeddings only (codes are dummy until FAISS fits).
+        Eval/inference (after ``on_train_end``): real predict + lookup.
+
+        Args:
+            batch (Batch): input batch data.
+
+        Return:
+            predictions (dict): a dict of predicted result.
+        """
+        embedding = self._extract_feature(batch)
+
+        # Training: reservoir-sample only; codes are dummy until the fit.
+        if self.is_train:
+            self._reservoir.add(embedding)
+            B = embedding.shape[0]
+            return {
+                "codes": torch.zeros(
+                    B, self._n_layers, dtype=torch.long, device=embedding.device
+                )
+            }
+
+        codes, quantized = self._quantizer(embedding)
+
+        predictions: Dict[str, torch.Tensor] = {
+            "codes": codes,
+        }
+
+        # Expose the centroid-sum reconstruction (``x_hat``) for update_metric
+        # only once fitted — pre-fit it is all-zeros, so omitting it skips the
+        # eval metrics. (Meaningful only with normalize_residuals=False.)
+        if self.is_eval and self._quantizer.is_fitted:
+            predictions["x_hat"] = quantized
+
+        return predictions
+
+    def loss(
+        self, predictions: Dict[str, torch.Tensor], batch: Batch
+    ) -> Dict[str, torch.Tensor]:
+        """Compute loss of the model.
+
+        Zero loss via ``_dummy_param * 0`` — gives TrainWrapper/DDP a compute
+        graph despite there being no real trainable params.
+
+        Args:
+            predictions (dict): a dict of predicted result.
+            batch (Batch): input batch data.
+
+        Return:
+            losses (dict): a dict of loss tensor.
+        """
+        return {"dummy_loss": self._dummy_param.sum() * 0.0}
+
+    @torch.no_grad()
+    def on_train_end(self) -> None:
+        """Fit the FAISS codebook once, after the train_eval loop exits.
+
+        Overrides :meth:`BaseModel.on_train_end` (called unconditionally by
+        ``tzrec.main``). Single-process only (enforced by the world_size guard
+        in ``__init__``): the fit runs on one process over its local reservoir,
+        with no cross-rank gather/broadcast. The tail ``final=True`` checkpoint
+        then persists the fitted codebook (SID runs with periodic checkpointing
+        disabled, so that save is never deduped away).
+
+        TODO: "periodic checkpointing disabled" is a convention, not enforced.
+        With save_checkpoints_steps/epochs > 0, a final-step in-loop save can
+        dedupe the tail save away, silently dropping the fitted codebook. Harden
+        this (enforce the contract / bypass the dedupe) in a future update.
+
+        An empty reservoir only happens for a pathologically tiny corpus; the
+        fit is then skipped.
+        """
+        # train_offline consumes its input; hand it the reservoir buffer
+        # directly (no copy) — nothing reads it after this.
+        local = self._reservoir.sample()
+        self._reservoir.reset()
+
+        if local.shape[0] == 0:
+            logger.warning(
+                "[SidRqkmeans.on_train_end] empty reservoir; skipping FAISS "
+                "fit. Did the train_eval loop run?"
+            )
+            return
+
+        logger.info(
+            "[SidRqkmeans.on_train_end] fitting FAISS on %d samples (D=%d)."
+            % (local.shape[0], local.shape[1])
+        )
+        self._quantizer.train_offline(local, verbose=True)
diff --git a/tzrec/models/sid_rqkmeans_test.py b/tzrec/models/sid_rqkmeans_test.py
new file mode 100644
index 000000000..0b68fefa6
--- /dev/null
+++ b/tzrec/models/sid_rqkmeans_test.py
@@ -0,0 +1,382 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+from unittest import mock
+
+import torch
+import torch.distributed as dist
+from torchrec import KeyedTensor
+
+from tzrec.datasets.utils import BASE_DATA_GROUP, Batch
+from tzrec.models.sid_rqkmeans import SidRqkmeans
+from tzrec.protos import model_pb2
+from tzrec.protos.models import sid_model_pb2
+from tzrec.utils.state_dict_util import init_parameters
+
+
+def _batch_from_rows(rows: torch.Tensor) -> Batch:
+    """Wrap explicit ``item_emb`` rows in a minimal Batch."""
+    dense_feature = KeyedTensor.from_tensor_list(keys=["item_emb"], tensors=[rows])
+    return Batch(
+        dense_features={BASE_DATA_GROUP: dense_feature},
+        sparse_features={},
+        labels={},
+    )
+
+
+def _make_batch(batch_size: int, input_dim: int, device: str = "cpu") -> Batch:
+    """Create a minimal Batch with random dense embedding features."""
+    return _batch_from_rows(torch.randn(batch_size, input_dim, device=device))
+
+
+class SidRqkmeansOfflineTest(unittest.TestCase):
+    """Single-process tests for SidRqkmeans (FAISS-only)."""
+
+    def setUp(self) -> None:
+        # SidRqkmeans is CPU-only and refuses to init when CUDA is visible. The
+        # GPU CI runners have CUDA, so simulate a CPU-only host for every
+        # construction-based test. (test_init_raises_on_gpu overrides this.)
+        patcher = mock.patch.object(torch.cuda, "is_available", return_value=False)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _create_model(
+        self,
+        input_dim=32,
+        n_layers=2,
+        niter=5,
+        codebook=None,
+        normalize_residuals=False,
+        train_sample_size=0,
+    ):
+        """Build a SidRqkmeans on CPU with params initialized.
+
+        SID models read the item-embedding dense feature directly from the
+        batch and do not consume feature_groups, so none is set.
+        """
+        n_embed_list = codebook if codebook is not None else [16] * n_layers
+        faiss_kwargs = sid_model_pb2.FaissKmeansConfig(
+            niter=niter, verbose=False, seed=1234
+        )
+        cfg = sid_model_pb2.SidRqkmeans(
+            input_dim=input_dim,
+            codebook=n_embed_list,
+            normalize_residuals=normalize_residuals,
+            faiss_kmeans_kwargs=faiss_kwargs,
+            embedding_feature_name="item_emb",
+            train_sample_size=train_sample_size,
+        )
+        model = SidRqkmeans(
+            model_config=model_pb2.ModelConfig(sid_rqkmeans=cfg),
+            features=[],
+            labels=[],
+        )
+        init_parameters(model, device=torch.device("cpu"))
+        return model
+
+    def test_proto_parse(self) -> None:
+        """Verify faiss_kmeans_kwargs are parsed correctly."""
+        model = self._create_model()
+        self.assertEqual(model._faiss_kwargs.get("niter"), 5)
+        self.assertEqual(model._faiss_kwargs.get("seed"), 1234)
+        self.assertFalse(model._faiss_kwargs.get("verbose"))
+        self.assertEqual(model._reservoir.n_seen, 0)
+        self.assertEqual(model._reservoir.n_filled, 0)
+
+    def test_sample_cap_from_train_sample_size(self) -> None:
+        """train_sample_size (when set) drives the reservoir cap directly."""
+        # Explicit train_sample_size: cap == train_sample_size.
+        model = self._create_model(train_sample_size=900)
+        self.assertEqual(model._reservoir.capacity, 900)
+
+        # Default (train_sample_size=0): cap == the FAISS fit's subsample size.
+        model = self._create_model()
+        self.assertEqual(
+            model._reservoir.capacity, model._quantizer.default_fit_sample_size()
+        )
+
+    def test_init_raises_on_too_small_train_sample_size(self) -> None:
+        """train_sample_size below the largest codebook fails fast at init."""
+        with self.assertRaisesRegex(RuntimeError, "largest codebook"):
+            self._create_model(codebook=[16, 16], train_sample_size=8)
+
+    def test_init_raises_on_empty_codebook(self) -> None:
+        """An empty codebook fails fast at construction."""
+        with self.assertRaisesRegex(ValueError, "codebook must be set"):
+            self._create_model(codebook=[])
+
+    def test_init_raises_on_zero_codebook_entry(self) -> None:
+        """A zero codebook entry fails fast at construction."""
+        with self.assertRaisesRegex(ValueError, "codebook entry must be >= 1"):
+            self._create_model(codebook=[16, 0])
+
+    def test_init_raises_on_zero_input_dim(self) -> None:
+        """input_dim < 1 fails fast at construction."""
+        with self.assertRaisesRegex(ValueError, "input_dim must be >= 1"):
+            self._create_model(input_dim=0)
+
+    def test_predict_collects_buffer(self) -> None:
+        """In train mode, predict reservoir-samples; never fits."""
+        B, input_dim = 8, 32
+        model = self._create_model(input_dim=input_dim)
+        model.train()
+
+        for _ in range(4):
+            batch = _make_batch(B, input_dim)
+            preds = model.predict(batch)
+            self.assertIn("codes", preds)
+
+        # Reservoir holds all 4*B samples (well under the cap) and tracks
+        # the running count.
+        self.assertEqual(model._reservoir.n_seen, 4 * B)
+        self.assertEqual(model._reservoir.n_filled, 4 * B)
+        # FAISS not yet triggered: layers should be uninitialized
+        for layer in model._quantizer.layers:
+            self.assertFalse(layer.is_initialized)
+
+    def test_on_train_end_runs_faiss(self) -> None:
+        """on_train_end triggers FAISS fit and clears buffer."""
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+
+        B, input_dim = 64, 32
+        model = self._create_model(input_dim=input_dim)
+        model.train()
+
+        # Accumulate enough samples (FAISS K-Means needs at least K points)
+        for _ in range(8):
+            model.predict(_make_batch(B, input_dim))
+        self.assertGreater(model._reservoir.n_seen, 0)
+
+        # Trigger one-shot FAISS fit.
+        model.on_train_end()
+
+        # Reservoir should be released after the fit
+        self.assertEqual(model._reservoir.n_seen, 0)
+        self.assertEqual(model._reservoir.n_filled, 0)
+        # All layers should be initialized + centroids non-zero
+        for layer in model._quantizer.layers:
+            self.assertTrue(bool(layer._is_initialized.item()))
+            self.assertGreater(layer.centroids.abs().sum().item(), 0.0)
+
+        # After fit, predict on eval should produce valid codes
+        model.eval()
+        preds = model.predict(_make_batch(B, input_dim))
+        codes = preds["codes"]
+        self.assertEqual(codes.shape, (B, 2))
+        self.assertTrue((codes >= 0).all() and (codes < 16).all())
+
+    def test_non_uniform_codebook_end_to_end(self) -> None:
+        """Non-uniform codebook [8, 4, 16]: fit then emit per-layer codes."""
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+
+        B, input_dim = 64, 32
+        codebook = [8, 4, 16]
+        model = self._create_model(input_dim=input_dim, codebook=codebook)
+        # Reservoir cap derives from the LARGEST K (16), not the first (8).
+        self.assertEqual(
+            model._reservoir.capacity,
+            16 * int(model._faiss_kwargs.get("max_points_per_centroid", 256)),
+        )
+
+        model.train()
+        for _ in range(8):
+            model.predict(_make_batch(B, input_dim))
+        model.on_train_end()
+
+        for k, layer in zip(codebook, model._quantizer.layers):
+            self.assertTrue(bool(layer._is_initialized.item()))
+            self.assertEqual(layer.centroids.shape[0], k)
+
+        model.eval()
+        codes = model.predict(_make_batch(B, input_dim))["codes"]
+        self.assertEqual(codes.shape, (B, 3))
+        for i, k in enumerate(codebook):
+            self.assertTrue((codes[:, i] >= 0).all() and (codes[:, i] < k).all())
+
+    def test_normalize_residuals_end_to_end(self) -> None:
+        """train_offline with normalize_residuals=True fits + predicts.
+
+        Exercises the ``F.normalize`` site inside ``train_offline`` (a second
+        normalize independent of ``_residual_pass``), which the other tests —
+        all built with normalize_residuals=False — never reach.
+        """
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+
+        B, input_dim = 64, 32
+        model = self._create_model(input_dim=input_dim, normalize_residuals=True)
+        self.assertTrue(model._quantizer.normalize_residuals)
+
+        model.train()
+        for _ in range(8):
+            model.predict(_make_batch(B, input_dim))
+        model.on_train_end()
+
+        for layer in model._quantizer.layers:
+            self.assertTrue(layer.is_initialized)
+
+        model.eval()
+        codes = model.predict(_make_batch(B, input_dim))["codes"]
+        self.assertEqual(codes.shape, (B, 2))
+        self.assertTrue((codes >= 0).all() and (codes < 16).all())
+
+    def test_eval_and_inference_predict_contract(self) -> None:
+        """Eval (post-fit) exposes codes + x_hat; inference is codes-only."""
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+
+        B, input_dim = 64, 32
+        model = self._create_model(input_dim=input_dim)
+        model.train()
+        for _ in range(8):
+            model.predict(_make_batch(B, input_dim))
+        model.on_train_end()
+
+        # Eval mode (fitted): the reconstruction is exposed as ``x_hat`` for
+        # update_metric; the input embedding is re-extracted from the batch
+        # there, not threaded through predictions.
+        model.eval()
+        eval_preds = model.predict(_make_batch(B, input_dim))
+        self.assertEqual(set(eval_preds.keys()), {"codes", "x_hat"})
+
+        # Inference (serving) mode: codes-only contract.
+        model.set_is_inference(True)
+        inf_preds = model.predict(_make_batch(B, input_dim))
+        self.assertEqual(set(inf_preds.keys()), {"codes"})
+
+    def test_eval_metric_path(self) -> None:
+        """init_metric/update_metric report finite mse + rel_loss in eval."""
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+
+        B, input_dim = 64, 32
+        model = self._create_model(input_dim=input_dim)
+        model.train()
+        for _ in range(8):
+            model.predict(_make_batch(B, input_dim))
+        model.on_train_end()
+
+        model.init_metric()
+        model.eval()
+        # Same batch through predict + update_metric: the reconstruction target
+        # is re-extracted from this batch, so it must match the predicted one.
+        batch = _make_batch(B, input_dim)
+        preds = model.predict(batch)
+        model.update_metric(preds, batch)
+        metrics = model.compute_metric()
+        for key in ("mse", "rel_loss", "unique_sid_ratio"):
+            self.assertIn(key, metrics)
+            self.assertTrue(torch.isfinite(torch.as_tensor(metrics[key])).all())
+
+    def test_update_metric_skipped_before_fit(self) -> None:
+        """Pre-fit eval (unfitted codebook) does not pollute metric state."""
+        B, input_dim = 8, 32
+        model = self._create_model(input_dim=input_dim)
+        model.init_metric()
+        model.eval()
+        # Codebook not fitted yet: predict emits zeros; update_metric must skip.
+        batch = _make_batch(B, input_dim)
+        model.update_metric(model.predict(batch), batch)
+        self.assertEqual(model._metric_modules["unique_sid_ratio"].count.item(), 0.0)
+
+    def test_on_train_end_noop_on_empty_buffer(self) -> None:
+        """on_train_end on an empty buffer is a warned no-op."""
+        model = self._create_model()
+        model.on_train_end()  # warns and returns without fitting; must not raise
+        self.assertFalse(model._quantizer.is_fitted)
+
+    def test_init_raises_under_ddp(self) -> None:
+        """SidRqkmeans is single-process only: world_size>1 fails fast in init."""
+        with (
+            mock.patch.object(dist, "is_available", return_value=True),
+            mock.patch.object(dist, "is_initialized", return_value=True),
+            mock.patch.object(dist, "get_world_size", return_value=2),
+            self.assertRaisesRegex(RuntimeError, "single-process"),
+        ):
+            self._create_model()
+
+    def test_init_raises_on_gpu(self) -> None:
+        """SidRqkmeans is CPU-only: a visible CUDA device fails fast in init."""
+        with (
+            mock.patch.object(torch.cuda, "is_available", return_value=True),
+            self.assertRaisesRegex(RuntimeError, "CPU-only"),
+        ):
+            self._create_model()
+
+    def test_post_fit_checkpoint_round_trips(self) -> None:
+        """Fit → save state_dict → load into fresh instance → predict.
+
+        The reloaded model must produce the *same* codes as the source on the
+        same batch — verifying the centroids round-trip exactly, not merely
+        that they came through as non-zero.
+        """
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+
+        B, input_dim = 64, 32
+        src = self._create_model(input_dim=input_dim)
+        src.train()
+        for _ in range(8):
+            src.predict(_make_batch(B, input_dim))
+        src.on_train_end()
+        sd = src.state_dict()
+
+        dst = self._create_model(input_dim=input_dim)
+        dst.load_state_dict(sd)
+
+        # Same batch through both → identical codes (exact round-trip).
+        batch = _make_batch(B, input_dim)
+        src.eval()
+        dst.eval()
+        src_codes = src.predict(batch)["codes"]
+        dst_codes = dst.predict(batch)["codes"]
+        self.assertGreater(
+            dst_codes.abs().sum().item(),
+            0,
+            "post-fit checkpoint resume produced all-zero codes",
+        )
+        torch.testing.assert_close(dst_codes, src_codes)
+
+    def test_mid_fit_checkpoint_rejected_on_load(self) -> None:
+        """Tampered state (_is_initialized=True + zero centroids) raises."""
+        model = self._create_model()
+        sd = model.state_dict()
+        # Simulate a checkpoint that captured the flag mid-fit (before
+        # load_centroids_ ran): True flag, zero centroids.
+        layer0_prefix = next(
+            k.rsplit("._is_initialized", 1)[0]
+            for k in sd
+            if k.endswith("._is_initialized")
+        )
+        sd[f"{layer0_prefix}._is_initialized"] = torch.tensor(True)
+
+        fresh = self._create_model()
+        with self.assertRaisesRegex(RuntimeError, "mid-FAISS-fit"):
+            fresh.load_state_dict(sd)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tzrec/modules/sid/kmeans_quantize.py b/tzrec/modules/sid/kmeans_quantize.py
new file mode 100644
index 000000000..6eb5b940a
--- /dev/null
+++ b/tzrec/modules/sid/kmeans_quantize.py
@@ -0,0 +1,230 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""K-Means utilities for the SID-generation stack.
+
+This module is the single home for torch-native K-Means code used by
+SID models:
+
+* :class:`KMeansQuantizeLayer` — the K-Means :class:`QuantizeLayer`: a
+  centroid container populated by the FAISS backend via ``load_centroids_``.
+* :class:`ReservoirSampler` — bounded uniform stream sample (Vitter
+  Algorithm R) that :class:`~tzrec.models.sid_rqkmeans.SidRqkmeans`
+  fills during training to feed the one-shot FAISS fit.
+"""
+
+from typing import Optional
+
+import torch
+
+from tzrec.modules.sid.quantize_layer import QuantizeLayer
+from tzrec.modules.sid.types import QuantizeOutput
+from tzrec.utils.logging_util import logger
+
+
+class ReservoirSampler:
+    """Bounded uniform sample of a stream (Vitter Algorithm R).
+
+    Keeps a uniform ``capacity``-row sample of all rows passed to ``add``, in
+    O(capacity) host (CPU) memory — used to subsample the training corpus for
+    the one-shot FAISS fit without buffering the whole corpus. The buffer is a
+    CPU float32 tensor, allocated lazily on the first ``add``.
+
+    Args:
+        capacity (int): max rows retained.
+        dim (int): row width (feature dimension).
+    """
+
+    def __init__(self, capacity: int, dim: int) -> None:
+        self._cap = capacity
+        self._dim = dim
+        # Allocated lazily on the first add. _n_filled = used slots;
+        # _n_seen = running count for the accept prob.
+        self._buf: Optional[torch.Tensor] = None
+        self._n_filled = 0
+        self._n_seen = 0
+        logger.info("[ReservoirSampler] capacity=%d, dim=%d", capacity, dim)
+
+    @property
+    def capacity(self) -> int:
+        """Max rows retained."""
+        return self._cap
+
+    @property
+    def n_seen(self) -> int:
+        """Total rows passed to ``add`` so far."""
+        return self._n_seen
+
+    @property
+    def n_filled(self) -> int:
+        """Rows currently held (<= capacity)."""
+        return self._n_filled
+
+    @torch.no_grad()
+    def add(self, x: torch.Tensor) -> None:
+        """Stream a batch of rows into the reservoir.
+
+        Args:
+            x (Tensor): rows to add, shape (B, dim).
+        """
+        x = x.detach()
+        cap = self._cap
+        if self._buf is None:
+            self._buf = torch.empty(cap, self._dim, dtype=torch.float32)
+
+        # Phase 1: fill empty slots first. x is on the host, so ``.to`` is a
+        # dtype cast into the buffer, not a device copy.
+        if self._n_filled < cap:
+            take = min(x.shape[0], cap - self._n_filled)
+            self._buf[self._n_filled : self._n_filled + take] = x[:take].to(
+                torch.float32
+            )
+            self._n_filled += take
+            self._n_seen += take
+            x = x[take:]
+            if x.shape[0] == 0:
+                return
+
+        # Phase 2: row j enters with prob cap/(n_seen+j+1), displacing a random
+        # slot. float64 keeps n_seen+j+1 exact past 2**24.
+        r = x.shape[0]
+        pos = self._n_seen + torch.arange(r)
+        accept = torch.rand(r) < (cap / (pos + 1).to(torch.float64))
+        idx = accept.nonzero(as_tuple=True)[0]
+        if idx.numel() > 0:
+            slots = torch.randint(0, cap, (idx.numel(),))
+            # Slot collisions are last-write-wins; O(B/cap) bias, negligible here.
+            self._buf[slots] = x[idx].to(torch.float32)
+        self._n_seen += r
+
+    def sample(self) -> torch.Tensor:
+        """Return the filled portion of the reservoir, shape (n_filled, dim)."""
+        if self._buf is None or self._n_filled == 0:
+            return torch.empty(0, self._dim, dtype=torch.float32)
+        return self._buf[: self._n_filled]
+
+    def reset(self) -> None:
+        """Drop the buffer and counters to free host memory."""
+        self._buf = None
+        self._n_filled = 0
+        self._n_seen = 0
+
+
+class KMeansQuantizeLayer(QuantizeLayer):
+    """K-Means :class:`QuantizeLayer`: a centroid codebook + nearest assignment.
+
+    Centroids are populated externally by ``load_centroids_`` (the FAISS
+    backend in :class:`ResidualKMeansQuantizer`); ``quantize`` is the only
+    forward path. (The k-means *fit* lives in the quantizer; this layer just
+    holds the resulting centroids.)
+
+    Args:
+        n_embed (int): number of centroids (codebook size).
+        embed_dim (int): feature dimension.
+    """
+
+    def __init__(self, n_embed: int, embed_dim: int) -> None:
+        super().__init__(n_embed, embed_dim)
+        self.register_buffer("centroids", torch.zeros(n_embed, embed_dim))
+        # Persistent so a post-fit checkpoint round-trips; a mid-fit poison
+        # (True flag + zero centroids) is caught in _load_from_state_dict.
+        self.register_buffer("_is_initialized", torch.tensor(False))
+        # Plain-Python mirror of the buffer, read on the per-batch forward
+        # path to avoid a .item() GPU->CPU sync. Synced only via
+        # mark_initialized_ and _load_from_state_dict.
+        self._initialized: bool = False
+
+    @property
+    def is_initialized(self) -> bool:
+        """Whether centroids have been injected via ``load_centroids_``."""
+        return self._initialized
+
+    def mark_initialized_(self) -> None:
+        """Flag centroids populated, syncing buffer + cached mirror."""
+        self._is_initialized.fill_(True)
+        self._initialized = True
+
+    @torch.no_grad()
+    def load_centroids_(self, centroids: torch.Tensor) -> None:
+        """Inject offline-trained centroids.
+
+        Args:
+            centroids (Tensor): externally trained centroids,
+                shape (n_embed, embed_dim).
+        """
+        # raise (not assert): under ``python -O`` a dropped assert would let a
+        # (1, D) tensor broadcast-replicate into all K centroid rows silently.
+        if centroids.shape != self.centroids.shape:
+            raise RuntimeError(
+                f"centroids shape mismatch: expected {tuple(self.centroids.shape)}, "
+                f"got {tuple(centroids.shape)}"
+            )
+        self.centroids.copy_(
+            centroids.to(dtype=self.centroids.dtype, device=self.centroids.device)
+        )
+        self.mark_initialized_()
+
+    def _load_from_state_dict(
+        self,
+        state_dict,
+        prefix,
+        local_metadata,
+        strict,
+        missing_keys,
+        unexpected_keys,
+        error_msgs,
+    ) -> None:
+        """Reject mid-fit-checkpoint state dicts (True flag + zero centroids)."""
+        super()._load_from_state_dict(
+            state_dict,
+            prefix,
+            local_metadata,
+            strict,
+            missing_keys,
+            unexpected_keys,
+            error_msgs,
+        )
+        # Mirror the restored buffer into the cached flag (one load-time sync).
+        self._initialized = bool(self._is_initialized.item())
+        if self._initialized and self.centroids.abs().sum() == 0:
+            error_msgs.append(
+                f"KMeansQuantizeLayer at '{prefix}': _is_initialized=True but "
+                "centroids are all zero — checkpoint was likely taken "
+                "mid-FAISS-fit. Re-run on_train_end to produce a valid checkpoint."
+            )
+
+    @torch.no_grad()
+    def quantize(self, x: torch.Tensor, temperature: float = 1.0) -> QuantizeOutput:
+        """Assign points to the nearest centroid and gather them.
+
+        Uses ``torch.cdist`` (L2); argmin is invariant to the monotonic sqrt,
+        so assignments match squared-L2 except at exact equidistant ties
+        (measure zero for real embeddings), where either centroid is valid.
+        Before the FAISS fit (uninitialized) this returns all-zero codes +
+        embeddings so the residual walk stays a no-op and the model is callable.
+        ``temperature`` is unused (no soft assignment).
+
+        Args:
+            x (Tensor): data points, shape (B, D).
+            temperature (float): unused.
+
+        Returns:
+            QuantizeOutput: ``ids`` (B,) and ``embeddings`` (B, D).
+        """
+        if not self.is_initialized:
+            ids = torch.zeros(x.shape[0], dtype=torch.long, device=x.device)
+            return QuantizeOutput(embeddings=torch.zeros_like(x), ids=ids)
+        ids = torch.cdist(x, self.centroids).argmin(dim=-1)
+        return QuantizeOutput(embeddings=self.centroids[ids], ids=ids)
+
+    def get_codebook_embeddings(self) -> torch.Tensor:
+        """Return the centroid table, shape (n_embed, embed_dim)."""
+        return self.centroids
diff --git a/tzrec/modules/sid/kmeans_quantize_test.py b/tzrec/modules/sid/kmeans_quantize_test.py
new file mode 100644
index 000000000..2f2883562
--- /dev/null
+++ b/tzrec/modules/sid/kmeans_quantize_test.py
@@ -0,0 +1,146 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import torch
+
+from tzrec.modules.sid.kmeans_quantize import (
+    KMeansQuantizeLayer,
+    ReservoirSampler,
+)
+
+
+class KMeansQuantizeLayerTest(unittest.TestCase):
+    """Tests for the single KMeansQuantizeLayer."""
+
+    def test_uninitialized_by_default(self) -> None:
+        layer = KMeansQuantizeLayer(n_embed=4, embed_dim=3)
+        self.assertFalse(layer.is_initialized)
+        self.assertEqual(layer.centroids.abs().sum().item(), 0.0)
+
+    def test_load_centroids_and_quantize(self) -> None:
+        layer = KMeansQuantizeLayer(n_embed=2, embed_dim=2)
+        centroids = torch.tensor([[0.0, 0.0], [10.0, 10.0]])
+        layer.load_centroids_(centroids)
+        self.assertTrue(layer.is_initialized)
+
+        batch = torch.tensor([[0.1, 0.0], [9.0, 11.0]])
+        out = layer.quantize(batch)
+        torch.testing.assert_close(out.ids, torch.tensor([0, 1]))
+        # embeddings are the gathered centroids; lookup matches.
+        torch.testing.assert_close(out.embeddings, centroids[out.ids])
+        torch.testing.assert_close(layer.lookup(out.ids), out.embeddings)
+
+    def test_quantize_uninitialized_returns_zeros(self) -> None:
+        layer = KMeansQuantizeLayer(n_embed=4, embed_dim=3)
+        out = layer.quantize(torch.randn(5, 3))
+        self.assertEqual(out.ids.shape, (5,))
+        self.assertEqual(int(out.ids.abs().sum()), 0)
+        torch.testing.assert_close(out.embeddings, torch.zeros(5, 3))
+
+    def test_load_centroids_shape_mismatch_raises(self) -> None:
+        layer = KMeansQuantizeLayer(n_embed=2, embed_dim=2)
+        with self.assertRaises(RuntimeError):
+            layer.load_centroids_(torch.zeros(3, 2))
+
+    def test_mid_fit_checkpoint_rejected(self) -> None:
+        layer = KMeansQuantizeLayer(n_embed=2, embed_dim=2)
+        sd = layer.state_dict()
+        # Simulate a mid-fit checkpoint: flag True but centroids still zero.
+        sd["_is_initialized"] = torch.tensor(True)
+        fresh = KMeansQuantizeLayer(n_embed=2, embed_dim=2)
+        with self.assertRaisesRegex(RuntimeError, "mid-FAISS-fit"):
+            fresh.load_state_dict(sd)
+
+    def test_post_fit_checkpoint_round_trips(self) -> None:
+        layer = KMeansQuantizeLayer(n_embed=2, embed_dim=2)
+        layer.load_centroids_(torch.tensor([[1.0, 2.0], [3.0, 4.0]]))
+        fresh = KMeansQuantizeLayer(n_embed=2, embed_dim=2)
+        fresh.load_state_dict(layer.state_dict())
+        self.assertTrue(fresh.is_initialized)
+        torch.testing.assert_close(fresh.centroids, layer.centroids)
+
+
+class ReservoirSamplerTest(unittest.TestCase):
+    """Tests for the bounded reservoir sampler (Vitter Algorithm R)."""
+
+    def test_empty_sample(self) -> None:
+        """sample() before any add returns an empty (0, dim) tensor."""
+        r = ReservoirSampler(capacity=10, dim=4)
+        self.assertEqual(r.sample().shape, (0, 4))
+        self.assertEqual(r.n_seen, 0)
+        self.assertEqual(r.n_filled, 0)
+
+    def test_caps_memory(self) -> None:
+        """The buffer is bounded at capacity regardless of stream length."""
+        cap, dim, B = 10, 8, 16
+        r = ReservoirSampler(capacity=cap, dim=dim)
+        for _ in range(20):  # 320 rows >> cap
+            r.add(torch.randn(B, dim))
+        self.assertEqual(r.n_seen, 20 * B)
+        self.assertEqual(r.n_filled, cap)
+        self.assertEqual(r.sample().shape, (cap, dim))
+
+    def test_phase2_replacement(self) -> None:
+        """Phase-2 replacement keeps a valid sample of real, in-range rows.
+
+        Feeds identifiable rows (each row's value == its global stream index),
+        then asserts every slot still holds an intact fed row, all indices are
+        in range, and replacement past the initial fill actually happened —
+        exercising the accept-prob / slot-write logic that the count/shape-only
+        ``test_caps_memory`` cannot.
+        """
+        torch.manual_seed(0)
+        dim, cap, B, n_batches = 4, 8, 4, 50
+        r = ReservoirSampler(capacity=cap, dim=dim)
+
+        gidx = 0
+        for _ in range(n_batches):
+            rows = (
+                torch.arange(gidx, gidx + B, dtype=torch.float32)
+                .unsqueeze(1)
+                .expand(B, dim)
+                .contiguous()
+            )
+            gidx += B
+            r.add(rows)
+
+        total = B * n_batches
+        self.assertEqual(r.n_seen, total)
+        self.assertEqual(r.n_filled, cap)
+
+        res = r.sample()
+        idx = res[:, 0].round().long()
+        # Each stored row is an intact fed row (all columns equal its index).
+        self.assertTrue(
+            torch.equal(res, idx.unsqueeze(1).float().expand_as(res)),
+            "reservoir holds corrupted (non-fed) rows",
+        )
+        # All indices are valid stream positions.
+        self.assertTrue((idx >= 0).all() and (idx < total).all())
+        # Phase-2 replacement happened: at least one slot holds a row added
+        # after the reservoir filled (index >= cap).
+        self.assertTrue((idx >= cap).any(), "no Phase-2 replacement occurred")
+
+    def test_reset(self) -> None:
+        """reset() drops the buffer and counters."""
+        r = ReservoirSampler(capacity=10, dim=4)
+        r.add(torch.randn(5, 4))
+        self.assertEqual(r.n_filled, 5)
+        r.reset()
+        self.assertEqual(r.n_seen, 0)
+        self.assertEqual(r.n_filled, 0)
+        self.assertEqual(r.sample().shape, (0, 4))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tzrec/modules/sid/quantize_layer.py b/tzrec/modules/sid/quantize_layer.py
new file mode 100644
index 000000000..e7f344fda
--- /dev/null
+++ b/tzrec/modules/sid/quantize_layer.py
@@ -0,0 +1,58 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""QuantizeLayer: the per-layer quantizer interface shared by SID backends."""
+
+from abc import abstractmethod
+
+import torch
+from torch import nn
+
+from tzrec.modules.sid.types import QuantizeOutput
+
+
+class QuantizeLayer(nn.Module):
+    """One quantize layer: assign inputs to a codebook and look codes up.
+
+    Shared interface for the K-Means backend
+    (:class:`~tzrec.modules.sid.kmeans_quantize.KMeansQuantizeLayer`) and the RQ-VAE
+    backend's vector-quantize layer, so the residual quantizer can drive either
+    uniformly. Owns the codebook shape; subclasses build the backend-specific
+    codebook (a buffer, an ``nn.Embedding``, …) from it.
+
+    Args:
+        n_embed (int): number of codebook entries.
+        embed_dim (int): feature dimension.
+    """
+
+    def __init__(self, n_embed: int, embed_dim: int) -> None:
+        super().__init__()
+        self.n_embed = n_embed
+        self.embed_dim = embed_dim
+
+    @abstractmethod
+    def quantize(self, x: torch.Tensor, temperature: float = 1.0) -> QuantizeOutput:
+        """Assign ``x`` (B, D) to the codebook, returning codes + embeddings."""
+        raise NotImplementedError
+
+    def lookup(self, ids: torch.Tensor) -> torch.Tensor:
+        """Gather codebook embeddings for ``ids`` (indexes the codebook)."""
+        return self.get_codebook_embeddings()[ids]
+
+    @abstractmethod
+    def get_codebook_embeddings(self) -> torch.Tensor:
+        """Return the full codebook, shape (n_embed, embed_dim).
+
+        The codebook lives in a backend-specific attribute (a ``centroids``
+        buffer for K-Means, an ``nn.Embedding`` for RQ-VAE), so this stays
+        abstract; :meth:`lookup` is then concrete in terms of it.
+        """
+        raise NotImplementedError
diff --git a/tzrec/modules/sid/quantize_layer_test.py b/tzrec/modules/sid/quantize_layer_test.py
new file mode 100644
index 000000000..28eb4849b
--- /dev/null
+++ b/tzrec/modules/sid/quantize_layer_test.py
@@ -0,0 +1,81 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import torch
+
+from tzrec.modules.sid.quantize_layer import QuantizeLayer
+from tzrec.modules.sid.types import QuantizeOutput
+
+
+class _StubQuantizeLayer(QuantizeLayer):
+    """Minimal concrete subclass: a fixed codebook, nearest-row assignment.
+
+    Exercises the base class's concrete ``__init__`` / ``lookup`` without
+    pulling in a backend (FAISS / nn.Embedding).
+    """
+
+    def __init__(self, n_embed: int, embed_dim: int) -> None:
+        super().__init__(n_embed, embed_dim)
+        # A deterministic codebook so lookup/quantize are checkable by hand.
+        self._codebook = torch.arange(n_embed * embed_dim, dtype=torch.float32).reshape(
+            n_embed, embed_dim
+        )
+
+    def quantize(self, x: torch.Tensor, temperature: float = 1.0) -> QuantizeOutput:
+        dist = torch.cdist(x, self._codebook)
+        ids = dist.argmin(dim=-1)
+        return QuantizeOutput(embeddings=self.lookup(ids), ids=ids)
+
+    def get_codebook_embeddings(self) -> torch.Tensor:
+        return self._codebook
+
+
+class QuantizeLayerTest(unittest.TestCase):
+    """Tests for the shared QuantizeLayer base class."""
+
+    def test_init_stores_codebook_shape(self) -> None:
+        layer = _StubQuantizeLayer(n_embed=4, embed_dim=3)
+        self.assertEqual(layer.n_embed, 4)
+        self.assertEqual(layer.embed_dim, 3)
+
+    def test_lookup_gathers_codebook_rows(self) -> None:
+        layer = _StubQuantizeLayer(n_embed=4, embed_dim=3)
+        ids = torch.tensor([0, 2, 3, 1])
+        out = layer.lookup(ids)
+        torch.testing.assert_close(out, layer.get_codebook_embeddings()[ids])
+        self.assertEqual(out.shape, (4, 3))
+
+    def test_quantize_assigns_exact_codebook_rows(self) -> None:
+        # Feeding codebook rows back in must recover their own indices.
+        layer = _StubQuantizeLayer(n_embed=4, embed_dim=3)
+        x = layer.get_codebook_embeddings().clone()
+        out = layer.quantize(x)
+        torch.testing.assert_close(out.ids, torch.arange(4))
+        torch.testing.assert_close(out.embeddings, x)
+
+    def test_abstract_methods_unoverridden_raise(self) -> None:
+        # The abstract methods are documented to raise if a subclass forgets
+        # to implement them; QuantizeLayer relies on nn.Module (no ABCMeta),
+        # so this guards that the bodies still fail loudly rather than no-op.
+        class _Incomplete(QuantizeLayer):
+            pass
+
+        layer = _Incomplete(n_embed=2, embed_dim=2)
+        with self.assertRaises(NotImplementedError):
+            layer.get_codebook_embeddings()
+        with self.assertRaises(NotImplementedError):
+            layer.quantize(torch.zeros(1, 2))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tzrec/modules/sid/residual_kmeans_quantizer.py b/tzrec/modules/sid/residual_kmeans_quantizer.py
new file mode 100644
index 000000000..11b06951c
--- /dev/null
+++ b/tzrec/modules/sid/residual_kmeans_quantizer.py
@@ -0,0 +1,261 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Multi-layer residual K-Means: ResidualKMeansQuantizer.
+
+Training is FAISS-only: the codebook is built once via ``train_offline``
+over the full embedding matrix; ``forward`` is read-only (predict + lookup).
+"""
+
+from typing import Dict, List, Optional, Tuple, Union
+
+import faiss
+import faiss.contrib.torch_utils  # noqa: F401  (registers torch tensor I/O)
+import torch
+from torch import nn
+from torch.nn import functional as F
+
+from tzrec.modules.sid.kmeans_quantize import KMeansQuantizeLayer
+from tzrec.modules.sid.residual_quantizer import ResidualQuantizer
+from tzrec.utils.logging_util import logger
+
+
+class ResidualKMeansQuantizer(ResidualQuantizer):
+    """Multi-layer residual K-Means with offline FAISS training.
+
+    Each layer quantizes the residual from the previous layer:
+        residual_0 = input
+        for each layer i:
+            (optionally) residual_i = L2_normalize(residual_i)
+            code_i, quantized_i = layer_i.quantize(residual_i)
+            residual_{i+1} = residual_i - quantized_i
+        output = sum of all quantized_i
+
+    Semantic ID = (code_0, code_1, ..., code_{n_layers-1})
+
+    Args:
+        embed_dim (int): feature dimension.
+        n_layers (int): number of residual quantization layers.
+        n_embed (int|List[int]): number of clusters per layer. Default: 256.
+            May differ per layer (non-uniform codebooks such as
+            ``[256, 512, 1024]`` are supported) — ``train_offline`` builds a
+            separate ``faiss.Kmeans`` per layer.
+        normalize_residuals (bool): whether to L2-normalize residuals
+            before each layer. Default: False, matching the ``SidRqkmeans``
+            proto default (and OpenOneRec's residual k-means, which fits raw
+            residuals with no per-layer normalization).
+        faiss_kmeans_kwargs (Dict|None): extra kwargs forwarded to
+            ``faiss.Kmeans(D, K, **kwargs)`` (e.g. {'niter': 20,
+            'verbose': True, 'spherical': False}). A ``gpu`` key is ignored —
+            the fit is CPU-only.
+    """
+
+    def __init__(
+        self,
+        embed_dim: int,
+        n_layers: int,
+        n_embed: Union[int, List[int]] = 256,
+        normalize_residuals: bool = False,
+        faiss_kmeans_kwargs: Optional[Dict] = None,
+    ) -> None:
+        super().__init__(embed_dim, n_layers, n_embed, normalize_residuals)
+        self.faiss_kmeans_kwargs = dict(faiss_kmeans_kwargs or {})
+
+        self.layers = nn.ModuleList(
+            [
+                KMeansQuantizeLayer(
+                    n_embed=self.n_embed_list[i],
+                    embed_dim=embed_dim,
+                )
+                for i in range(n_layers)
+            ]
+        )
+
+    def _quantize_layer(
+        self,
+        layer_idx: int,
+        residual: torch.Tensor,
+        temperature: float = 1.0,
+    ) -> Tuple[torch.Tensor, torch.Tensor]:
+        """Nearest-centroid assignment for one layer (delegates to the layer).
+
+        Uninitialized layers (before ``train_offline``) return zeros, so the
+        residual walk is a no-op and the model stays callable.
+
+        Args:
+            layer_idx (int): quantization layer index.
+            residual (Tensor): current residual, shape (B, D).
+            temperature (float): unused (no soft assignment).
+
+        Returns:
+            codes (Tensor): cluster indices, shape (B,).
+            quantized (Tensor): selected centroids, shape (B, D).
+        """
+        out = self.layers[layer_idx].quantize(residual, temperature)
+        return out.ids, out.embeddings
+
+    def forward(self, input: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
+        """Assign codes per layer and sum the centroids.
+
+        Codebook is read-only here; training happens in ``train_offline``.
+        Uninitialized layers contribute zeros (see :meth:`_quantize_layer`) so
+        the model is callable before the one-shot FAISS fit completes.
+
+        Args:
+            input (Tensor): input embeddings, shape (B, D).
+
+        Returns:
+            codes (Tensor): cluster indices per layer, shape (B, n_layers).
+            quantized (Tensor): sum of quantized embeddings, shape (B, D).
+        """
+        cluster_ids, quantized_sum, _ = self._residual_pass(input)
+        return cluster_ids, quantized_sum
+
+    @property
+    def is_fitted(self) -> bool:
+        """Whether ``train_offline`` has populated every layer's codebook.
+
+        ``forward`` is callable before the fit (uninitialized layers emit
+        zeros), so reconstruction outputs are meaningful only once this is True.
+        """
+        return all(layer.is_initialized for layer in self.layers)
+
+    @torch.no_grad()
+    def get_codebook_embeddings(self, layer_idx: int) -> torch.Tensor:
+        """Get centroid weights for a specific layer.
+
+        Args:
+            layer_idx (int): index of the quantization layer.
+
+        Returns:
+            Tensor: centroids, shape (n_embed, embed_dim).
+        """
+        return self.layers[layer_idx].get_codebook_embeddings()
+
+    def _lookup_code(self, layer_idx: int, code_idx: torch.Tensor) -> torch.Tensor:
+        """Look up codebook vectors via the layer's centroid table."""
+        return self.layers[layer_idx].lookup(code_idx)
+
+    def default_fit_sample_size(self) -> int:
+        """Points the FAISS fit subsamples to: max(K) * max_points_per_centroid.
+
+        ``faiss.Kmeans`` caps each layer's training set at
+        ``K * max_points_per_centroid`` (default 256), so fitting on more is
+        wasted. Callers use this to size their training-sample reservoir.
+        """
+        max_ppc = int(self.faiss_kmeans_kwargs.get("max_points_per_centroid", 256))
+        return max(self.n_embed_list) * max_ppc
+
+    @torch.no_grad()
+    def train_offline(
+        self,
+        inputs: torch.Tensor,
+        verbose: bool = True,
+    ) -> None:
+        """Train the multi-layer codebook via offline FAISS K-Means.
+
+        CPU-only: ``inputs`` is already a host tensor (SidRqkmeans refuses to
+        run when CUDA is visible) and the FAISS fit runs on CPU. The post-fit
+        ``index.search`` assignment streams all N rows through in
+        ``SEARCH_CHUNK``-sized chunks to cap peak memory.
+
+        Args:
+            inputs (Tensor): embedding matrix (N, D) on CPU. CONSUMED: the
+                residual pass may mutate it in place, so the caller must not
+                rely on its contents afterward (copy first if it needs them).
+            verbose (bool): print per-layer reconstruction loss. Default: True.
+        """
+        # Host-tensor contract, checked here (not deep in faiss). raise (not
+        # assert): these guard data corruption and must survive ``python -O``.
+        if inputs.is_cuda:
+            raise RuntimeError("train_offline is CPU-only; got a CUDA tensor")
+        if inputs.dim() != 2 or inputs.shape[1] != self.embed_dim:
+            raise RuntimeError(
+                f"inputs must be (N, {self.embed_dim}), got {tuple(inputs.shape)}"
+            )
+        # The loop below mutates x in place (``x -= q``); the dtype/layout
+        # normalize is a no-op view when already float32 + contiguous, so the
+        # mutation lands in the caller's buffer (intended — see Args: CONSUMED).
+        x = inputs.detach().to(dtype=torch.float32).contiguous()
+        N = x.shape[0]
+        # Clear message before faiss's own opaque C++ throw for N < K. (The
+        # K <= N < K * min_points_per_centroid case, where faiss only warns and
+        # returns a degenerate codebook, is not guarded here.)
+        max_k = max(self.n_embed_list)
+        if N < max_k:
+            raise RuntimeError(
+                f"need >= {max_k} points to fit the codebook (largest layer K), "
+                f"got N={N}"
+            )
+        out = torch.zeros_like(x)
+        # x0 (original input) feeds the per-layer recon log. Without
+        # normalization ``out + x == x0``, so it's rebuilt on the fly below and
+        # the persistent (N, D) clone is skipped; normalization rescales x and
+        # breaks that invariant, so clone then.
+        x0 = x.clone() if (verbose and self.normalize_residuals) else None
+
+        # CPU-only fit (SidRqkmeans refuses CUDA). Drop any stale ``gpu`` kwarg
+        # so a faiss-gpu build can't target an absent GPU.
+        kwargs = dict(self.faiss_kmeans_kwargs)
+        kwargs.pop("gpu", None)
+        if verbose:
+            logger.info(
+                "[ResidualKMeansQuantizer] fitting %d-layer codebook on CPU "
+                "(N=%d, D=%d).",
+                self.n_layers,
+                N,
+                self.embed_dim,
+            )
+
+        # Chunk index.search to cap peak memory (~1 GB at 500K × 512 × 4B).
+        SEARCH_CHUNK = 500_000
+
+        for layer_idx in range(self.n_layers):
+            if self.normalize_residuals:
+                x = F.normalize(x, dim=-1)
+
+            # Fresh Kmeans per layer so each can use its own K (non-uniform
+            # codebooks).
+            kmeans = faiss.Kmeans(
+                self.embed_dim, self.n_embed_list[layer_idx], **kwargs
+            )
+            kmeans.train(x)
+            centroids = torch.as_tensor(kmeans.centroids, dtype=torch.float32)
+
+            for start in range(0, N, SEARCH_CHUNK):
+                end = min(start + SEARCH_CHUNK, N)
+                _, idx = kmeans.index.search(x[start:end], 1)
+                idx = torch.as_tensor(idx).reshape(-1).long()
+                q = centroids[idx]  # (chunk, D)
+                out[start:end] += q
+                x[start:end] -= q  # residual
+                del idx, q
+
+            if verbose:
+                # x0 == out + x without normalization (see above).
+                ref = x0 if self.normalize_residuals else out + x
+                logger.info(
+                    "[ResidualKMeansQuantizer][offline_faiss][layer %d] %s",
+                    layer_idx,
+                    self._calc_loss(ref, out),  # cumulative recon of original input
+                )
+
+            self.layers[layer_idx].load_centroids_(centroids)
+            if verbose:
+                logger.info(
+                    "[ResidualKMeansQuantizer][offline_faiss] layer %d finished",
+                    layer_idx,
+                )
+
+    @staticmethod
+    def _calc_loss(x: torch.Tensor, out: torch.Tensor) -> Dict[str, float]:
+        """Per-layer reconstruction MSE for the offline-fit log."""
+        return {"loss": float(((out - x) ** 2).mean().item())}
diff --git a/tzrec/modules/sid/residual_kmeans_quantizer_test.py b/tzrec/modules/sid/residual_kmeans_quantizer_test.py
new file mode 100644
index 000000000..265991143
--- /dev/null
+++ b/tzrec/modules/sid/residual_kmeans_quantizer_test.py
@@ -0,0 +1,126 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import torch
+
+from tzrec.modules.sid.residual_kmeans_quantizer import (
+    ResidualKMeansQuantizer,
+)
+from tzrec.modules.sid.residual_quantizer import (
+    ResidualQuantizer,
+)
+
+
+class ResidualKMeansQuantizerTest(unittest.TestCase):
+    def test_is_subclass(self) -> None:
+        rkq = ResidualKMeansQuantizer(embed_dim=4, n_layers=2, n_embed=8)
+        self.assertIsInstance(rkq, ResidualQuantizer)
+
+    def test_non_uniform_codebook_supported(self) -> None:
+        rkq = ResidualKMeansQuantizer(embed_dim=4, n_layers=3, n_embed=[8, 4, 16])
+        self.assertEqual(rkq.n_embed_list, [8, 4, 16])
+        self.assertEqual([layer.centroids.shape[0] for layer in rkq.layers], [8, 4, 16])
+
+    def test_train_offline_raises_on_too_few_points(self) -> None:
+        """N < largest K fails fast (clear message before faiss's own throw)."""
+        rkq = ResidualKMeansQuantizer(embed_dim=4, n_layers=1, n_embed=8)
+        with self.assertRaisesRegex(RuntimeError, "largest layer K"):
+            rkq.train_offline(torch.randn(4, 4), verbose=False)
+
+    def test_train_offline_raises_on_wrong_dim(self) -> None:
+        """An input whose width != embed_dim fails fast."""
+        rkq = ResidualKMeansQuantizer(embed_dim=4, n_layers=1, n_embed=8)
+        with self.assertRaisesRegex(RuntimeError, "inputs must be"):
+            rkq.train_offline(torch.randn(16, 8), verbose=False)
+
+    def test_forward_returns_zeros_before_fit(self) -> None:
+        rkq = ResidualKMeansQuantizer(embed_dim=4, n_layers=2, n_embed=8)
+        self.assertFalse(all(layer.is_initialized for layer in rkq.layers))
+        codes, quantized = rkq(torch.randn(5, 4))
+        self.assertEqual(codes.shape, (5, 2))
+        self.assertEqual(quantized.shape, (5, 4))
+
+    def test_forward_is_fx_traceable(self) -> None:
+        """Predict forward must FX-trace.
+
+        torchrec's inference pipeline symbolically traces the model, so the
+        per-batch distance path must be free of data-dependent control flow.
+        """
+        import torch.fx as fx
+
+        torch.manual_seed(0)
+        rkq = ResidualKMeansQuantizer(embed_dim=4, n_layers=2, n_embed=8)
+        for layer in rkq.layers:  # populate centroids -> is_initialized=True
+            layer.load_centroids_(torch.randn(8, 4))
+        traced = fx.symbolic_trace(rkq)
+        x = torch.randn(5, 4)
+        c_eager, q_eager = rkq(x)
+        c_traced, q_traced = traced(x)
+        torch.testing.assert_close(c_traced, c_eager)
+        torch.testing.assert_close(q_traced, q_eager)
+
+    def test_train_offline_non_uniform(self) -> None:
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+        torch.manual_seed(0)
+        n_embed = [8, 4, 16]
+        rkq = ResidualKMeansQuantizer(
+            embed_dim=4, n_layers=3, n_embed=n_embed, faiss_kmeans_kwargs={"niter": 5}
+        )
+        rkq.train_offline(torch.randn(512, 4), verbose=False)
+        self.assertTrue(all(layer.is_initialized for layer in rkq.layers))
+        # Each layer fit its own K centroids; codes stay in per-layer range.
+        codes, _ = rkq(torch.randn(7, 4))
+        self.assertEqual(codes.shape, (7, 3))
+        for i, k in enumerate(n_embed):
+            self.assertTrue((codes[:, i] >= 0).all() and (codes[:, i] < k).all())
+
+    def test_train_offline_then_decode(self) -> None:
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+        torch.manual_seed(0)
+        rkq = ResidualKMeansQuantizer(
+            embed_dim=4, n_layers=2, n_embed=8, faiss_kmeans_kwargs={"niter": 5}
+        )
+        rkq.train_offline(torch.randn(256, 4), verbose=False)
+        self.assertTrue(all(layer.is_initialized for layer in rkq.layers))
+
+        codes, _ = rkq(torch.randn(5, 4))
+        self.assertTrue((codes >= 0).all() and (codes < 8).all())
+        recon = rkq.decode_codes(codes)  # inherited from the base
+        self.assertEqual(recon.shape, (5, 4))
+
+    def test_forward_get_codes_consistent(self) -> None:
+        """Forward ids and get_codes both route through the shared walk."""
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+        torch.manual_seed(0)
+        rkq = ResidualKMeansQuantizer(
+            embed_dim=4, n_layers=3, n_embed=8, faiss_kmeans_kwargs={"niter": 5}
+        )
+        rkq.train_offline(torch.randn(256, 4), verbose=False)
+        x = torch.randn(9, 4)
+        fwd_ids, fwd_quant = rkq(x)
+        torch.testing.assert_close(rkq.get_codes(x), fwd_ids)
+        # forward's residual-sum equals the centroid-sum reconstruction.
+        torch.testing.assert_close(fwd_quant, rkq.decode_codes(fwd_ids))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tzrec/modules/sid/types.py b/tzrec/modules/sid/types.py
new file mode 100644
index 000000000..2f0cf3c60
--- /dev/null
+++ b/tzrec/modules/sid/types.py
@@ -0,0 +1,28 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Data types for SID generation: output tuples shared across quantizers."""
+
+from typing import NamedTuple
+
+import torch
+
+
+class QuantizeOutput(NamedTuple):
+    """One quantize layer's output.
+
+    Attributes:
+        embeddings (Tensor): quantized embeddings, shape (B, D).
+        ids (Tensor): codebook indices, shape (B,).
+    """
+
+    embeddings: torch.Tensor
+    ids: torch.Tensor
diff --git a/tzrec/protos/model.proto b/tzrec/protos/model.proto
index bef2062ea..58b719a7a 100644
--- a/tzrec/protos/model.proto
+++ b/tzrec/protos/model.proto
@@ -5,6 +5,7 @@ import "tzrec/protos/models/rank_model.proto";
 import "tzrec/protos/models/multi_task_rank.proto";
 import "tzrec/protos/models/match_model.proto";
 import "tzrec/protos/models/general_rank_model.proto";
+import "tzrec/protos/models/sid_model.proto";
 import "tzrec/protos/loss.proto";
 import "tzrec/protos/metric.proto";
 import "tzrec/protos/seq_encoder.proto";
@@ -76,6 +77,10 @@ message ModelConfig {
         TDM tdm = 400;
 
         RocketLaunching rocket_launching = 500;
+
+        // SID generation models
+        // (600 is reserved for SidRqvae, arriving in the follow-up PR)
+        SidRqkmeans sid_rqkmeans = 601;
     }
 
     optional uint32 num_class = 2 [default = 1];
diff --git a/tzrec/protos/models/sid_model.proto b/tzrec/protos/models/sid_model.proto
new file mode 100644
index 000000000..e51462efa
--- /dev/null
+++ b/tzrec/protos/models/sid_model.proto
@@ -0,0 +1,42 @@
+syntax = "proto2";
+package tzrec.protos;
+
+// Strictly-typed subset of faiss.Kmeans(D, K, **kwargs) knobs. Unset fields
+// fall back to faiss's own defaults (so it is safe to leave partially set).
+// ``gpu`` is intentionally omitted — the fit is CPU-only (SidRqkmeans refuses
+// a visible CUDA device).
+message FaissKmeansConfig {
+    optional uint32 niter = 1;
+    optional uint32 nredo = 2;
+    optional uint32 seed = 3;
+    optional uint32 max_points_per_centroid = 4;
+    optional uint32 min_points_per_centroid = 5;
+    optional bool spherical = 6;
+    optional bool verbose = 7;
+}
+
+message SidRqkmeans {
+    // Input embedding dimension (K-Means runs directly on raw embeddings,
+    // no encoder).
+    optional uint32 input_dim = 1 [default = 512];
+    // Per-layer cluster counts, e.g. [256, 256, 256].
+    // List length is the number of residual quantization layers. Entries
+    // may differ per layer (non-uniform codebooks such as [256, 512, 1024]
+    // are supported — the FAISS backend fits a separate ``faiss.Kmeans``
+    // per layer).
+    repeated uint32 codebook = 3;
+    // L2-normalize residuals before each layer.
+    optional bool normalize_residuals = 4 [default = false];
+    // Strictly-typed extra kwargs forwarded to faiss.Kmeans(D, K, **kwargs).
+    optional FaissKmeansConfig faiss_kmeans_kwargs = 5;
+    // Target number of embeddings to reservoir-sample for the FAISS fit.
+    // Bounds host memory regardless of corpus
+    // size. 0 (the default) auto-derives it as max(K) * max_points_per_centroid
+    // (the largest per-layer codebook, for non-uniform codebooks) — exactly
+    // what FAISS subsamples to internally (default 256), so no training points
+    // are wasted.
+    optional uint32 train_sample_size = 6 [default = 0];
+
+    // Name of the item embedding feature inside the input Batch.
+    optional string embedding_feature_name = 40 [default = "item_emb"];
+}
diff --git a/tzrec/tests/configs/sid_rqkmeans_mock.config b/tzrec/tests/configs/sid_rqkmeans_mock.config
new file mode 100644
index 000000000..0e6dec907
--- /dev/null
+++ b/tzrec/tests/configs/sid_rqkmeans_mock.config
@@ -0,0 +1,56 @@
+train_input_path: ""
+eval_input_path: ""
+model_dir: "experiments/sid_rqkmeans_mock"
+train_config {
+    sparse_optimizer {
+        adagrad_optimizer {
+            lr: 0.001
+        }
+        constant_learning_rate {
+        }
+    }
+    dense_optimizer {
+        adam_optimizer {
+            lr: 0.001
+        }
+        constant_learning_rate {
+        }
+    }
+    num_epochs: 1
+    save_checkpoints_steps: 0
+    save_checkpoints_epochs: 0
+}
+eval_config {
+}
+data_config {
+    batch_size: 256
+    dataset_type: ParquetDataset
+    fg_mode: FG_DAG
+    num_workers: 2
+}
+feature_configs {
+    raw_feature {
+        feature_name: "item_emb"
+        expression: "item:embedding"
+        value_dim: 16
+    }
+}
+model_config {
+    feature_groups {
+        group_name: "deep"
+        feature_names: "item_emb"
+        group_type: DEEP
+    }
+    sid_rqkmeans {
+        input_dim: 16
+        codebook: 16
+        codebook: 16
+        codebook: 16
+        normalize_residuals: false
+        embedding_feature_name: "item_emb"
+        faiss_kmeans_kwargs {
+            niter: 5
+            seed: 42
+        }
+    }
+}
diff --git a/tzrec/tests/sid_integration_test.py b/tzrec/tests/sid_integration_test.py
new file mode 100644
index 000000000..53f24a1d3
--- /dev/null
+++ b/tzrec/tests/sid_integration_test.py
@@ -0,0 +1,122 @@
+# Copyright (c) 2026, Alibaba Group;
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#    http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import glob
+import json
+import math
+import os
+import shutil
+import tempfile
+import unittest
+from unittest import mock
+
+import numpy as np
+import pyarrow as pa
+import pyarrow.parquet as pq
+import torch
+
+from tzrec.tests import utils
+from tzrec.utils import config_util
+
+
+class SidIntegrationTest(unittest.TestCase):
+    def setUp(self):
+        self.success = False
+        if not os.path.exists("./tmp"):
+            os.makedirs("./tmp")
+        self.test_dir = tempfile.mkdtemp(prefix="tzrec_", dir="./tmp")
+        os.chmod(self.test_dir, 0o755)
+        # SidRqkmeans is single-process; pin nproc=1 (the CI harness defaults
+        # to 2, which would trip the world_size>1 guard).
+        patcher = mock.patch.dict(os.environ, {"TEST_NPROC_PER_NODE": "1"})
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def tearDown(self):
+        if self.success and os.path.exists(self.test_dir):
+            shutil.rmtree(self.test_dir)
+
+    def _prepare_config(self, num_rows: int, dim: int) -> str:
+        """Write an embedding parquet + a SID config pointed at it.
+
+        Single dense ``embedding`` column, no labels — SID reads the item
+        embedding straight from the batch. Returns the saved config path.
+        """
+        data_dir = os.path.join(self.test_dir, "sid_data")
+        os.makedirs(data_dir, exist_ok=True)
+        emb = np.random.rand(num_rows, dim).astype(np.float32)
+        pq.write_table(
+            pa.table({"embedding": pa.array(list(emb))}),
+            os.path.join(data_dir, "part-0.parquet"),
+        )
+        data_glob = os.path.join(data_dir, "*.parquet")
+
+        # train_input_path set -> load_config_for_test uses it as-is (the
+        # FG_DAG auto-mock path is match-model-specific; SID is single-table).
+        config = config_util.load_pipeline_config(
+            "tzrec/tests/configs/sid_rqkmeans_mock.config"
+        )
+        config.train_input_path = data_glob
+        config.eval_input_path = data_glob
+        config_path = os.path.join(self.test_dir, "sid.config")
+        config_util.save_message(config, config_path)
+        return config_path
+
+    @unittest.skipIf(
+        torch.cuda.is_available(),
+        "SidRqkmeans is CPU-only; this end-to-end test runs on the CPU CI job. "
+        "Forcing CPU on a CUDA-built (GPU) image is unreliable.",
+    )
+    def test_sid_rqkmeans_train_eval(self):
+        """End-to-end train -> on_train_end FAISS fit -> checkpoint -> eval.
+
+        Locks down the load-bearing path: the codebook exists only after
+        ``on_train_end``, which forces the final checkpoint; the post-fit eval
+        then reports finite reconstruction metrics.
+        """
+        try:
+            import faiss  # noqa: F401
+        except ImportError:
+            self.skipTest("faiss not installed")
+
+        config_path = self._prepare_config(num_rows=2048, dim=16)
+
+        self.success = utils.test_train_eval(config_path, self.test_dir)
+        if self.success:
+            self.success = utils.test_eval(
+                os.path.join(self.test_dir, "pipeline.config"), self.test_dir
+            )
+        self.assertTrue(self.success)
+        # on_train_end fitted the codebook and forced a final checkpoint.
+        self.assertTrue(
+            glob.glob(os.path.join(self.test_dir, "train", "model.ckpt-*")),
+            "no checkpoint persisted after on_train_end",
+        )
+        # A fitted codebook yields finite metrics; a degenerate/unfitted one
+        # never exposes x_hat -> metrics stay NaN. So assert finiteness, plus
+        # rel_loss < 1.0 (all-zero baseline ~ 1.0) and nonzero SID variety.
+        result_path = os.path.join(self.test_dir, "train", "eval_result.txt")
+        self.assertTrue(os.path.exists(result_path), "no eval_result.txt produced")
+        with open(result_path) as f:
+            lines = [ln for ln in f.read().splitlines() if ln.strip()]
+        self.assertTrue(lines, "eval_result.txt is empty")
+        metrics = json.loads(lines[-1])
+        for key in ("mse", "rel_loss", "unique_sid_ratio"):
+            self.assertIn(key, metrics)
+            self.assertTrue(
+                math.isfinite(metrics[key]), f"{key} not finite: {metrics[key]}"
+            )
+        self.assertLess(metrics["rel_loss"], 1.0)
+        self.assertGreater(metrics["unique_sid_ratio"], 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tzrec/version.py b/tzrec/version.py
index 52c53fa4f..a5d3e8f5c 100644
--- a/tzrec/version.py
+++ b/tzrec/version.py
@@ -9,4 +9,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = "1.2.18"
+__version__ = "1.2.19"