diff --git a/nemo_rl/algorithms/loss/loss_functions.py b/nemo_rl/algorithms/loss/loss_functions.py
index c615de7199b..9e9cc872c0b 100755
--- a/nemo_rl/algorithms/loss/loss_functions.py
+++ b/nemo_rl/algorithms/loss/loss_functions.py
@@ -15,7 +15,7 @@
 from typing import Any, NotRequired, Optional, TypedDict, TypeVar
 
 import torch
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from nemo_rl.algorithms.loss.interfaces import (
     LossFunction,
@@ -126,7 +126,7 @@ class ClippedPGLossConfig(BaseModel, extra="allow"):
     ratio_clip_c: Optional[float] = None
 
     # --- KL regularization ---
-    reference_policy_kl_penalty: float = 0.01
+    reference_policy_kl_penalty: float = Field(default=0.01, ge=0, allow_inf_nan=False)
     # Can be set to k1, k2, k3
     # For more details, see http://joschu.net/blog/kl-approx.html
     reference_policy_kl_type: str = "k3"
diff --git a/nemo_rl/algorithms/single_controller.py b/nemo_rl/algorithms/single_controller.py
index 405c9ec436a..aefae215a62 100644
--- a/nemo_rl/algorithms/single_controller.py
+++ b/nemo_rl/algorithms/single_controller.py
@@ -101,7 +101,7 @@
 from nemo_rl.data.interfaces import DatumSpec
 from nemo_rl.data_plane import DATA_PLANE_CHECKPOINT_SCHEMA_VERSION, KVBatchMeta
 from nemo_rl.data_plane.async_utils import call_data_plane
-from nemo_rl.data_plane.schema import DP_CALIB_INPUT_FIELDS
+from nemo_rl.data_plane.schema import DP_CALIB_INPUT_FIELDS, DP_TRAIN_FIELDS
 from nemo_rl.distributed.batched_data_dict import BatchedDataDict
 from nemo_rl.distributed.refit_watchdog import RefitAborted, is_refit_context_lost
 from nemo_rl.environments.nemo_gym import should_use_nemo_gym
@@ -149,6 +149,20 @@ def _pooled_opd_metrics(
     }
 
 
+def _train_fields_for_step(
+    *,
+    policy_logprobs_required: bool,
+    reference_logprobs_required: bool,
+) -> tuple[str, ...]:
+    """Return only the data-plane columns produced for this train step."""
+    return tuple(
+        field
+        for field in DP_TRAIN_FIELDS
+        if (policy_logprobs_required or field != "prev_logprobs")
+        and (reference_logprobs_required or field != "reference_policy_logprobs")
+    )
+
+
 @ray.remote(num_cpus=1, num_gpus=0)  # pragma: no cover
 class SingleControllerActor:
     """CPU-only Ray actor that orchestrates the RL training loop.
@@ -206,11 +220,17 @@ def __init__(
             master_config.loss_fn.force_on_policy_ratio
             and self._algo_cfg.seq_logprob_error_threshold is None
         )
+        # _build_trainer initializes the reference model only for a positive KL
+        # penalty, so the controller must use the same gate before requesting it.
         self._reference_logprobs_required = bool(
             master_config.loss_fn.reference_policy_kl_penalty > 0
             and not self._algo_cfg.skip_reference_policy_logprobs_calculation
         )
         self._teacher_logprobs_required = opd_module.is_opd_enabled(master_config)
+        self._train_fields = _train_fields_for_step(
+            policy_logprobs_required=self._policy_logprobs_required,
+            reference_logprobs_required=self._reference_logprobs_required,
+        )
         self._dp_client = actor_args.dp_client
         self._gen: Generation = actor_args.gen_handle
         self._trainer: TQPolicy = actor_args.trainer_handle
@@ -1874,6 +1894,7 @@ async def _train_pump(self) -> None:
                                     await asyncio.to_thread(
                                         self._trainer.train_microbatches_from_meta,
                                         train_meta,
+                                        train_fields=self._train_fields,
                                     )
                                     # A PPO step is one chunk: nothing to
                                     # accumulate, so close every epoch here.
diff --git a/nemo_rl/algorithms/single_controller_utils/config.py b/nemo_rl/algorithms/single_controller_utils/config.py
index 03b2998a0eb..88f0c65b5e5 100644
--- a/nemo_rl/algorithms/single_controller_utils/config.py
+++ b/nemo_rl/algorithms/single_controller_utils/config.py
@@ -1020,6 +1020,21 @@ def validate_single_controller_config(master_config: MasterConfig) -> None:
             "loss_fn.reference_policy_kl_penalty=0."
         )
 
+    if (
+        master_config.loss_fn.use_kl_in_reward
+        and reference_policy_kl_penalty > 0
+        and master_config.loss_fn.force_on_policy_ratio
+        and algo_cfg.seq_logprob_error_threshold is None
+    ):
+        raise ValueError(
+            "loss_fn.use_kl_in_reward=true with a nonzero "
+            "loss_fn.reference_policy_kl_penalty requires policy logprobs, but "
+            "loss_fn.force_on_policy_ratio=true without "
+            "seq_logprob_error_threshold skips them. Set "
+            "loss_fn.force_on_policy_ratio=false or configure "
+            "seq_logprob_error_threshold."
+        )
+
     # ``env`` is required in production configs, but model_construct-based unit
     # configs can omit it. Only apply rollout-path validation when it is present.
     env_config = getattr(master_config, "env", None)
diff --git a/nemo_rl/models/policy/tq_policy.py b/nemo_rl/models/policy/tq_policy.py
index 9f68526a937..bd332e54195 100644
--- a/nemo_rl/models/policy/tq_policy.py
+++ b/nemo_rl/models/policy/tq_policy.py
@@ -438,6 +438,7 @@ def train_microbatches_from_meta(
         self,
         meta: KVBatchMeta,
         timer: Optional[Timer] = None,
+        train_fields: tuple[str, ...] = DP_TRAIN_FIELDS,
     ) -> None:
         """Dispatch one meta slice (DP-sharded) into an open train step.
 
@@ -452,12 +453,17 @@ def train_microbatches_from_meta(
         ``.grad``. Returns nothing — per-microbatch metrics accumulate in
         the workers' open-step state and surface once via
         :meth:`finish_train_step`.
+
+        Args:
+            meta: Data-plane metadata for the samples in this chunk.
+            timer: Optional timer for nested policy-training measurements.
+            train_fields: Columns produced for this step and fetched by workers.
         """
         spa, dba = self._packing_args("train_mb_tokens")
         train_meta = self._isolated_meta(
             meta,
             fields=fields_with_optional_routed_experts(
-                DP_TRAIN_FIELDS, enabled=self._router_replay_enabled
+                train_fields, enabled=self._router_replay_enabled
             ),
             task_name="train",
         )
diff --git a/tests/unit/algorithms/test_loss_functions.py b/tests/unit/algorithms/test_loss_functions.py
index 12aa3f4d4d5..3b672ac1f42 100644
--- a/tests/unit/algorithms/test_loss_functions.py
+++ b/tests/unit/algorithms/test_loss_functions.py
@@ -43,6 +43,17 @@
 )
 
 
+@pytest.mark.parametrize(
+    "invalid_penalty",
+    [-0.01, float("nan"), float("inf"), float("-inf")],
+)
+def test_clipped_pg_loss_config_rejects_invalid_reference_kl_penalty(
+    invalid_penalty: float,
+) -> None:
+    with pytest.raises(ValueError):
+        ClippedPGLossConfig(reference_policy_kl_penalty=invalid_penalty)
+
+
 def setup_dpo_loss_test_data(vocab_size=16, batch_size=1):
     seq_len = 4
     data = {
diff --git a/tests/unit/models/policy/test_split_api_wrappers.py b/tests/unit/models/policy/test_split_api_wrappers.py
index 34ccd1c1a9c..b5db49e5601 100644
--- a/tests/unit/models/policy/test_split_api_wrappers.py
+++ b/tests/unit/models/policy/test_split_api_wrappers.py
@@ -164,6 +164,27 @@ def test_train_microbatches_from_meta_dispatches_and_returns_none(self):
         # get_all_worker_results (unlike the single-data fan-outs)
         wg.get_all_worker_results.assert_called_once()
 
+    def test_train_microbatches_fetches_only_requested_fields(self):
+        p, _ = _make_tq_policy()
+        meta = _meta()
+        train_fields = tuple(
+            field
+            for field in DP_TRAIN_FIELDS
+            if field not in {"prev_logprobs", "reference_policy_logprobs"}
+        )
+        with (
+            patch.object(TQPolicy, "_stamp_pad_seqlen"),
+            patch.object(TQPolicy, "_packing_args", return_value=(None, None)),
+            patch(
+                "nemo_rl.models.policy.tq_policy.shard_meta_for_dp",
+                return_value=([meta, meta], None),
+            ) as mock_shard,
+        ):
+            p.train_microbatches_from_meta(meta, train_fields=train_fields)
+
+        train_meta = mock_shard.call_args.args[0]
+        assert train_meta.fields == list(train_fields)
+
     def test_train_microbatches_requests_routed_experts_for_router_replay(self):
         p, _ = _make_tq_policy()
         p._router_replay_enabled = True
diff --git a/tests/unit/single_controller/test_checkpointing.py b/tests/unit/single_controller/test_checkpointing.py
index 69b5caada71..c2070b0f8c6 100644
--- a/tests/unit/single_controller/test_checkpointing.py
+++ b/tests/unit/single_controller/test_checkpointing.py
@@ -126,7 +126,9 @@ def prepare_for_training(self) -> None:
     def begin_train_step(self, loss_fn: Any) -> None:
         pass
 
-    def train_microbatches_from_meta(self, meta: KVBatchMeta) -> None:
+    def train_microbatches_from_meta(
+        self, meta: KVBatchMeta, *, train_fields: tuple[str, ...]
+    ) -> None:
         pass
 
     def finish_train_step(self) -> dict[str, Any]:
diff --git a/tests/unit/single_controller/test_resiliency_config.py b/tests/unit/single_controller/test_resiliency_config.py
index a080bfe9050..6b1cb1fdbd7 100644
--- a/tests/unit/single_controller/test_resiliency_config.py
+++ b/tests/unit/single_controller/test_resiliency_config.py
@@ -21,7 +21,6 @@
 """
 
 import warnings
-from types import SimpleNamespace
 from typing import get_args
 
 import pytest
@@ -29,6 +28,7 @@
 
 from nemo_rl.algorithms.async_utils.staleness_sampler import SamplerConfig
 from nemo_rl.algorithms.grpo import GRPOConfig
+from nemo_rl.algorithms.loss import ClippedPGLossConfig
 from nemo_rl.algorithms.single_controller_utils.config import (
     AsyncRLConfig,
     FleetHealthConfig,
@@ -68,9 +68,7 @@ def _master_config(*, num_prompts_per_step: int = 8, **async_kwargs) -> MasterCo
             "train_global_batch_size": num_prompts_per_step * 4,
             "generation": {"colocated": {"enabled": False}},
         },
-        # The last two are read only on the ready_first branch, which rejects a run
-        # without them before it reaches anything under test here.
-        loss_fn=SimpleNamespace(
+        loss_fn=ClippedPGLossConfig(
             reference_policy_kl_penalty=0,
             use_importance_sampling_correction=True,
             force_on_policy_ratio=False,
@@ -409,7 +407,7 @@ def _master_config(*, use_nemo_gym: bool, rollout_failure: dict) -> MasterConfig
                 "train_global_batch_size": 8,
                 "generation": {"colocated": {"enabled": False}},
             },
-            loss_fn=SimpleNamespace(reference_policy_kl_penalty=0),
+            loss_fn=ClippedPGLossConfig(reference_policy_kl_penalty=0),
             env={"should_use_nemo_gym": use_nemo_gym},
             # Read by the metric_name check upstream #3429 added to this same
             # validator, which runs before the wrong-path check under test.
diff --git a/tests/unit/single_controller/test_setup.py b/tests/unit/single_controller/test_setup.py
index f1ad9957a15..2b940e2a723 100644
--- a/tests/unit/single_controller/test_setup.py
+++ b/tests/unit/single_controller/test_setup.py
@@ -427,6 +427,36 @@ def test_single_controller_mopd_recipe_resolves_to_runtime_contract():
     )
 
 
+@pytest.mark.parametrize(
+    ("reference_policy_kl_penalty", "expected_init_reference_model"),
+    [(0.0, False), (0.01, True)],
+)
+def test_build_trainer_initializes_reference_model_only_for_nonzero_kl(
+    reference_policy_kl_penalty: float,
+    expected_init_reference_model: bool,
+) -> None:
+    master_config = _make_master_config(
+        loss_cfg=ClippedPGLossConfig(
+            reference_policy_kl_penalty=reference_policy_kl_penalty
+        )
+    )
+
+    with patch.object(sc_setup_mod, "TQPolicy") as mock_policy:
+        sc_setup_mod._build_trainer(
+            MagicMock(name="train_cluster"),
+            master_config,
+            MagicMock(name="tokenizer"),
+            None,
+            weights_path=None,
+            optimizer_path=None,
+        )
+
+    assert (
+        mock_policy.call_args.kwargs["init_reference_model"]
+        is expected_init_reference_model
+    )
+
+
 class TestSetup:
     """setup arg validation + actor_args assembly."""
 
@@ -435,6 +465,35 @@ def test_raises_when_data_plane_disabled(self):
         with pytest.raises(ValueError, match="data_plane.enabled=True"):
             setup_single_controller(mc, MagicMock())
 
+    def test_nonzero_kl_rejects_skipping_reference_logprobs(self, patched_factories):
+        mc = _make_master_config(
+            loss_cfg=ClippedPGLossConfig(reference_policy_kl_penalty=0.01)
+        )
+        mc.grpo.skip_reference_policy_logprobs_calculation = True
+
+        with pytest.raises(ValueError, match="requires reference_policy_logprobs"):
+            setup_single_controller(mc, MagicMock(pad_token_id=0))
+
+        patched_factories["setup_response_data"].assert_not_called()
+        patched_factories["_build_clusters"].assert_not_called()
+        patched_factories["_build_trainer"].assert_not_called()
+
+    def test_reward_kl_rejects_skipping_policy_logprobs(self, patched_factories):
+        mc = _make_master_config(
+            loss_cfg=ClippedPGLossConfig(
+                reference_policy_kl_penalty=0.01,
+                use_kl_in_reward=True,
+                force_on_policy_ratio=True,
+            )
+        )
+
+        with pytest.raises(ValueError, match="requires policy logprobs"):
+            setup_single_controller(mc, MagicMock(pad_token_id=0))
+
+        patched_factories["setup_response_data"].assert_not_called()
+        patched_factories["_build_clusters"].assert_not_called()
+        patched_factories["_build_trainer"].assert_not_called()
+
     def test_rejects_mooncake_data_plane_checkpointing(self):
         mc = _make_master_config()
         mc.data_plane["backend"] = "mooncake_cpu"
diff --git a/tests/unit/single_controller/test_single_controller_actor.py b/tests/unit/single_controller/test_single_controller_actor.py
index d9c7d359b21..5e643f4ed69 100644
--- a/tests/unit/single_controller/test_single_controller_actor.py
+++ b/tests/unit/single_controller/test_single_controller_actor.py
@@ -223,11 +223,18 @@ def test_logs_hyperparameters_and_concrete_weight_synchronizer(
 
 
 @pytest.mark.parametrize(
-    ("reference_policy_kl_penalty", "skip_reference_logprobs", "expected_required"),
+    (
+        "reference_policy_kl_penalty",
+        "skip_reference_logprobs",
+        "force_on_policy_ratio",
+        "expected_policy_required",
+        "expected_reference_required",
+    ),
     [
-        (0.0, False, False),
-        (0.0, True, False),
-        (0.01, False, True),
+        (0.0, False, False, True, False),
+        (0.0, True, False, True, False),
+        (0.01, False, False, True, True),
+        (0.01, False, True, False, True),
     ],
 )
 def test_reference_logprobs_required_only_when_kl_enabled(
@@ -235,7 +242,9 @@ def test_reference_logprobs_required_only_when_kl_enabled(
     tmp_path,
     reference_policy_kl_penalty: float,
     skip_reference_logprobs: bool,
-    expected_required: bool,
+    force_on_policy_ratio: bool,
+    expected_policy_required: bool,
+    expected_reference_required: bool,
 ) -> None:
     """KL-disabled SingleController runs do not request reference logprobs."""
     monkeypatch.setattr(single_controller, "Logger", lambda _: MagicMock())
@@ -250,7 +259,7 @@ def test_reference_logprobs_required_only_when_kl_enabled(
             skip_reference_policy_logprobs_calculation=skip_reference_logprobs,
         ),
         loss_fn=ClippedPGLossConfig(
-            force_on_policy_ratio=False,
+            force_on_policy_ratio=force_on_policy_ratio,
             reference_policy_kl_penalty=reference_policy_kl_penalty,
         ),
         async_rl=AsyncRLConfig(
@@ -264,7 +273,12 @@ def test_reference_logprobs_required_only_when_kl_enabled(
 
     controller = _init_controller(master_config, _actor_args_for_init())
 
-    assert controller._reference_logprobs_required is expected_required
+    assert controller._policy_logprobs_required is expected_policy_required
+    assert controller._reference_logprobs_required is expected_reference_required
+    assert ("prev_logprobs" in controller._train_fields) is expected_policy_required
+    assert (
+        "reference_policy_logprobs" in controller._train_fields
+    ) is expected_reference_required
 
 
 @pytest.mark.parametrize("with_critic", [True, False], ids=["ppo", "grpo"])
@@ -928,6 +942,12 @@ async def select(self, **kwargs):
         return meta, 2 if num_groups else 0
 
 
+class _FullStepSampler(_OneThenEmptySampler):
+    async def select(self, **kwargs):
+        meta, num_groups = await super().select(**kwargs)
+        return meta, 2 if num_groups else 0
+
+
 class _ChunkedSampler(_EmptySampler):
     """Assembles one step out of several single-group chunks, then goes empty.
 
@@ -977,8 +997,10 @@ def prepare_for_training(self) -> None:
     def begin_train_step(self, loss_fn) -> None:
         del loss_fn
 
-    def train_microbatches_from_meta(self, meta: KVBatchMeta) -> None:
-        del meta
+    def train_microbatches_from_meta(
+        self, meta: KVBatchMeta, *, train_fields: tuple[str, ...]
+    ) -> None:
+        del meta, train_fields
 
     def finish_train_step(self) -> dict:
         return {}
@@ -1000,6 +1022,27 @@ def get_logprobs_from_meta(self, meta: KVBatchMeta) -> None:
         del meta
 
 
+class _LogprobRecordingTrainer(_NoOpTrainer):
+    def __init__(self) -> None:
+        self.policy_logprob_calls = 0
+        self.reference_logprob_calls = 0
+        self.train_fields_calls: list[tuple[str, ...]] = []
+
+    def get_logprobs_from_meta(self, meta: KVBatchMeta) -> None:
+        del meta
+        self.policy_logprob_calls += 1
+
+    def get_reference_policy_logprobs_from_meta(self, meta: KVBatchMeta) -> None:
+        del meta
+        self.reference_logprob_calls += 1
+
+    def train_microbatches_from_meta(
+        self, meta: KVBatchMeta, *, train_fields: tuple[str, ...]
+    ) -> None:
+        del meta
+        self.train_fields_calls.append(train_fields)
+
+
 class _OrderRecordingTrainer(_NoOpTrainer):
     """Records the policy lifecycle into a log shared with the critic double."""
 
@@ -1031,8 +1074,10 @@ def begin_train_step(self, loss_fn) -> None:
         del loss_fn
         self.calls.append("policy.begin_train_step")
 
-    def train_microbatches_from_meta(self, meta: KVBatchMeta) -> None:
-        del meta
+    def train_microbatches_from_meta(
+        self, meta: KVBatchMeta, *, train_fields: tuple[str, ...]
+    ) -> None:
+        del meta, train_fields
         self.calls.append("policy.train_microbatches_from_meta")
 
     def finish_train_step(self) -> dict:
@@ -1074,6 +1119,10 @@ def _train_pump_controller(*, sampler) -> object:
     ctrl._policy_logprobs_required = False
     ctrl._reference_logprobs_required = False
     ctrl._teacher_logprobs_required = False
+    ctrl._train_fields = single_controller._train_fields_for_step(
+        policy_logprobs_required=False,
+        reference_logprobs_required=False,
+    )
     ctrl._advantage_estimator = None
     ctrl._partition_id = "rollout_data"
     ctrl._sampler = sampler
@@ -1263,6 +1312,59 @@ def test_train_pump_prunes_stamps_older_than_the_step_that_just_closed(
     assert ctrl._batch_shortfall == {5: 1}
 
 
+@pytest.mark.parametrize(
+    (
+        "policy_logprobs_required",
+        "reference_logprobs_required",
+        "expected_policy_calls",
+        "expected_reference_calls",
+    ),
+    [
+        (False, False, 0, 0),
+        (True, False, 1, 0),
+        (False, True, 0, 1),
+        (True, True, 1, 1),
+    ],
+)
+def test_train_pump_requests_and_fetches_only_required_logprobs(
+    monkeypatch,
+    policy_logprobs_required: bool,
+    reference_logprobs_required: bool,
+    expected_policy_calls: int,
+    expected_reference_calls: int,
+) -> None:
+    meta = KVBatchMeta(
+        partition_id="rollout_data",
+        task_name="train",
+        sample_ids=["sample-0", "sample-1"],
+        fields=[],
+        sequence_lengths=[1, 1],
+        tags=[{"weight_version": 0}, {"weight_version": 0}],
+    )
+    ctrl = _train_pump_controller(sampler=_FullStepSampler(meta))
+    ctrl._policy_logprobs_required = policy_logprobs_required
+    ctrl._reference_logprobs_required = reference_logprobs_required
+    ctrl._train_fields = single_controller._train_fields_for_step(
+        policy_logprobs_required=policy_logprobs_required,
+        reference_logprobs_required=reference_logprobs_required,
+    )
+    trainer = _LogprobRecordingTrainer()
+    ctrl._trainer = trainer
+    ctrl._sync_weights = AsyncMock(return_value=1)
+    ctrl._logger = MagicMock()
+    monkeypatch.setattr(single_controller.ray, "cluster_resources", lambda: {})
+
+    asyncio.run(asyncio.wait_for(ctrl._train_pump(), timeout=1.0))
+
+    assert trainer.policy_logprob_calls == expected_policy_calls
+    assert trainer.reference_logprob_calls == expected_reference_calls
+    assert trainer.train_fields_calls == [ctrl._train_fields]
+    assert ("prev_logprobs" in ctrl._train_fields) is policy_logprobs_required
+    assert (
+        "reference_policy_logprobs" in ctrl._train_fields
+    ) is reference_logprobs_required
+
+
 def test_train_pump_rejects_step_with_no_valid_training_chunks() -> None:
     meta = KVBatchMeta(
         partition_id="rollout_data",
@@ -1327,7 +1429,9 @@ def test_train_pump_skips_empty_chunk_and_trains_later_valid_chunk(
 
     assert trainer.prepare_for_training.call_count == 2
     trainer.begin_train_step.assert_called_once_with(None)
-    trainer.train_microbatches_from_meta.assert_called_once_with(valid_meta)
+    trainer.train_microbatches_from_meta.assert_called_once_with(
+        valid_meta, train_fields=ctrl._train_fields
+    )
     trainer.finish_train_step.assert_called_once_with()
     assert ctrl._train_steps == 1