feat: add MOPD to single-controller text path

docs/about/algorithms/mopd.md

@@ -125,9 +125,26 @@ policy and generation nodes.
 
 ## Running MOPD
 
-MOPD collects rollouts through NeMo Gym, so use the NeMo Gym GRPO entrypoint
-with an MOPD recipe. The checked-in recipe uses placeholder dataset paths;
-override them for your local data:
+MOPD collects rollouts through NeMo Gym and supports both the legacy async GRPO
+runtime and the Single-Controller runtime. The checked-in recipes use
+placeholder dataset paths; override them for your local data.
+
+### Single-Controller text path
+
+The Single-Controller path moves rollout and teacher-logprob tensors through
+TransferQueue. It currently supports text-only MOPD rollouts:
+
+```sh
+uv run examples/run_grpo_single_controller.py \
+  --config examples/configs/recipes/llm/mopd-qwen3-1.7b-3n8g-megatron-pack-single-controller.yaml \
+  data.train.data_path=/path/to/train.jsonl \
+  data.validation.data_path=/path/to/val.jsonl
+```
+
+See [Train with Single-Controller](../../guides/single-controller.md) for the
+runtime's configuration and architecture.
+
+### Legacy async GRPO path
 
 ```sh
 uv run examples/nemo_gym/run_grpo_nemo_gym.py \
@@ -136,10 +153,10 @@ uv run examples/nemo_gym/run_grpo_nemo_gym.py \
   data.validation.data_path=/path/to/val.jsonl
 ```
 
-The reference recipe self-distills `Qwen/Qwen3-1.7B` (student == teacher) across
-3 nodes (1 policy + 1 vLLM + 1 teacher) with sequence packing enabled. Because
-student and teacher are identical, the OPD loss stays near zero — it is a
-correctness smoke test, not a demonstration of distillation gains.
+Both reference recipes self-distill `Qwen/Qwen3-1.7B` (student == teacher)
+across 3 nodes (1 policy + 1 vLLM + 1 teacher) with sequence packing enabled.
+Because student and teacher are identical, the OPD loss stays near zero — it is
+a correctness smoke test, not a demonstration of distillation gains.
 
 ## References
 

docs/guides/single-controller.md

@@ -175,6 +175,9 @@ SC reads its async knobs from `async_rl:` and **requires `grpo.async_grpo: null`
 
 The SC path is still under active development. Feature gaps are tracked in [issue #2625](https://github.com/NVIDIA-NeMo/RL/issues/2625). Notable items:
 
+- Multi-Teacher On-Policy Distillation (MOPD) is supported for text-only NeMo
+  Gym rollouts; multimodal/VLM MOPD is not yet supported. See
+  [Multi-Teacher On-Policy Distillation](../about/algorithms/mopd.md#running-mopd).
 - Train backend: only Megatron is supported and validated; the AutoModel training path has not been tested on SC.
 - Generation backend: only vLLM is supported and validated; Megatron generation, SGLang, and TRT-LLM have not been tested on SC.
 - Validation is not yet supported (setup raises on `val_period > 0`, `val_at_start`, or `val_at_end`).

examples/configs/recipes/llm/mopd-qwen3-1.7b-3n8g-megatron-pack-single-controller.yaml

@@ -0,0 +1,29 @@
+defaults: ./mopd-qwen3-1.7b-3n8g-megatron-pack.yaml
+
+# SingleController consumes async_rl and requires the legacy async block to be absent.
+grpo:
+  async_grpo: null
+  val_period: 0
+  # OPD needs the student pass, but not a separate frozen reference-policy pass.
+  skip_reference_policy_logprobs_calculation: true
+
+async_rl:
+  sampler:
+    name: in_order
+    max_lookahead_versions: 1
+  min_groups_for_streaming_train: ${grpo.num_prompts_per_step}
+  max_inflight_prompts: ${grpo.num_prompts_per_step}
+  max_buffered_rollouts: 64
+
+data_plane:
+  enabled: true
+
+checkpointing:
+  checkpoint_dir: results/mopd-qwen3-1.7b-3n8g-megatron-pack-single-controller
+  metric_name: null
+
+logger:
+  wandb:
+    name: mopd-qwen3-1.7b-3n8g-megatron-pack-single-controller
+  mlflow:
+    run_name: mopd-qwen3-1.7b-3n8g-megatron-pack-single-controller

examples/run_grpo_single_controller.py

@@ -167,6 +167,13 @@ def main() -> None:
                 except Exception as kill_error:
                     print(f"Env {env_name!r} kill failed: {kill_error}")
 
+        teacher_worker_groups = getattr(actor_args, "teacher_worker_groups", None) or {}
+        for teacher_alias, teacher in teacher_worker_groups.items():
+            try:
+                teacher.shutdown()
+            except Exception as e:
+                print(f"Teacher {teacher_alias!r} shutdown failed: {e}")
+
         for resource_name, resource in (
             ("Generation", actor_args.gen_handle),
             ("Trainer", actor_args.trainer_handle),

nemo_rl/algorithms/async_utils/replay_buffer.py

@@ -19,7 +19,7 @@
 import uuid
 from collections import Counter
 from collections.abc import Mapping
-from typing import Any, Iterable, Optional
+from typing import Any, Awaitable, Callable, Iterable, Optional
 
 import ray
 import torch
@@ -37,6 +37,10 @@
 from nemo_rl.utils.r3_trace import trace_rollout_payload
 
 
+class PostWriteEnrichmentError(RuntimeError):
+    """A rollout reached TQ but failed in required post-write processing."""
+
+
 # Classes with @ray.remote can't be inherited from, so we split the implementation out.
 class ReplayBufferImpl(ReplayBufferProtocol):
     """Replay buffer storing per-prompt groups.
@@ -774,6 +778,16 @@ def __init__(
         self.target_step_list: list[Optional[int]] = []
         self.ready_list: list[bool] = []
         self._group_ids: list[str] = []
+        self._post_write_enricher: Optional[
+            Callable[[KVBatchMeta, PromptGroupRecord], Awaitable[KVBatchMeta]]
+        ] = None
+
+    def set_post_write_enricher(
+        self,
+        enricher: Callable[[KVBatchMeta, PromptGroupRecord], Awaitable[KVBatchMeta]],
+    ) -> None:
+        """Install the required enrichment stage run before slots become ready."""
+        self._post_write_enricher = enricher
 
     def reserve(
         self,
@@ -864,6 +878,14 @@ async def commit(
                 tags=[dict(t) for t in tags],
             )
 
+            if self._post_write_enricher is not None:
+                try:
+                    meta = await self._post_write_enricher(meta, record)
+                except Exception as error:
+                    raise PostWriteEnrichmentError(
+                        f"post-write enrichment failed for group_id={group_id!r}"
+                    ) from error
+
             idx = self._group_ids.index(group_id)
             self.meta_list[idx] = meta
             self.end_weight_list[idx] = end_weight_version

nemo_rl/algorithms/metric_utils.py

@@ -40,7 +40,8 @@ class SetupTimingMetrics:
     parallel_wall_time_s: Optional[float] = None
     parallel_init_enabled: Optional[float] = None
 
-    # grpo-only phases (non-colocated OPD teachers, sparse refit, checkpoint-engine).
+    # Optional setup phases. OPD teacher timings are shared by legacy GRPO and SC;
+    # sparse refit and checkpoint-engine timings remain legacy-GRPO-only.
     teacher_reservation_time_s: Optional[float] = None
     teacher_model_init_time_s: Optional[float] = None
     teacher_init_time_s: Optional[float] = None