feat(recipe): add Dynamo as rollout backend

[sophiayyya]

Author: @sophiayyya @hou2lin

Summary

Add a recipe-side Dynamo rollout backend for verl training.

This PR adds a recipe.dynamo integration that registers a Dynamo rollout replica, launches Dynamo subprocesses for async generation,
routes generation through Dynamo frontend, and supports rollout weight updates from the trainer side. Validated using retool recipe.

Changes

• Add recipe.dynamo.main_dynamo entry point and Hydra config for actor_rollout_ref.rollout.name=dynamo.
• Add DynamoReplica / DynamoHttpServer implementation to launch and manage etcd, NATS, Dynamo vLLM workers, and Dynamo frontend.
• Add Dynamo rollout adapter for control RPCs, sleep/wake, KV cache clear, global step propagation, and bucketed weight update.
• Add Dynamo vLLM worker extension so trainer-side and engine-side IPC/ZMQ handles agree in the multi-DP-per-node topology.
• Add Dynamo-specific agent loop/server manager plumbing.

Validation

• Experiment: retool
• Model: Qwen30B
• wandb report: https://wandb.ai/yangjingyi_algo/retool_32b_dynamo_smoke?nw=nwuseryangjingyi_algo

[gemini-code-assist]

Code Review

This pull request introduces the Dynamo rollout backend integration with verl, establishing a wrapper with a ZMQ control sidecar, a dedicated server manager, and a custom worker extension to handle node-global ranks. It also refactors the retool dataset processing to support AIME 2025 datasets and patches SandboxFusion to use system Python. The review feedback highlights a critical security vulnerability regarding insecure deserialization via pickle.loads in the ZMQ control protocol, recommending JSON or HMAC signatures. Additionally, the feedback suggests redirecting the output of etcd and nats-server subprocesses to dedicated log files, utilizing stable node-local port allocation for the control port to prevent EADDRINUSE collisions, and eliminating duplicate mapping logic in retool.py by reusing retool_dataset_utils.py.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Using pickle.loads to deserialize control requests received over a TCP network socket (VERL_DYNAMO_CONTROL_ZMQ) poses a high security risk of Remote Code Execution (RCE). Since the control payloads only contain simple metadata (method names, basic arguments, and timeouts), they can be safely serialized using json instead of pickle. Consider migrating the ZMQ control protocol to JSON, or sign the payloads using hmac with a shared secret key to verify the sender's authenticity before deserializing.

[gemini-code-assist]

The etcd process is launched without redirecting its stdout/stderr. If it produces a large volume of logs, it can clutter the main process's output. Consider redirecting its output to a dedicated log file in the log directory for better observability and debugging.

Suggested change

	def _wait_for_etcd(self, timeout: float):
	log_root = os.environ.get("VERL_DYNAMO_LOG_DIR", "/tmp")
	log_path = os.path.join(log_root, f"verl_dynamo_replica{self.replica_rank}_etcd.log")
	with open(log_path, "w") as f:
	self._etcd_process = subprocess.Popen(cmd, env=env, stdout=f, stderr=subprocess.STDOUT)

[gemini-code-assist]

The nats-server process is launched without redirecting its stdout/stderr. Consider redirecting its output to a dedicated log file in the log directory to keep the main process logs clean and assist in troubleshooting.

Suggested change

	self._nats_process = subprocess.Popen(cmd)
	log_root = os.environ.get("VERL_DYNAMO_LOG_DIR", "/tmp")
	log_path = os.path.join(log_root, f"verl_dynamo_replica{self.replica_rank}_nats.log")
	with open(log_path, "w") as f:
	self._nats_process = subprocess.Popen(cmd, stdout=f, stderr=subprocess.STDOUT)

[gemini-code-assist]

Using _allocate_tcp_port for control_port can lead to EADDRINUSE port collisions during concurrent startup of multiple replicas on the same physical node, as it lacks the replica-slot and node-slot segmentation used by _allocate_stable_node_port. Consider allocating control_port using _allocate_stable_node_port with a dedicated base port (e.g., 30000).

Suggested change

	control_port = self._allocate_tcp_port(bind_wildcard=False)
	control_port = self._allocate_stable_node_port(30000, spec_idx)

[gemini-code-assist]

The dataset mapping functions _map_fn_aime and _map_fn_default defined locally in this file duplicate the logic in retool/retool_dataset_utils.py. To maintain clean, DRY (Don't Repeat Yourself) code, consider keeping the import from retool_dataset_utils.py and updating the mapping functions there instead of duplicating them locally.