feat(data-plane): track data-plane time, latency percentiles and byte volume

examples/configs/grpo_math_1B.yaml

@@ -583,8 +583,9 @@ data_plane:
                                      # serialized groups and cost throughput;
                                      # higher buys little at linear HBM cost.
                                      # GDR needs that headroom to pay off.
-  # observability:                    # NotRequired
-  #   enabled: false
+  observability:                      # per-op data-plane timing/volume
+    enabled: true                     # per-op timing/volume; cost is below measurement noise
+    verify_tensor_hash: false         # debug: per-row hash of each row's values+dtype+shape, wire-in vs wire-out
 
 # Multi-Teacher On-Policy Distillation (MOPD): distills from one or more teacher
 # models into the policy via token-level teacher-minus-student logprob advantages,

examples/configs/recipes/llm/grpo-deepscaler-1.5b-8K-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-deepscaler-1.5b-8K.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-gemma3-1b-it-1n8g-fsdp2tp1-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-gemma3-1b-it-1n8g-fsdp2tp1.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-gspo-deepscaler-1.5b-8K-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-gspo-deepscaler-1.5b-8K.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.1-8b-instruct-1n8g-megatron-fp8-rollouts.v3-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-llama3.1-8b-instruct-1n8g-megatron-fp8-rollouts.v3.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.1-8b-instruct-2n8g-async-1off-single-controller-streaming2.yaml

@@ -20,6 +20,8 @@ checkpointing:
 # TransferQueue data plane is mandatory for the SingleController path.
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true
 
 # SC async-RL runtime knobs.
 async_rl:

examples/configs/recipes/llm/grpo-llama3.1-8b-instruct-2n8g-fsdp2tp1-noncolocated-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-llama3.1-8b-instruct-2n8g-fsdp2tp1-noncolocated.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.1-8b-instruct-2n8g-megatron-fp8-e2e-tq_mooncake.yaml

@@ -2,3 +2,5 @@ defaults: grpo-llama3.1-8b-instruct-2n8g-megatron-fp8-e2e.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.2-1b-instruct-1n8g-fsdp2tp1.v3-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-llama3.2-1b-instruct-1n8g-fsdp2tp1.v3.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.2-1b-instruct-1n8g-fsdp2tp2-temp0.8-topp0.9-topk50-tq_mooncake.yaml

@@ -2,3 +2,5 @@ defaults: grpo-llama3.2-1b-instruct-1n8g-fsdp2tp2-temp0.8-topp0.9-topk50.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.2-1b-instruct-1n8g-megatron-temp0.8-topp0.9-topk50-tq_mooncake.yaml

@@ -2,3 +2,5 @@ defaults: grpo-llama3.2-1b-instruct-1n8g-megatron-temp0.8-topp0.9-topk50.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.2-1b-instruct-1n8g-megatron-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-llama3.2-1b-instruct-1n8g-megatron.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-llama3.2-1b-instruct-1n8g-megatron_generation-tq_mooncake.yaml

@@ -2,3 +2,5 @@ defaults: grpo-llama3.2-1b-instruct-1n8g-megatron_generation.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-moonlight-16ba3b-4n8g-megatron-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-moonlight-16ba3b-4n8g-megatron.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-nanov3-30BA3B-1n8g-fsdp2-tq_mooncake.v2.yaml

@@ -2,3 +2,5 @@ defaults: grpo-nanov3-30BA3B-1n8g-fsdp2.v2.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-nanov3-30BA3B-2n8g-megatron-pack-cp-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-nanov3-30BA3B-2n8g-megatron-pack-cp.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-fsdp2tp1.v3-tq_simple.yaml

@@ -1,3 +1,5 @@
 defaults: grpo-qwen2.5-math-1.5b-instruct-1n8g-fsdp2tp1.v3.yaml
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron-single-controller-sync.yaml

@@ -20,6 +20,8 @@ checkpointing:
 # TransferQueue data plane is mandatory for the SingleController path.
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true
 
 # SC async-RL runtime knobs.
 async_rl:

examples/configs/recipes/llm/grpo-qwen3-1.7b-1n8g-megatron-eagle3-tq_mooncake.yaml

@@ -2,3 +2,5 @@ defaults: grpo-qwen3-1.7b-1n8g-megatron-eagle3.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/grpo-qwen3-30ba3b-10n8g-megatron-cp2-r3-async-single-controller.yaml

@@ -18,6 +18,8 @@ checkpointing:
 # TransferQueue is mandatory for the SingleController path.
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true
 
 async_rl:
   sampler:

examples/configs/recipes/llm/grpo-qwen3-30ba3b-8n8g-megatron-cp2-r3-tq_simple.yaml

@@ -3,6 +3,8 @@ checkpointing:
   checkpoint_dir: results/grpo-qwen3-30ba3b-8n8g-megatron-cp2-r3-tq_simple
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true
 logger:
   log_dir: logs/grpo-qwen3-30ba3b-8n8g-megatron-cp2-r3-tq_simple
   wandb:

examples/configs/recipes/llm/grpo-qwen3-8B-base-1n8g-fsdp2-lora-tq_mooncake.yaml

@@ -2,3 +2,5 @@ defaults: grpo-qwen3-8B-base-1n8g-fsdp2-lora.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

examples/configs/recipes/llm/mopd-qwen3-1.7b-3n8g-megatron-pack-single-controller.yaml

@@ -17,6 +17,8 @@ async_rl:
 
 data_plane:
   enabled: true
+  observability:
+    verify_tensor_hash: true
 
 checkpointing:
   checkpoint_dir: results/mopd-qwen3-1.7b-3n8g-megatron-pack-single-controller

examples/configs/recipes/llm/ppo-qwen2.5-1.5b-gsm8k-2n8g-megatron-valuetp2sp-dynbatch-noncolocated-async-single-controller.yaml

@@ -55,6 +55,8 @@ data_plane:
     local_buffer_size:    4294967296
     reuse_registered_buffers: true
     staging_buffer_size:   268435456
+  observability:
+    verify_tensor_hash: true
 
 # SC async-RL runtime knobs, replacing the nulled ppo.async_ppo block.
 async_rl:

examples/configs/recipes/llm/prorlv2-qwen2.5-math-1.5b-instruct-1n8g-fsdp2tp1.v2-tq_mooncake.yaml

@@ -2,3 +2,5 @@ defaults: prorlv2-qwen2.5-math-1.5b-instruct-1n8g-fsdp2tp1.v2.yaml
 data_plane:
   enabled: true
   backend: mooncake_cpu
+  observability:
+    verify_tensor_hash: true

nemo_rl/algorithms/grpo_sync.py

@@ -33,6 +33,7 @@
 
 import gc
 import os
+import time
 import warnings
 from typing import TYPE_CHECKING, Any, Optional
 
@@ -78,6 +79,13 @@
 from nemo_rl.data.llm_message_utils import batched_message_log_to_flat_message
 from nemo_rl.data.multimodal_utils import present_multimodal_fields
 from nemo_rl.data_plane.interfaces import KVBatchMeta
+from nemo_rl.data_plane.observability import (
+    MetricsDataPlaneClient,
+    cluster_step_metrics,
+    log_step_metrics,
+    merge_snapshots,
+    metrics_never_fail_the_step,
+)
 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.environments.interfaces import EnvironmentInterface
@@ -381,6 +389,63 @@ def _compute_seq_logprob_error_metrics(
     return masking_data["sample_mask"], seq_logprob_error_metrics
 
 
+def _log_data_plane_metrics(
+    policy: Any, logger: Logger, step: int, total_step_time: float
+) -> None:
+    """Log this step's data-plane cost. Never raises.
+
+    On by default, so this runs every step of every recipe.
+    """
+    with metrics_never_fail_the_step(step):
+        _log_data_plane_metrics_impl(policy, logger, step, total_step_time)
+
+
+def _log_data_plane_metrics_impl(
+    policy: Any, logger: Logger, step: int, total_step_time: float
+) -> None:
+    """Log this step's data-plane cost. No-op unless observability is enabled.
+
+    Prefers the cluster view -- the driver's counters plus every policy
+    worker's, summed -- and falls back to the driver's alone when the
+    fan-out reaches only one process. Reported one way or the other, never
+    both, so there is a single answer to "what did the data plane cost"
+    rather than two that disagree by roughly the DP degree.
+
+    The prefix names the scope because the two differ by a lot: the driver
+    issues about one op of each kind per step while the bulk traffic is the
+    workers' per-DP-rank ``get_samples``. Note that even the cluster view
+    omits the rollout actor, which builds its own client and is not on the
+    worker group -- so ``kv_first_write`` is not in these totals.
+
+    The previous reading lives on the policy, alongside the client whose
+    counters it differences, rather than in module state: two trainers in
+    one process would otherwise interleave one ``prev`` and produce
+    negative deltas.
+    """
+    client = getattr(policy, "dp_client", None)
+    if not isinstance(client, MetricsDataPlaneClient):
+        return  # observability disabled -> plain adapter
+
+    collect = getattr(policy, "collect_data_plane_snapshots", None)
+    collect_started = time.perf_counter()
+    snapshots = collect() if callable(collect) else []
+    if len(snapshots) > 1:
+        merged = merge_snapshots(snapshots)
+        # The fan-out is part of what observability costs, and the larger
+        # part: omitting it reported a twentieth of the real bill.
+        collect_ms = (time.perf_counter() - collect_started) * 1e3
+        prev = getattr(policy, "_prev_cluster_snapshot", {})
+        metrics = cluster_step_metrics(
+            merged, prev, total_step_time, collect_ms=collect_ms
+        )
+        policy._prev_cluster_snapshot = merged
+        log_step_metrics(logger, metrics, step, "cluster")
+    else:
+        # Single process, or the fan-out could not reach the workers.
+        metrics = client.get_step_metrics(total_step_time)
+        log_step_metrics(logger, metrics, step, "driver")
+
+
 def grpo_train_sync(
     policy: ColocatablePolicyInterface,
     policy_generation: GenerationInterface,
@@ -1375,6 +1440,10 @@ def grpo_train_sync(
             logger.log_metrics(
                 performance_metrics, total_steps + 1, prefix="performance"
             )
+            # Before the step_finished=True log below, which commits the step:
+            # anything logged against a committed step is dropped by wandb, so
+            # these series were computed, printed, and silently discarded.
+            _log_data_plane_metrics(policy, logger, total_steps + 1, total_time)
             logger.log_metrics(
                 timing_metrics,
                 total_steps + 1,

nemo_rl/algorithms/single_controller.py

@@ -119,6 +119,11 @@
 from nemo_rl.data.multimodal_utils import present_multimodal_fields
 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.observability import (
+    MetricsDataPlaneClient,
+    log_step_metrics,
+    metrics_never_fail_the_step,
+)
 from nemo_rl.data_plane.schema import (
     DP_CALIB_INPUT_FIELDS,
     DP_TRAIN_FIELDS,
@@ -1519,6 +1524,38 @@ async def _cleanup_consumed_metas_unlocked(
         if errors:
             raise BaseExceptionGroup("post-train DataPlane cleanup failed", errors)
 
+    def _log_data_plane_metrics(self, total_step_time: float) -> None:
+        """Log this step's data-plane cost. Never raises.
+
+        On by default, so this runs every step of every recipe. Mirrors
+        ``grpo_sync._log_data_plane_metrics``.
+        """
+        with metrics_never_fail_the_step(self._train_steps):
+            self._log_data_plane_metrics_impl(total_step_time)
+
+    def _log_data_plane_metrics_impl(self, total_step_time: float) -> None:
+        """Log this step's data-plane cost. No-op unless observability is enabled.
+
+        The synchronous loop logs these series from ``_log_data_plane_metrics``
+        in ``grpo_sync``. Without the same call here the single-controller path
+        builds the metrics client, pays for its counters on every op, and emits
+        nothing -- the failure is silent, because an empty dashboard looks the
+        same as a data plane that cost nothing.
+
+        Driver scope only, and the prefix says so. This client issues the
+        advantage stage's get plus the post-train clear; the bulk traffic is
+        the trainer and generation workers' own clients, in their own
+        processes with their own counters, so ``comm_volume_mb`` here is well
+        under what the job actually moved. ``grpo_sync`` gets a cluster view by
+        fanning out over its policy worker group; this loop has no such group to
+        fan out over, so driver scope is all there is here.
+        """
+        if not isinstance(self._dp_client, MetricsDataPlaneClient):
+            return  # observability disabled -> plain adapter
+
+        metrics = self._dp_client.get_step_metrics(total_step_time)
+        log_step_metrics(self._logger, metrics, self._train_steps, "driver")
+
     @staticmethod
     def _group_ids_from_meta(meta: KVBatchMeta) -> list[str]:
         """Return stable prompt-group IDs in canonical sample order."""
@@ -2849,6 +2886,11 @@ async def _train_pump(self) -> None:
             self._logger.log_metrics(
                 step_metrics, step=self._train_steps, prefix="train"
             )
+            # Must precede the step_finished=True log below. That log commits
+            # the wandb step, and wandb silently discards anything logged
+            # against a step it has already committed -- no exception, no
+            # failed return, just an empty chart. grpo_sync had the same bug.
+            self._log_data_plane_metrics(total_time)
             # step_finished=True here since this is the final log of our current step.
             self._logger.log_metrics(
                 timing_metrics,