fix(registry): make reward manager registration idempotent

tests/workers/reward_manager/test_registry_on_cpu.py

@@ -92,3 +92,21 @@ def method(setup):
 
     assert OriginalClass().method() == 42
     assert REWARD_MANAGER_REGISTRY["return_test"] == OriginalClass
+
+
+def test_idempotent_registration_same_class(setup):
+    """Test that registering the same class twice is idempotent (no error).
+
+    This is important for distributed environments like Ray where modules
+    may be imported multiple times across workers.
+    """
+
+    @register("idempotent_test")
+    class IdempotentManager:
+        pass
+
+    # Register the same class again - should not raise
+    decorated = register("idempotent_test")(IdempotentManager)
+
+    assert decorated == IdempotentManager
+    assert REWARD_MANAGER_REGISTRY["idempotent_test"] == IdempotentManager

verl/workers/reward_manager/registry.py

@@ -21,19 +21,41 @@
 REWARD_MANAGER_REGISTRY: dict[str, type[AbstractRewardManager]] = {}
 
 
+def _get_class_identifier(cls: type) -> str:
+    """Get a unique identifier for a class based on its module and qualified name.
+
+    This is used instead of direct class comparison because in distributed environments
+    (like Ray), the same class may be imported multiple times with different object ids.
+    """
+    return f"{cls.__module__}.{cls.__qualname__}"
+
+
 def register(name: str) -> Callable[[type[AbstractRewardManager]], type[AbstractRewardManager]]:
     """Decorator to register a reward manager class with a given name.
 
     Args:
         name: `(str)`
             The name of the reward manager.
+
+    Note:
+        Registration is idempotent - registering the same class multiple times
+        (identified by module and qualname) will not raise an error. This is
+        necessary for distributed environments like Ray where modules may be
+        imported multiple times across workers.
     """
 
     def decorator(cls: type[AbstractRewardManager]) -> type[AbstractRewardManager]:
-        if name in REWARD_MANAGER_REGISTRY and REWARD_MANAGER_REGISTRY[name] != cls:
-            raise ValueError(
-                f"Reward manager {name} has already been registered: {REWARD_MANAGER_REGISTRY[name]} vs {cls}"
-            )
+        if name in REWARD_MANAGER_REGISTRY:
+            existing_cls = REWARD_MANAGER_REGISTRY[name]
+            existing_id = _get_class_identifier(existing_cls)
+            new_id = _get_class_identifier(cls)
+            if existing_id != new_id:
+                raise ValueError(
+                    f"Reward manager '{name}' has already been registered with a different class: "
+                    f"{existing_id} vs {new_id}"
+                )
+            # Same class (by qualname) - idempotent registration, skip silently
+            return cls
         REWARD_MANAGER_REGISTRY[name] = cls
         return cls