[trainer] fix: return NaN for empty tensors in compute_data_metrics (#5899)

### What does this PR do?

Fixes #5894

When all samples are aborted or `response_mask` is all `False`,
`compute_data_metrics` crashes with:
```
RuntimeError: max(): Expected reduction dim to be specified for input.numel() == 0. Specify the reduction dim with the 'dim' argument.
```

This PR replaces the crash with a warning + NaN return, so training can
continue gracefully in edge cases such as:
- **System debugging**: early training runs may produce invalid
responses; crashing blocks iteration
- **Unique sampling strategies**: some strategies may occasionally
produce batches where all responses are invalid

Note: `compute_data_metrics` is purely a logging/monitoring function —
its return values do not feed into loss computation. When all responses
are invalid, the actual loss gradient is 0 (via `masked_sum /
(num_tokens + 1e-8) = 0`), meaning no parameter updates occur, which is
the expected behavior.

### Checklist Before Starting

- [x] Search for similar PRs:
-
https://github.com/volcengine/verl/pulls?q=is%3Apr+compute_data_metrics
- Found related PR #5860 which fixes a similar issue in
`calculate_debug_metrics`

### Test

This is a defensive bug fix for an edge case.

### API and Usage Example

No API changes. The function signature remains the same:

```python
from verl.trainer.ppo.metric_utils import compute_data_metrics

# Normal case - works as before
metrics = compute_data_metrics(batch)
# Returns: {'critic/score/mean': 0.5, 'critic/rewards/max': 1.0, ...}

# Edge case - now handled gracefully instead of crashing
metrics = compute_data_metrics(batch_with_all_aborted)
# Logs warning, returns: {'critic/score/mean': nan, 'critic/rewards/max': nan, ...}
# Training continues normally (loss gradient = 0, no parameter updates)
```

### Design & Code Changes

**File changed:** `verl/trainer/ppo/metric_utils.py`

**Root Cause:**
`torch.max()` / `torch.min()` operations on empty tensors raise
RuntimeError when:
- All samples are aborted (empty `non_aborted_sequence_score/reward`)
- `response_mask` is all False (empty
`valid_adv/valid_returns/valid_values`)

**Solution:** Add `numel() > 0` checks with NaN fallback (same pattern
as PR #5860):

```python
if non_aborted_sequence_score.numel() > 0:
    score_mean = torch.mean(non_aborted_sequence_score).detach().item()
    score_max = torch.max(non_aborted_sequence_score).detach().item()
    score_min = torch.min(non_aborted_sequence_score).detach().item()
else:
    logger.warning("All samples are aborted, returning default score metrics")
    score_mean = score_max = score_min = float("nan")
```

Applied to:
- `non_aborted_sequence_score/reward` → score/reward metrics
- `valid_adv/valid_returns` → advantage/return metrics  
- `valid_values` → critic/values metrics (when use_critic=True)
- `non_aborted_response_length` → response length metrics

### Checklist Before Submitting

- [x] Read the [Contribute
Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md)
- [x] Apply pre-commit checks (ruff check & format passed locally)
- [x] Add / Update documentation (N/A - internal bug fix)
- [x] Add unit or end-to-end test(s) (N/A - edge case, no tests needed)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Boundless <ruihang_wu@163.com>

Author: 3 people

verl/trainer/ppo/metric_utils.py

@@ -15,6 +15,7 @@
 Metrics related to the PPO trainer.
 """
 
+import logging
 from collections import defaultdict
 from functools import partial
 from typing import Any, Callable
@@ -26,6 +27,8 @@
 from verl import DataProto
 from verl.utils.import_utils import deprecated
 
+logger = logging.getLogger(__name__)
+
 
 @deprecated("verl.utils.metric.reduce_metrics")
 def reduce_metrics(metrics: dict[str, list[Any]]) -> dict[str, Any]:
@@ -128,22 +131,40 @@ def compute_data_metrics(batch: DataProto, use_critic: bool = True) -> dict[str,
     non_aborted_sequence_score = sequence_score[non_aborted_mask]
     non_aborted_sequence_reward = sequence_reward[non_aborted_mask]
 
-    score_mean = torch.mean(non_aborted_sequence_score).detach().item()
-    score_max = torch.max(non_aborted_sequence_score).detach().item()
-    score_min = torch.min(non_aborted_sequence_score).detach().item()
+    if non_aborted_sequence_score.numel() > 0:
+        score_mean = torch.mean(non_aborted_sequence_score).detach().item()
+        score_max = torch.max(non_aborted_sequence_score).detach().item()
+        score_min = torch.min(non_aborted_sequence_score).detach().item()
+    else:
+        logger.warning("All samples are aborted, returning default score metrics")
+        score_mean = score_max = score_min = float("nan")
 
-    reward_mean = torch.mean(non_aborted_sequence_reward).detach().item()
-    reward_max = torch.max(non_aborted_sequence_reward).detach().item()
-    reward_min = torch.min(non_aborted_sequence_reward).detach().item()
+    if non_aborted_sequence_reward.numel() > 0:
+        reward_mean = torch.mean(non_aborted_sequence_reward).detach().item()
+        reward_max = torch.max(non_aborted_sequence_reward).detach().item()
+        reward_min = torch.min(non_aborted_sequence_reward).detach().item()
+    else:
+        logger.warning("All samples are aborted, returning default reward metrics")
+        reward_mean = reward_max = reward_min = float("nan")
 
     valid_adv = torch.masked_select(advantages, response_mask)
     valid_returns = torch.masked_select(returns, response_mask)
 
-    if use_critic:
-        values = batch.batch["values"]
-        valid_values = torch.masked_select(values, response_mask)
-        return_diff_var = torch.var(valid_returns - valid_values)
-        return_var = torch.var(valid_returns)
+    if valid_adv.numel() > 0:
+        adv_mean = torch.mean(valid_adv).detach().item()
+        adv_max = torch.max(valid_adv).detach().item()
+        adv_min = torch.min(valid_adv).detach().item()
+    else:
+        logger.warning("Response mask is all False, returning default advantage metrics")
+        adv_mean = adv_max = adv_min = float("nan")
+
+    if valid_returns.numel() > 0:
+        returns_mean = torch.mean(valid_returns).detach().item()
+        returns_max = torch.max(valid_returns).detach().item()
+        returns_min = torch.min(valid_returns).detach().item()
+    else:
+        logger.warning("Response mask is all False, returning default return metrics")
+        returns_mean = returns_max = returns_min = float("nan")
 
     # Aborted samples and non-aborted response length statistics
     # response_length_non_aborted/*: statistics computed on non-aborted samples only
@@ -158,7 +179,37 @@ def compute_data_metrics(batch: DataProto, use_critic: bool = True) -> dict[str,
             torch.mean(torch.eq(non_aborted_response_length, max_response_length).float()).detach().item()
         )
     else:
-        raise ValueError("All samples are aborted, this should not happen.")
+        logger.warning("All samples are aborted, returning default response length metrics")
+        non_aborted_response_length_mean = float("nan")
+        non_aborted_response_length_max = float("nan")
+        non_aborted_response_length_min = float("nan")
+        non_aborted_response_length_clip_ratio = float("nan")
+
+    if use_critic:
+        values = batch.batch["values"]
+        valid_values = torch.masked_select(values, response_mask)
+        if valid_returns.numel() > 0 and valid_values.numel() > 0:
+            return_diff_var = torch.var(valid_returns - valid_values)
+            return_var = torch.var(valid_returns)
+            critic_value_metrics = {
+                # values
+                "critic/values/mean": torch.mean(valid_values).detach().item(),
+                "critic/values/max": torch.max(valid_values).detach().item(),
+                "critic/values/min": torch.min(valid_values).detach().item(),
+                # vf explained var
+                "critic/vf_explained_var": (1.0 - return_diff_var / (return_var + 1e-5)).detach().item(),
+            }
+        else:
+            logger.warning("Response mask is all False, returning default value metrics")
+            critic_value_metrics = {
+                "critic/values/mean": float("nan"),
+                "critic/values/max": float("nan"),
+                "critic/values/min": float("nan"),
+                # vf explained var
+                "critic/vf_explained_var": float("nan"),
+            }
+    else:
+        critic_value_metrics = {}
 
     metrics = {
         # score
@@ -170,25 +221,14 @@ def compute_data_metrics(batch: DataProto, use_critic: bool = True) -> dict[str,
         "critic/rewards/max": reward_max,
         "critic/rewards/min": reward_min,
         # adv
-        "critic/advantages/mean": torch.mean(valid_adv).detach().item(),
-        "critic/advantages/max": torch.max(valid_adv).detach().item(),
-        "critic/advantages/min": torch.min(valid_adv).detach().item(),
+        "critic/advantages/mean": adv_mean,
+        "critic/advantages/max": adv_max,
+        "critic/advantages/min": adv_min,
         # returns
-        "critic/returns/mean": torch.mean(valid_returns).detach().item(),
-        "critic/returns/max": torch.max(valid_returns).detach().item(),
-        "critic/returns/min": torch.min(valid_returns).detach().item(),
-        **(
-            {
-                # values
-                "critic/values/mean": torch.mean(valid_values).detach().item(),
-                "critic/values/max": torch.max(valid_values).detach().item(),
-                "critic/values/min": torch.min(valid_values).detach().item(),
-                # vf explained var
-                "critic/vf_explained_var": (1.0 - return_diff_var / (return_var + 1e-5)).detach().item(),
-            }
-            if use_critic
-            else {}
-        ),
+        "critic/returns/mean": returns_mean,
+        "critic/returns/max": returns_max,
+        "critic/returns/min": returns_min,
+        **critic_value_metrics,
         # response length
         "response_length/mean": torch.mean(response_length).detach().item(),
         "response_length/max": torch.max(response_length).detach().item(),