[ci] fix: fully async ci break (verl-project#5166)

### What does this PR do?

as title

### Checklist Before Starting

- [ ] Search for similar PRs. Paste at least one query link here: ...
- [ ] Format the PR title as `[{modules}] {type}: {description}` (This
will be checked by the CI)
- `{modules}` include `fsdp`, `megatron`, `veomni`, `sglang`, `vllm`,
`rollout`, `trainer`, `ci`, `training_utils`, `recipe`, `hardware`,
`deployment`, `ray`, `worker`, `single_controller`, `misc`, `perf`,
`model`, `algo`, `env`, `tool`, `ckpt`, `doc`, `data`, `cfg`, `reward`
- If this PR involves multiple modules, separate them with `,` like
`[megatron, fsdp, doc]`
  - `{type}` is in `feat`, `fix`, `refactor`, `chore`, `test`
- If this PR breaks any API (CLI arguments, config, function signature,
etc.), add `[BREAKING]` to the beginning of the title.
  - Example: `[BREAKING][fsdp, megatron] feat: dynamic batching`

### Test

> For changes that can not be tested by CI (e.g., algorithm
implementation, new model support), validate by experiment(s) and show
results like training curve plots, evaluation results, etc.

### API and Usage Example

> Demonstrate how the API changes if any, and provide usage example(s)
if possible.

```python
# Add code snippet or script demonstrating how to use this
```

### Design & Code Changes

> Demonstrate the high-level design if this PR is complex, and list the
specific changes.

### Checklist Before Submitting

> [!IMPORTANT]
> Please check all the following items before requesting a review,
otherwise the reviewer might deprioritize this PR for review.

- [ ] Read the [Contribute
Guide](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md).
- [ ] Apply [pre-commit
checks](https://github.com/volcengine/verl/blob/main/CONTRIBUTING.md#code-linting-and-formatting):
`pre-commit install && pre-commit run --all-files --show-diff-on-failure
--color=always`
- [ ] Add / Update [the
documentation](https://github.com/volcengine/verl/tree/main/docs).
- [ ] Add unit or end-to-end test(s) to [the CI
workflow](https://github.com/volcengine/verl/tree/main/.github/workflows)
to cover all the code. If not feasible, explain why: ...
- [ ] Once your PR is ready for CI, send a message in [the `ci-request`
channel](https://verl-project.slack.com/archives/C091TCESWB1) in [the
`verl` Slack
workspace](https://join.slack.com/t/verl-project/shared_invite/zt-3855yhg8g-CTkqXu~hKojPCmo7k_yXTQ).
(If not accessible, please try [the Feishu group
(飞书群)](https://applink.larkoffice.com/client/chat/chatter/add_by_link?link_token=772jd4f1-cd91-441e-a820-498c6614126a).)
- [ ] If your PR is related to the `recipe` submodule, please also
update the reference to the submodule commit via `git submodule update
--remote` or `cd recipe && git pull origin main`.

Author: yyDing1

tests/special_e2e/run_fully_async_policy.sh

@@ -15,7 +15,7 @@ MODEL_PATH=${MODEL_PATH:-${HOME}/models/${MODEL_ID}}
 
 
 rollout_mode="async"
-rollout_name="vllm" # sglang or vllm
+rollout_name="sglang" # sglang or vllm
 if [ "$rollout_mode" = "async" ]; then
     export VLLM_USE_V1=1
     return_raw_chat="True"

verl/experimental/fully_async_policy/agent_loop/agent_loop.py

@@ -32,7 +32,7 @@
 )
 from verl.experimental.agent_loop.prometheus_utils import update_prometheus_config
 from verl.protocol import DataProto
-from verl.single_controller.ray import RayResourcePool, RayWorkerGroup
+from verl.single_controller.ray import RayWorkerGroup
 from verl.utils.rollout_trace import (
     rollout_trace_attr,
     rollout_trace_op,
@@ -78,10 +78,13 @@ async def generate_for_partial(
 @ray.remote
 class FullyAsyncAgentLoopWorker(AgentLoopWorker):
     def __init__(
-        self, config: DictConfig, server_handles: list[ray.actor.ActorHandle], reward_router_address: str = None
+        self,
+        config: DictConfig,
+        server_handles: list[ray.actor.ActorHandle],
+        reward_loop_worker_handles: list[ray.actor.ActorHandle] = None,
     ):
         self.server_manager = FullyAsyncLLMServerManager(config, server_handles)
-        super().__init__(config, server_handles, reward_router_address)
+        super().__init__(config, server_handles, reward_loop_worker_handles)
         # A shared cancellation event for all agent loops running on this worker.
         self.cancellation_event = asyncio.Event()
 
@@ -211,12 +214,14 @@ async def resume_agent_loops(self):
 
 class FullyAsyncAgentLoopManager(AgentLoopManager):
     def __init__(
-        self, config: DictConfig, worker_group: RayWorkerGroup = None, rm_resource_pool: RayResourcePool = None
+        self,
+        config: DictConfig,
+        worker_group: RayWorkerGroup = None,
+        reward_loop_worker_handles: list[ray.actor.ActorHandle] = None,
     ):
         self.config = config
         self.worker_group = worker_group
-        self.reward_model_manager = None
-        self.reward_router_address = None
+        self.reward_loop_worker_handles = reward_loop_worker_handles
         self.agent_loop_workers_class = FullyAsyncAgentLoopWorker
 
         # Select rollout replica class based on rollout name
@@ -234,27 +239,23 @@ def __init__(
         else:
             raise ValueError(f"Unsupported rollout name: {rollout_name}. Supported values are 'sglang' and 'vllm'.")
 
-        self.rm_resource_pool = rm_resource_pool
         self.rollout_replicas = None
         self.server_handles = None
         self.server_addresses = None
         self.agent_loop_workers = None
 
     @classmethod
     async def create(
-        cls, config: DictConfig, worker_group: RayWorkerGroup = None, rm_resource_pool: RayResourcePool = None
+        cls,
+        config: DictConfig,
+        worker_group: RayWorkerGroup = None,
+        reward_loop_worker_handles: list[ray.actor.ActorHandle] = None,
     ):
-        instance = cls(config, worker_group, rm_resource_pool)
+        instance = cls(config, worker_group, reward_loop_worker_handles)
         await instance._async_init()
         return instance
 
     async def _async_init(self):
-        if self.config.reward_model.enable and self.config.reward_model.enable_resource_pool:
-            from verl.experimental.reward_loop import RewardModelManager
-
-            self.reward_model_manager = RewardModelManager(self.config.reward_model, self.rm_resource_pool)
-            self.reward_router_address = self.reward_model_manager.get_router_address()
-
         await self._initialize_llm_servers_async()
         self._init_agent_loop_workers()
 

verl/experimental/fully_async_policy/fsdp_workers.py

@@ -85,7 +85,23 @@ async def init_engine():
                         return self.rollout._engine
 
                     inference_model = self._run_async_safely(init_engine())
-                    if inference_model is None:
+                    # For ServerAdapter, only TP rank 0 initializes the engine
+                    # TP rank != 0 can safely have inference_model as None
+                    from verl.workers.rollout.sglang_rollout.sglang_rollout import ServerAdapter
+
+                    is_server_adapter = isinstance(self.rollout, ServerAdapter)
+                    is_non_tp_rank = False
+                    if (
+                        is_server_adapter
+                        and hasattr(self.rollout, "device_mesh")
+                        and self.rollout.device_mesh is not None
+                    ):
+                        try:
+                            is_non_tp_rank = self.rollout.device_mesh["infer_tp"].get_local_rank() != 0
+                        except Exception:
+                            pass
+
+                    if inference_model is None and not (is_server_adapter and is_non_tp_rank):
                         raise RuntimeError(
                             f"Failed to initialize rollout engine. "
                             f"rollout type: {type(self.rollout)}, "

verl/experimental/fully_async_policy/megatron_worker.py

@@ -82,7 +82,23 @@ async def init_engine():
                         return self.rollout._engine
 
                     inference_model = self._run_async_safely(init_engine())
-                    if inference_model is None:
+                    # For ServerAdapter, only TP rank 0 initializes the engine
+                    # TP rank != 0 can safely have inference_model as None
+                    from verl.workers.rollout.sglang_rollout.sglang_rollout import ServerAdapter
+
+                    is_server_adapter = isinstance(self.rollout, ServerAdapter)
+                    is_non_tp_rank = False
+                    if (
+                        is_server_adapter
+                        and hasattr(self.rollout, "device_mesh")
+                        and self.rollout.device_mesh is not None
+                    ):
+                        try:
+                            is_non_tp_rank = self.rollout.device_mesh["infer_tp"].get_local_rank() != 0
+                        except Exception:
+                            pass
+
+                    if inference_model is None and not (is_server_adapter and is_non_tp_rank):
                         raise RuntimeError(
                             f"Failed to initialize rollout engine. "
                             f"rollout type: {type(self.rollout)}, "
@@ -179,7 +195,23 @@ async def init_engine():
                         return self.rollout._engine
 
                     inference_model = self._run_async_safely(init_engine())
-                    if inference_model is None:
+                    # For ServerAdapter, only TP rank 0 initializes the engine
+                    # TP rank != 0 can safely have inference_model as None
+                    from verl.workers.rollout.sglang_rollout.sglang_rollout import ServerAdapter
+
+                    is_server_adapter = isinstance(self.rollout, ServerAdapter)
+                    is_non_tp_rank = False
+                    if (
+                        is_server_adapter
+                        and hasattr(self.rollout, "device_mesh")
+                        and self.rollout.device_mesh is not None
+                    ):
+                        try:
+                            is_non_tp_rank = self.rollout.device_mesh["infer_tp"].get_local_rank() != 0
+                        except Exception:
+                            pass
+
+                    if inference_model is None and not (is_server_adapter and is_non_tp_rank):
                         raise RuntimeError(
                             f"Failed to initialize rollout engine. "
                             f"rollout type: {type(self.rollout)}, "

verl/experimental/one_step_off_policy/ray_trainer.py

@@ -291,14 +291,7 @@ def _init_async_rollout_manager(self):
 
         self.async_rollout_mode = True
 
-        if self.config.reward_model.enable and self.config.reward_model.enable_resource_pool:
-            rm_resource_pool = self.resource_pool_manager.get_resource_pool(Role.RewardModel)
-        else:
-            rm_resource_pool = None
-
-        self.async_rollout_manager = OneStepOffAgentLoopManager(
-            config=self.config, worker_group=self.rollout_wg, rm_resource_pool=rm_resource_pool
-        )
+        self.async_rollout_manager = OneStepOffAgentLoopManager(config=self.config, worker_group=self.rollout_wg)
 
     def sync_rollout_weights(self):
         self.actor_wg.sync_rollout_weights()