[data] fix: merge metrics from all workers in DataProto.concat()

[szrlee]

Summary

Fix DataProto.concat() to properly merge all meta_info keys from all workers, preventing silent data loss when workers have different non-metric keys.

Problem

Previous implementation only preserved non-metric meta_info from the first worker:

# Old code - only looks at data[0]
merged_meta_info = {k: v for k, v in data[0].meta_info.items() if k != "metrics"}

This caused silent data loss when workers had different non-metric keys:

• data[0].meta_info = {"config": "A"} ✓ preserved
• data[1].meta_info = {"extra_key": "B"} ❌ lost
• Result: {"config": "A"} - missing extra_key

This contradicts the docstring which states meta_info is "merged".

Solution

This PR iterates through ALL workers to merge their non-metric meta_info while aggregating metrics:

# Merge non-metric meta_info and aggregate metrics from all workers
all_metrics = []
for d in data:
    for k, v in d.meta_info.items():
        if k == "metrics":
            if v is not None:
                if isinstance(v, list):
                    all_metrics.extend(v)
                else:
                    all_metrics.append(v)
        else:
            if k in merged_meta_info:
                # Ensure consistency for overlapping non-metric keys
                assert merged_meta_info[k] == v, f"Conflicting values for meta_info key '{k}'"
            else:
                merged_meta_info[k] = v

if all_metrics:
    merged_meta_info["metrics"] = all_metrics

Key improvements:

• ✅ All non-metric keys from all workers are preserved
• ✅ Detects conflicting values for the same key across workers
• ✅ Aggregates metrics from all workers in a single loop
• ✅ Handles edge cases: missing metrics, non-list values

Testing

Added 6 comprehensive unit tests in tests/test_protocol_on_cpu.py:

• test_concat_metrics_from_multiple_workers - All workers have metrics
• test_concat_with_empty_and_non_list_meta_info - Partial metrics coverage
• test_concat_first_worker_missing_metrics - First worker has no metrics
• test_concat_non_list_metrics - Single dict instead of list
• test_concat_merge_different_non_metric_keys - Different keys across workers
• test_concat_conflicting_non_metric_keys - Conflict detection

Files Changed

• verl/protocol.py: Updated DataProto.concat() to merge all meta_info keys
• tests/test_protocol_on_cpu.py: Added 2 new tests (6 total) covering all edge cases

---

Checklist

• Read the Contribute Guide
• Pre-commit checks passed (ruff, mypy, etc.)
• Documentation updated (N/A - bug fix, no API changes)
• Unit tests added (4 comprehensive tests covering all edge cases)
• CI request (pending)

[gemini-code-assist]

Code Review

This pull request effectively addresses a critical bug in DataProto.concat() by ensuring metrics from all distributed workers are aggregated. The addition of comprehensive unit tests is excellent and covers various important edge cases for the metrics merging logic.

I've identified one potential issue related to how non-metric meta_info is handled, which could lead to data loss in certain scenarios. My review comment includes a detailed explanation and a suggested code improvement to make the merging logic more robust and consistent with the updated documentation.

[vermouth1992]

This PR makes DataProto not general. We make perform metrics allgather inside workers