diff --git a/docs/guides/async-grpo.md b/docs/guides/async-grpo.md
index 960197303b9..6de30e72691 100644
--- a/docs/guides/async-grpo.md
+++ b/docs/guides/async-grpo.md
@@ -207,7 +207,7 @@ If no `replay_buffer.pt` file is found in the latest checkpoint directory, train
 
 4. **In-Flight Weight Updates**: Enable `in_flight_weight_updates: true` to refit without waiting for the longest in-flight generation to finish. Except for managed Dynamo, the collector requests a generation pause and resume from every async backend around the weight transfer. Async vLLM implements this contract while preserving request state. A backend that does not implement the hook emits a warning once per backend type per process and refits without a collector-side pause or drain; SGLang is in this group today and instead relies on the pause its own weight synchronizer performs around the transfer. Managed Dynamo always drains active trajectories before refit. vLLM requires `async_engine: true`; the Megatron backend is always async-engine.
 
-5. **Recompute KV Cache After Weight Updates**: Set `recompute_kv_cache_after_weight_updates: true` to invalidate reusable KV/prefix caches when weights change. On the native async vLLM in-flight path, caches are cleared while generation is paused, so preserved requests recompute their KV after resuming. Other refit paths keep their existing post-update invalidation behavior. When false, in-flight requests retain their pre-update KV cache.
+5. **Recompute KV Cache After Weight Updates**: Set `recompute_kv_cache_after_weight_updates: true` to invalidate reusable KV/prefix caches when weights change. On the native async vLLM in-flight path, caches are cleared while generation is paused, so preserved requests recompute their KV after resuming. Other refit paths keep their existing post-update invalidation behavior. When false, in-flight requests retain their pre-update KV cache. On the Megatron generation backend, this must agree with `policy.generation.mcore_generation_config.kv_cache_management_mode`; setup errors on a mismatch.
 
 ## Why Importance Sampling Correction Is Required for Async
 
diff --git a/docs/guides/single-controller.md b/docs/guides/single-controller.md
index f5815ec359f..d84c2850840 100644
--- a/docs/guides/single-controller.md
+++ b/docs/guides/single-controller.md
@@ -25,7 +25,7 @@ uv run examples/run_grpo_single_controller.py --config <your-sc.yaml>
       enabled: true
     ```
 
-2. **Enable vLLM async engine** and **disable colocated inference** (SC drives rollout via `RolloutManager.generate_and_push`, which is only supported on the disaggregated async engine; setup rejects `colocated.enabled: true`):
+2. **Pick a generation backend** and **disable colocated inference** (setup rejects `colocated.enabled: true`). With vLLM, enable the async engine (SC drives rollout via `RolloutManager.generate_and_push`, which is only supported on the disaggregated async engine):
 
     ```yaml
     policy:
@@ -40,6 +40,23 @@ uv run examples/run_grpo_single_controller.py --config <your-sc.yaml>
             gpus_per_node: 4  # inference GPUs; remainder go to training
     ```
 
+    Megatron generation is also supported, non-colocated only. It requires the Megatron trainer (`policy.megatron_cfg.enabled: true`) and NeMo-Gym rollouts additionally require `policy.generation.mcore_generation_config.expose_http_server: true`. The exemplar — a NeMo-Gym run with the OpenAI server exposed — lives at [examples/nemo_gym/grpo_qwen3_0_6b_megatron_generation_single_controller.yaml](../../examples/nemo_gym/grpo_qwen3_0_6b_megatron_generation_single_controller.yaml):
+
+    ```yaml
+    policy:
+      megatron_cfg:
+        enabled: true
+      generation:
+        backend: "megatron"
+        mcore_generation_config:
+          expose_http_server: true  # required for NeMo-Gym rollouts
+        colocated:
+          enabled: false
+          resources:
+            num_nodes: 1
+            gpus_per_node: 1  # inference GPUs; remainder go to training
+    ```
+
 3. **One RL step = one training batch.** The batch a step trains on is the whole step (see `validate_single_controller_config` in [nemo_rl/algorithms/single_controller_utils/config.py](../../nemo_rl/algorithms/single_controller_utils/config.py)). A GRPO step is also one optimizer step; a PPO step is `ppo.ppo_epochs` of them over that same batch.
 
     ```python
@@ -181,8 +198,8 @@ The SC path is still under active development. Feature gaps are tracked in [issu
   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`).
+- Generation backend: vLLM and Megatron generation are supported; 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`); checkpointing is.
 - (PPO) Rollout drop budgets — `async_rl.rollout_failure.max_skipped_prompts` and `max_consecutive_dropped_prompts` must both be `0`. A drop shortens the step, and the critic shards it against the configured `value.train_global_batch_size` rather than its actual size, so setup rejects a non-zero budget. The resiliency layer stays available on GRPO.
 - Reward shaping and sample filtering — `overlong_filtering`, `reward_shaping`, `reward_scaling`, and `use_dynamic_sampling` are implemented on neither algorithm block, so setup rejects them rather than silently skipping the shaping.
 - The `windowed` sampler has no `over_sampling_ratio` cap — over-produced groups aged past the window are evicted, wasting rollout compute.
diff --git a/examples/configs/grpo_math_1B.yaml b/examples/configs/grpo_math_1B.yaml
index 53c8c13abce..d7f17aeac11 100644
--- a/examples/configs/grpo_math_1B.yaml
+++ b/examples/configs/grpo_math_1B.yaml
@@ -62,7 +62,7 @@ grpo:
     # when transient generation errors are expected and acceptable to drop.
     max_generation_failures: 0
     in_flight_weight_updates: false # Set to true to enable in-flight weight updates
-    recompute_kv_cache_after_weight_updates: false # Set to true to recompute kv cache after weight updates
+    recompute_kv_cache_after_weight_updates: false # Set to true to recompute kv cache after weight updates.
 
 # Reward-zeroing penalties applied to NeMo-Gym rollout results.
 reward_penalties:
@@ -410,7 +410,7 @@ policy:
       enable_chunked_prefill: true  # Split long prefills into chunks for better memory management
       enable_prefix_caching: false  # Reuse KV blocks across requests sharing a prompt prefix.
       max_tokens: 16384 # Maximum number of tokens to use in a single step. Analogous to vllm's max_num_batched_tokens
-      kv_cache_management_mode: "persist"  # KV cache lifecycle across suspend/resume. Options: "persist", "offload". To select "recompute", set grpo.async_grpo.recompute_kv_cache_after_weight_updates=true.
+      kv_cache_management_mode: "persist"  # KV cache lifecycle across suspend/resume. Options: "persist", "offload", "recompute".
       materialize_only_last_token_logits: true
       num_speculative_tokens: 0
       logprobs_mode: processed_logprobs  # Return log-probs after sampling processors. Use raw_logprobs for parity with policy recomputation.
diff --git a/examples/configs/recipes/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync.yaml b/examples/configs/recipes/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync.yaml
new file mode 100644
index 00000000000..a0d70c08ea4
--- /dev/null
+++ b/examples/configs/recipes/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync.yaml
@@ -0,0 +1,12 @@
+defaults: ./grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron-single-controller-sync.yaml
+logger:
+  log_dir: logs/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync
+  wandb:
+    name: grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync
+checkpointing:
+  checkpoint_dir: results/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync
+policy:
+  generation:
+    backend: megatron
+    mcore_generation_config:
+      kv_cache_management_mode: recompute
diff --git a/examples/nemo_gym/grpo_nanov3.yaml b/examples/nemo_gym/grpo_nanov3.yaml
index 4f1f91d8107..8d762438ac5 100644
--- a/examples/nemo_gym/grpo_nanov3.yaml
+++ b/examples/nemo_gym/grpo_nanov3.yaml
@@ -242,7 +242,7 @@ policy:
       use_cuda_graphs_for_non_decode_steps: true  # Enable CUDA graphs for prefill/context processing
       enable_chunked_prefill: true
       max_tokens: ${policy.max_total_sequence_length} # Maximum number of tokens to use in a single step. Analogous to vllm's max_num_batched_tokens
-      kv_cache_management_mode: "persist"  # KV cache lifecycle across suspend/resume. Options: "persist", "offload". To select "recompute", set grpo.async_grpo.recompute_kv_cache_after_weight_updates=true.
+      kv_cache_management_mode: "persist"  # KV cache lifecycle across suspend/resume. Options: "persist", "offload", "recompute".
       materialize_only_last_token_logits: true
       num_speculative_tokens: 0
       logprobs_mode: processed_logprobs
diff --git a/examples/nemo_gym/grpo_qwen3_0_6b_megatron_generation_single_controller.yaml b/examples/nemo_gym/grpo_qwen3_0_6b_megatron_generation_single_controller.yaml
new file mode 100644
index 00000000000..dd3ac8f54f4
--- /dev/null
+++ b/examples/nemo_gym/grpo_qwen3_0_6b_megatron_generation_single_controller.yaml
@@ -0,0 +1,78 @@
+# GRPO on the NeMo-Gym workplace-assistant environment via the SingleController path,
+# using non-colocated Megatron Inference.
+# Gym rollouts go through the persistent Megatron engine's OpenAI-compatible server,
+# Smoke-test scale; Qwen3-0.6B on one node with 2 GPUs (1 training + 1 generation).
+# The CI-run variant, tests/functional/grpo_megatron_generation_gym_single_controller.sh,
+# loads this file and overrides only test scale, data paths, and logging; the resolved
+# config stays value-equal to the vLLM SC test (grpo_async_gym_single_controller.sh)
+# everywhere but the generation backend.
+defaults: "grpo_qwen3_30ba3b_instruct.yaml"
+
+grpo:
+  # SC requires one optimizer step per RL step:
+  # num_prompts_per_step * num_generations_per_prompt == policy.train_global_batch_size
+  num_prompts_per_step: 4
+  num_generations_per_prompt: 2
+  max_num_steps: 10  # short demo; raise for a real run
+  # SC does not support validation yet (setup raises when it is enabled).
+  val_period: 0
+  val_at_start: false
+  # The KL term below needs reference logprobs; the base skips them.
+  skip_reference_policy_logprobs_calculation: false
+  # SC replaces the legacy async-GRPO path.
+  async_grpo: null
+
+loss_fn:
+  # A small KL term (the base uses 0) exercises the reference-model path end to end.
+  reference_policy_kl_penalty: 0.01
+  use_importance_sampling_correction: true
+
+policy:
+  model_name: Qwen/Qwen3-0.6B
+  train_global_batch_size: 8
+  # Full workplace-assistant prompts (all tools attached) run past 4k tokens.
+  max_total_sequence_length: 8192
+
+  megatron_cfg:
+    tensor_model_parallel_size: 1
+    expert_model_parallel_size: 1
+    context_parallel_size: 1
+    sequence_parallel: false
+
+  generation:
+    backend: "megatron"
+    mcore_generation_config:
+      # NeMo-Gym drives rollouts through the engine's OpenAI server.
+      expose_http_server: true
+    colocated:
+      enabled: false
+      resources:
+        num_nodes: 1
+        gpus_per_node: 1
+
+# The gym family sets no data_plane block. Only keys without a schema default
+# are set here; see nemo_rl/data_plane/interfaces.py, and the fully documented
+# block in examples/configs/grpo_math_1B.yaml.
+data_plane:
+  enabled: true
+  impl: transfer_queue
+  backend: "simple"
+  claim_meta_poll_interval_s: 0.5
+  simple:
+    num_storage_units: ${mul:2, ${cluster.num_nodes}}  # TQ wants >= 2 per node
+
+async_rl:
+  sampler:
+    name: in_order
+    # 0 = fully synchronous, so the importance sampling correction above is an
+    # inert no-op (all ratios are 1); it is enabled to match the vLLM SC test.
+    max_lookahead_versions: 0
+  min_groups_for_streaming_train: ${grpo.num_prompts_per_step}
+  max_inflight_prompts: ${grpo.num_prompts_per_step}
+  max_buffered_rollouts: ${grpo.num_prompts_per_step}
+
+checkpointing:
+  enabled: false
+
+cluster:
+  gpus_per_node: 2
diff --git a/nemo_rl/algorithms/grpo.py b/nemo_rl/algorithms/grpo.py
index 2b9ab8d5ee1..4c3ca649841 100644
--- a/nemo_rl/algorithms/grpo.py
+++ b/nemo_rl/algorithms/grpo.py
@@ -1180,18 +1180,26 @@ def _spinup_nemo_gym(base_urls, model_name):
         )
         policy_config["megatron_cfg"]["train_iters"] = total_train_iters
 
-        # When the user opts into recompute-after-refit on the megatron side,
-        # override mcore's kv_cache_management_mode to "recompute" directly.
+    # Megatron generation expresses recompute-after-refit engine-side via
+    # `kv_cache_management_mode="recompute"`; the loop-level flag must agree.
+    if generation_config["backend"] == "megatron":
         async_grpo_config = grpo_config.async_grpo
-        if async_grpo_config.recompute_kv_cache_after_weight_updates:
-            mcore_cfg = policy_config["generation"]["mcore_generation_config"]
-            prior_mode = mcore_cfg.get("kv_cache_management_mode", "persist")
-            if prior_mode != "recompute":
-                print(
-                    f"kv_cache_management_mode overridden '{prior_mode}' -> 'recompute' by "
-                    f"grpo.async_grpo.recompute_kv_cache_after_weight_updates=True."
-                )
-            mcore_cfg["kv_cache_management_mode"] = "recompute"
+        recompute_kv_cache = bool(
+            async_grpo_config is not None
+            and async_grpo_config.recompute_kv_cache_after_weight_updates
+        )
+        kv_cache_mode = generation_config["mcore_generation_config"][
+            "kv_cache_management_mode"
+        ]
+        if recompute_kv_cache != (kv_cache_mode == "recompute"):
+            raise ValueError(
+                "grpo.async_grpo.recompute_kv_cache_after_weight_updates="
+                f"{recompute_kv_cache} conflicts with policy.generation."
+                f"mcore_generation_config.kv_cache_management_mode={kv_cache_mode!r}: "
+                "with policy.generation.backend='megatron' the two must agree. "
+                "Either set the flag to true with kv_cache_management_mode="
+                "'recompute', or leave the flag false with 'persist'/'offload'."
+            )
 
     # Define initialization functions that will be used in all paths
     init_reference_model = loss_config.reference_policy_kl_penalty > 0
@@ -1694,12 +1702,9 @@ def init_dynamo():
         if policy_generation.weight_synchronizer is None:
             init_megatron_weight_synchronizer(policy, policy_generation)
         if enable_nemo_gym:
-            served_urls = policy_generation.dp_openai_server_base_urls
-            if served_urls != [reserved_url]:
-                raise RuntimeError(
-                    "Megatron server came up at a different address than the one "
-                    f"pre-published to NeMo Gym: reserved {reserved_url}, serving {served_urls}."
-                )
+            MegatronGeneration.verify_served_address(
+                policy_generation.dp_openai_server_base_urls, reserved_url
+            )
     # if it is not colocated inference, initialize collective communication for update weights
     elif (
         not colocated_inference
diff --git a/nemo_rl/algorithms/single_controller.py b/nemo_rl/algorithms/single_controller.py
index 37021873b59..0597ee1e514 100644
--- a/nemo_rl/algorithms/single_controller.py
+++ b/nemo_rl/algorithms/single_controller.py
@@ -89,6 +89,7 @@
 from nemo_rl.experience.failures import RolloutStall
 from nemo_rl.experience.rollout_manager import RolloutOutcome
 from nemo_rl.models.generation.fleet_health import ShardState
+from nemo_rl.models.generation.megatron.megatron_generation import MegatronGeneration
 from nemo_rl.models.generation.sglang.sglang_generation import SGLangGeneration
 from nemo_rl.models.generation.vllm import VllmGeneration
 from nemo_rl.models.policy.tq_policy import TQPolicy
@@ -97,7 +98,7 @@
 from nemo_rl.utils.logger import Logger
 from nemo_rl.utils.timer import TimeoutChecker, Timer
 
-Generation = Union[VllmGeneration, SGLangGeneration]
+Generation = Union[VllmGeneration, SGLangGeneration, MegatronGeneration]
 
 # Named `log` rather than `logger` to keep it distinct from the experiment
 # Logger this module also uses as `self._logger`.
@@ -350,8 +351,9 @@ def __init__(
 
     async def run(self) -> dict[str, Any]:
         """Main entry point. Runs until max_train_steps is reached."""
-        # Synchronize weights before starting the pumps
-        await self._sync_weights()
+        # Synchronize weights before starting the pumps, unless setup already delivered them.
+        if self._weight_synchronizer.is_stale:
+            await self._sync_weights()
         self._rollout_manager.set_weight_version(self._trainer_version)
 
         await self._maybe_restore_replay_buffer()
diff --git a/nemo_rl/algorithms/single_controller_utils/setup.py b/nemo_rl/algorithms/single_controller_utils/setup.py
index 82e1ad0a503..25daea5f0e7 100644
--- a/nemo_rl/algorithms/single_controller_utils/setup.py
+++ b/nemo_rl/algorithms/single_controller_utils/setup.py
@@ -78,6 +78,7 @@
     RolloutTimeouts,
 )
 from nemo_rl.experience.rollouts import should_mask_flagged_samples
+from nemo_rl.models.generation import resolve_generation_class
 from nemo_rl.models.generation.fleet_health import (
     FleetHealthPolicy,
     GenerationFleetHealth,
@@ -90,6 +91,7 @@
 from nemo_rl.models.generation.interfaces import (
     resolve_routed_experts_dtype_name_for_model,
 )
+from nemo_rl.models.generation.megatron.megatron_generation import MegatronGeneration
 from nemo_rl.models.generation.sglang.config import SGLangConfig
 from nemo_rl.models.generation.sglang.sglang_generation import SGLangGeneration
 from nemo_rl.models.generation.vllm import VllmGeneration
@@ -237,10 +239,6 @@ def _build_clusters(
         return cluster, cluster, teacher_topology
 
     # Non-colocated: split node into train + inference clusters.
-    assert backend != "megatron", (
-        "The Megatron generation backend does not support non-colocated inference "
-        "in SingleController."
-    )
     inference_resources = generation_config["colocated"]["resources"]
     inference_gpus_per_node = inference_resources["gpus_per_node"]
     if inference_gpus_per_node is None:
@@ -300,10 +298,14 @@ def _build_clusters(
                 gpus_per_instance = generation_config_dict["sglang_cfg"].get(
                     "gpus_per_server", 1
                 )
+            elif backend == "megatron":
+                gpus_per_instance = MegatronGeneration.nvlink_domain_span(
+                    master_config.policy
+                )
             else:
                 raise ValueError(
-                    "single_controller_utils.setup only supports vllm or sglang "
-                    f"generation; got {backend!r}"
+                    "single_controller_utils.setup only supports vllm, sglang, "
+                    f"or megatron generation; got {backend!r}"
                 )
             nodes_per_instance = (
                 gpus_per_instance + inference_gpus_per_node - 1
@@ -362,17 +364,23 @@ def _build_generation(
     master_config: MasterConfig,
     *,
     defer_model_load: bool = False,
+    reserved_http_server_port: Optional[int] = None,
+    tokenizer: Optional[PreTrainedTokenizerBase] = None,
+    processor: Optional[AutoProcessor] = None,
 ) -> tuple[Any, float]:
-    """Spin up the generation backend (vLLM or SGLang).
+    """Spin up the generation backend (vLLM, SGLang, or Megatron).
 
     Args:
         inference_cluster: Ray virtual cluster the generation workers run on.
         master_config: SC MasterConfig.
-        defer_model_load: If True (for the NeMo-Gym flow), reserve OpenAI server URLs without loading weights; caller runs gen.load_and_start() later.
+        defer_model_load: If True (for the NeMo-Gym flow), reserve OpenAI server URLs without loading weights; caller runs gen.load_and_start() later (vLLM only).
+        reserved_http_server_port: OpenAI server port pre-published to NeMo-Gym (Megatron only).
+        tokenizer: Tokenizer for the dedicated Megatron inference policy (Megatron only).
+        processor: Optional AutoProcessor for VLM paths (Megatron only).
 
     Returns:
         A tuple of (generation object, wall time spent in this call). The
-        generation object is a VllmGeneration or SGLangGeneration.
+        generation object is a VllmGeneration, SGLangGeneration, or MegatronGeneration.
     """
     t0 = time.perf_counter()
     generation_config = master_config.policy["generation"]
@@ -404,9 +412,27 @@ def _build_generation(
             sglang_cfg=sglang_config,
         )
 
+    elif backend == "megatron":
+        assert not defer_model_load, (
+            "defer_model_load is only supported for the vllm backend"
+        )
+        assert tokenizer is not None, "Megatron generation requires a tokenizer"
+        # Non-colocated only (colocated is rejected at config validation).
+        # The inference and trainer policies build in parallel;
+        # the inference engine only becomes live at the initial refit, which delivers weights.
+        gen = MegatronGeneration(
+            config=master_config.policy,
+            tokenizer=tokenizer,
+            cluster=inference_cluster,
+            reserved_http_server_port=reserved_http_server_port,
+            processor=processor,
+            skip_weight_load=True,
+        )
+
     else:
         raise ValueError(
-            f"single_controller_utils.setup only supports vllm or sglang generation; got {backend!r}"
+            "single_controller_utils.setup only supports vllm, sglang, or megatron "
+            f"generation; got {backend!r}"
         )
 
     if not defer_model_load:
@@ -544,8 +570,7 @@ def _generation_max_seq_len(generation_config) -> int:
     """Return the per-backend max sequence length.
 
     vllm uses vllm_cfg.max_model_len; sglang uses sglang_cfg.context_length;
-    megatron generation has no dedicated field and routes max_new_tokens
-    through as max_sequence_length on the inference worker.
+    megatron uses mcore_generation_config.max_model_len.
     """
     backend = generation_config["backend"]
     if backend == "vllm":
@@ -553,7 +578,7 @@ def _generation_max_seq_len(generation_config) -> int:
     if backend == "sglang":
         return generation_config["sglang_cfg"]["context_length"]
     if backend == "megatron":
-        return generation_config["max_new_tokens"]
+        return generation_config["mcore_generation_config"]["max_model_len"]
     raise ValueError(f"Unknown generation backend: {backend!r}")
 
 
@@ -634,9 +659,15 @@ def _shard_base_urls(generation: Any) -> Optional[list[Optional[str]]]:
     return urls
 
 
-def _maybe_start_generation_router(generation: Any, master_config: MasterConfig) -> Any:
+def _maybe_start_generation_router(
+    base_urls: list[Optional[str]], master_config: MasterConfig
+) -> Any:
     """Start the NeMo-Gym-facing router, if enabled.
 
+    Args:
+        base_urls: OpenAI server URLs for the router.
+        master_config: SingleController MasterConfig.
+
     Returns:
         The router actor handle, or None when the router is disabled.
     """
@@ -656,7 +687,7 @@ def _maybe_start_generation_router(generation: Any, master_config: MasterConfig)
             flush=True,
         )
 
-    backend_urls = [url for url in (generation.dp_openai_server_base_urls or []) if url]
+    backend_urls = [url for url in (base_urls or []) if url]
     if not backend_urls:
         raise ValueError(
             "async_rl.generation_router.enabled=true requires generation backends that "
@@ -827,11 +858,13 @@ def setup_single_controller(
     # ==========================
     # TODO: add validate dataset wiring.
     use_nemo_gym = should_use_nemo_gym(master_config)
-    if use_nemo_gym and generation_config["backend"] != "vllm":
+    if use_nemo_gym and generation_config["backend"] not in ("vllm", "megatron"):
         raise NotImplementedError(
-            "SC NeMo-Gym integration currently supports the vllm backend "
-            f"only; got {generation_config['backend']!r}"
+            "SC NeMo-Gym integration currently supports the vllm and megatron backends only; got "
+            f"{generation_config['backend']!r}"
         )
+    # Backend settings checks are pure config: run them before anything builds.
+    resolve_generation_class(generation_config).validate_settings(master_config)
     if use_nemo_gym:
         # NeMo-Gym creates the env actor outside setup_response_data; we wire
         # it in after generation is up (it needs the OpenAI server URLs).
@@ -901,6 +934,12 @@ def setup_single_controller(
     # live generation to front. None is also the correct value whenever the router
     # is disabled or NeMo-Gym is not in play -- it is Gym that needs one stable URL.
     generation_router = None
+    megatron_backend = generation_config["backend"] == "megatron"
+    megatron_reserved_url = None
+    megatron_port_holder = None
+    reserved_http_server_port = None
+    if megatron_backend:
+        generation_config["model_name"] = master_config.policy["model_name"]
 
     def _build_trainer_and_value() -> tuple[Any, Optional[TQValue], dict[str, float]]:
         """Build the trainer, then the critic when this is a PPO run.
@@ -983,26 +1022,52 @@ def _build_generation_then_trainer(
         )
 
     if use_nemo_gym:
-        # defer generation, only get base_urls for nemo_gym spinup
-        generation, gen_reserve_time = _build_generation(
-            inference_cluster,
-            master_config=master_config,
-            defer_model_load=True,
-        )
-        defer_generation_model_load = True
+        if megatron_backend:
+            # Megatron serves from rank 0 of the generation workers; pre-publish that address.
+            t0 = time.perf_counter()
+            (
+                megatron_reserved_url,
+                reserved_http_server_port,
+                megatron_port_holder,
+            ) = MegatronGeneration.reserve_http_server_address(
+                inference_cluster,
+                master_config.policy,
+            )
+            gen_reserve_time = time.perf_counter() - t0
+            print(
+                f"  ✓ Reserved Megatron server URL: {megatron_reserved_url}",
+                flush=True,
+            )
+            gym_base_urls: list[Optional[str]] = [megatron_reserved_url]
+        else:
+            # defer generation, only get base_urls for nemo_gym spinup
+            generation, gen_reserve_time = _build_generation(
+                inference_cluster,
+                master_config=master_config,
+                defer_model_load=True,
+            )
+            defer_generation_model_load = True
+            gym_base_urls = generation.dp_openai_server_base_urls
         # Before the Gym task is built, so Gym can be handed the router's single URL.
-        generation_router = _maybe_start_generation_router(generation, master_config)
+        # These two statements are the only failable ones related to the port holder's creation.
+        try:
+            generation_router = _maybe_start_generation_router(
+                gym_base_urls, master_config
+            )
+            gym_spinup_base_urls = (
+                [ray.get(generation_router.base_url.remote())]
+                if generation_router is not None
+                else gym_base_urls
+            )
+        except BaseException:
+            if megatron_port_holder is not None:
+                ray.kill(megatron_port_holder)
+            raise
         # add nemo_gym spinup task
         build_tasks["nemo_gym"] = partial(
             _spinup_gym,
             master_config=master_config,
-            # The whole point of the router: Gym holds one NeMo-RL-owned URL and
-            # never has to fail over, which is the thing it cannot do.
-            base_urls=(
-                [ray.get(generation_router.base_url.remote())]
-                if generation_router is not None
-                else generation.dp_openai_server_base_urls
-            ),
+            base_urls=gym_spinup_base_urls,
             tokenizer=tokenizer,
         )
 
@@ -1025,20 +1090,54 @@ def _build_generation_then_trainer(
                 _build_generation,
                 inference_cluster=inference_cluster,
                 master_config=master_config,
+                reserved_http_server_port=reserved_http_server_port,
+                tokenizer=tokenizer,
+                processor=processor,
             )
         build_tasks["trainer"] = _build_trainer_and_value
 
     # Submit build tasks and get results
-    with ThreadPoolExecutor(max_workers=len(build_tasks)) as executor:
-        submitted = {k: executor.submit(fn) for k, fn in build_tasks.items()}
-        results = {k: f.result() for k, f in submitted.items()}
+    weight_synchronizer: Optional[WeightSynchronizer] = None
+    try:
+        with ThreadPoolExecutor(max_workers=len(build_tasks)) as executor:
+            submitted = {k: executor.submit(fn) for k, fn in build_tasks.items()}
+            if "generation_trainer" in submitted:
+                generation, trainer, value, time_metrics = submitted[
+                    "generation_trainer"
+                ].result()
+                gen_load_time = time_metrics["gen_time"]
+            else:
+                generation, gen_load_time = submitted["generation"].result()
+                trainer, value, time_metrics = submitted["trainer"].result()
+            if megatron_reserved_url is not None:
+                # Gym initialization needs a live URL that will respond to health checks.
+                # Megatron generation can only respond to health checks once initialized.
+                # The Megatron engine cannot be initialized with dummy weights.
+                # Thus, we must do an initial refit during initialization,
+                # before Gym can spin up.
+                t0 = time.perf_counter()
+                weight_synchronizer = create_weight_synchronizer(
+                    policy=trainer,
+                    generation=generation,
+                    generation_backend=generation_config["backend"],
+                    colocated=colocated,
+                    train_cluster=train_cluster,
+                    inference_cluster=inference_cluster,
+                    refit_buffer_size_gb=policy_config.get("refit_buffer_size_gb"),
+                )
+                weight_synchronizer.init_communicator()
+                setup_timing_metrics.collective_init_time_s = time.perf_counter() - t0
+                t0 = time.perf_counter()
+                weight_synchronizer.sync_weights()
+                setup_timing_metrics.weight_sync_time_s = time.perf_counter() - t0
+            if use_nemo_gym:
+                env_handles["nemo_gym"], gym_time = submitted["nemo_gym"].result()
+                setup_timing_metrics.nemo_gym_init_time_s = gym_time
+    finally:
+        if megatron_port_holder is not None:
+            # Rank 0 adopted (or will never adopt) the held socket; drop the holder.
+            ray.kill(megatron_port_holder)
 
-    if colocated:
-        generation, trainer, value, time_metrics = results["generation_trainer"]
-        gen_load_time = time_metrics["gen_time"]
-    else:
-        generation, gen_load_time = results["generation"]
-        trainer, value, time_metrics = results["trainer"]
     setup_timing_metrics.generation_init_time_s = gen_reserve_time + gen_load_time
 
     setup_timing_metrics.policy_init_time_s = time_metrics["trainer_time"]
@@ -1046,12 +1145,15 @@ def _build_generation_then_trainer(
         setup_timing_metrics.value_init_time_s = time_metrics["value_time"]
 
     if use_nemo_gym:
-        env_handles["nemo_gym"], gym_time = results["nemo_gym"]
-        setup_timing_metrics.nemo_gym_init_time_s = gym_time
         # the two fields are only meaningful when use_nemo_gym enabled
         setup_timing_metrics.generation_init_reserve_time_s = gen_reserve_time
         setup_timing_metrics.generation_init_load_time_s = gen_load_time
 
+    if megatron_reserved_url is not None:
+        MegatronGeneration.verify_served_address(
+            generation.dp_openai_server_base_urls, megatron_reserved_url
+        )
+
     # Loading a teacher with the same checkpoint as the student must happen only
     # after student initialization finishes: both use the same HF-to-Megatron
     # cache path, and concurrent conversion can expose a partial checkpoint.
@@ -1103,20 +1205,20 @@ def _build_generation_then_trainer(
         grpo_group_size=algo_cfg.num_generations_per_prompt,
     )
 
-    t0 = time.perf_counter()
-    weight_synchronizer = create_weight_synchronizer(
-        policy=trainer,
-        generation=generation,
-        generation_backend=generation_config["backend"],
-        colocated=colocated,
-        train_cluster=train_cluster,
-        inference_cluster=inference_cluster,
-        refit_buffer_size_gb=policy_config.get("refit_buffer_size_gb"),
-        # Only armed when configured; None leaves the refit path unchanged.
-        refit_timeout_s=master_config.async_rl.generation_fleet_health.refit_timeout_s,
-    )
-    weight_synchronizer.init_communicator()
-    setup_timing_metrics.collective_init_time_s = time.perf_counter() - t0
+    if weight_synchronizer is None:
+        t0 = time.perf_counter()
+        weight_synchronizer = create_weight_synchronizer(
+            policy=trainer,
+            generation=generation,
+            generation_backend=generation_config["backend"],
+            colocated=colocated,
+            train_cluster=train_cluster,
+            inference_cluster=inference_cluster,
+            refit_buffer_size_gb=policy_config.get("refit_buffer_size_gb"),
+            refit_timeout_s=master_config.async_rl.generation_fleet_health.refit_timeout_s,
+        )
+        weight_synchronizer.init_communicator()
+        setup_timing_metrics.collective_init_time_s = time.perf_counter() - t0
 
     # ==========================
     # Setup Algorithm + Rollout Wiring
diff --git a/nemo_rl/models/generation/__init__.py b/nemo_rl/models/generation/__init__.py
index 52b80e6d317..87c2d5de14a 100644
--- a/nemo_rl/models/generation/__init__.py
+++ b/nemo_rl/models/generation/__init__.py
@@ -16,7 +16,7 @@
 
 from transformers import PreTrainedTokenizerBase
 
-from nemo_rl.models.generation.interfaces import GenerationConfig
+from nemo_rl.models.generation.interfaces import GenerationConfig, GenerationInterface
 from nemo_rl.models.generation.trtllm import TrtllmConfig
 from nemo_rl.models.generation.vllm import VllmConfig
 from nemo_rl.models.generation.vllm.config import VLLM_SPARSE_REFIT_TRANSPORTS
@@ -24,6 +24,38 @@
 TokenizerType = PreTrainedTokenizerBase
 
 
+def resolve_generation_class(
+    generation_config: GenerationConfig,
+) -> type[GenerationInterface]:
+    """Map `generation_config` to its GenerationInterface class."""
+    backend = generation_config["backend"]
+    if backend == "vllm":
+        from nemo_rl.models.generation.vllm import VllmGeneration
+
+        return VllmGeneration
+    if backend == "sglang":
+        from nemo_rl.models.generation.sglang.sglang_generation import (
+            SGLangGeneration,
+        )
+
+        return SGLangGeneration
+    if backend == "megatron":
+        from nemo_rl.models.generation.megatron.megatron_generation import (
+            MegatronGeneration,
+        )
+
+        return MegatronGeneration
+    if backend == "trtllm":
+        from nemo_rl.models.generation.trtllm import TrtllmGeneration
+
+        return TrtllmGeneration
+    if backend == "dynamo":
+        from nemo_rl.models.generation.dynamo import DynamoGeneration
+
+        return DynamoGeneration
+    raise ValueError(f"Unknown generation backend: {backend!r}")
+
+
 def configure_generation_config(
     config: GenerationConfig,
     tokenizer: TokenizerType,
diff --git a/nemo_rl/models/generation/interfaces.py b/nemo_rl/models/generation/interfaces.py
index 5f867519b1d..bfd62367364 100644
--- a/nemo_rl/models/generation/interfaces.py
+++ b/nemo_rl/models/generation/interfaces.py
@@ -14,13 +14,16 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from functools import cache
-from typing import Any, NotRequired, Optional, TypedDict, Union
+from typing import TYPE_CHECKING, Any, NotRequired, Optional, TypedDict, Union
 
 import ray
 import torch
 
 from nemo_rl.distributed.batched_data_dict import BatchedDataDict
 
+if TYPE_CHECKING:
+    from nemo_rl.algorithms.single_controller_utils.config import MasterConfig
+
 # Routed-expert index tensors ([seq, layers, topk]) are carried in the narrowest
 # signed dtype that fits ids 0..num_experts-1 plus the -1 missing-route sentinel:
 # int8 for <=128 experts (e.g. Qwen3-MoE), int16 for <=32768 (e.g. DeepSeek-V3),
@@ -425,6 +428,14 @@ def reject_unenforceable_refit_deadline(
 class GenerationInterface(ABC):
     """Abstract base class defining the interface for RL policies."""
 
+    @classmethod
+    def validate_settings(cls, master_config: "MasterConfig") -> None:
+        """Backend-specific pure-config validation, run before any build.
+
+        Args:
+            master_config: The single-controller MasterConfig.
+        """
+
     @abstractmethod
     def init_collective(
         self, ip: str, port: int, world_size: int, *, train_world_size: int
diff --git a/nemo_rl/models/generation/megatron/config.py b/nemo_rl/models/generation/megatron/config.py
index 09753cc4f89..d79c4d5ba81 100644
--- a/nemo_rl/models/generation/megatron/config.py
+++ b/nemo_rl/models/generation/megatron/config.py
@@ -58,11 +58,8 @@ class MCoreGenerationSpecificArgs(TypedDict):
     # KV cache lifecycle across suspend/resume:
     # - "persist": cache stays allocated; CUDA graphs remain valid (default)
     # - "offload": cache is moved off-GPU between iterations
-    #
-    # The third mcore value, "recompute" (drop + rebuild on resume), must be set via
-    # `grpo.async_grpo.recompute_kv_cache_after_weight_updates=true`.
-    # TODO: Unify `kv_cache_management_mode` and `recompute_kv_cache_after_weight_updates`.
-    kv_cache_management_mode: Literal["persist", "offload"]
+    # - "recompute": cache is dropped and rebuilt on resume
+    kv_cache_management_mode: Literal["persist", "offload", "recompute"]
 
     logging_step_interval: NotRequired[int]
     # Whether MCore returns selected-token log-probs before or after sampling
@@ -84,10 +81,18 @@ class MCoreGenerationConfig(GenerationConfig):
 def merged_inference_megatron_cfg(policy_config: PolicyConfig) -> dict[str, Any]:
     """The `megatron_cfg` a dedicated inference model runs with."""
     generation_config = cast(MCoreGenerationConfig, policy_config["generation"])
+    overrides = dict(generation_config.get("mcore_generation_config") or {})
+    explicit_cp = overrides.pop("context_parallel_size", None)
+    if explicit_cp is not None and explicit_cp != 1:
+        raise ValueError(
+            "Megatron generation does not support context parallelism: remove "
+            "policy.generation.mcore_generation_config.context_parallel_size or set it to 1."
+        )
     merged: dict[str, Any] = {
         **cast(dict[str, Any], policy_config["megatron_cfg"]),
-        **(generation_config.get("mcore_generation_config") or {}),
+        **overrides,
         "activation_checkpointing": False,
+        "context_parallel_size": 1,
     }
     # inference_optimized layers hard-require SP with TP>1. Raise with the
     # config key: the colocated build bypasses validate_and_set_config, so this
@@ -114,13 +119,12 @@ def dedicated_inference_megatron_cfg(
     Colocated Megatron generation shares the training model unless the resolved
     inference layout or `transformer_impl` differs from training; then the worker
     builds a second model and reshards into it on every wake. Inference never
-    uses CP, so CP is pinned to 1 (CP>1 training therefore always differs).
+    uses CP, and CP is already pinned to 1 (CP>1 training therefore always differs).
 
     Returns None when the resolved config matches training (reshardless:
     generate directly on the shared training model).
     """
     inference_mcfg = merged_inference_megatron_cfg(policy_config)
-    inference_mcfg["context_parallel_size"] = 1
 
     train_mcfg = cast(dict[str, Any], policy_config["megatron_cfg"])
     layout_keys = (
diff --git a/nemo_rl/models/generation/megatron/megatron_generation.py b/nemo_rl/models/generation/megatron/megatron_generation.py
index 3eaffbc1748..7298ec95687 100644
--- a/nemo_rl/models/generation/megatron/megatron_generation.py
+++ b/nemo_rl/models/generation/megatron/megatron_generation.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import TYPE_CHECKING, Any, AsyncGenerator, Optional
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Optional, cast
 
 import ray
 from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
@@ -37,6 +37,8 @@
 from nemo_rl.weight_sync.interfaces import WeightSynchronizer
 
 if TYPE_CHECKING:
+    from nemo_rl.algorithms.single_controller_utils.config import MasterConfig
+    from nemo_rl.distributed.worker_groups import RayWorkerGroup
     from nemo_rl.models.policy.lm_policy import Policy
 
 
@@ -140,6 +142,56 @@ def reserve_http_server_address(
         node_ip, port = ray.get(holder.address.remote())
         return f"http://{node_ip}:{port}/v1", port, holder
 
+    @classmethod
+    def validate_settings(cls, master_config: "MasterConfig") -> None:
+        """Reject config the Megatron generation backend cannot honor."""
+        policy_config: PolicyConfig = master_config.policy
+        recompute_kv_cache_after_weight_updates: bool = (
+            master_config.async_rl.recompute_kv_cache_after_weight_updates
+        )
+        if not (
+            "megatron_cfg" in policy_config and policy_config["megatron_cfg"]["enabled"]
+        ):
+            raise ValueError(
+                "policy.generation.backend='megatron' requires the Megatron trainer "
+                "(policy.megatron_cfg.enabled=true): refit transfers weights via Megatron reshard "
+                "collective from the Megatron trainer."
+            )
+
+        mcore_cfg = cast(MCoreGenerationConfig, policy_config["generation"])[
+            "mcore_generation_config"
+        ]
+        # Recompute-after-refit is implemented engine-side (kv_cache_management_mode="recompute");
+        # the loop-level flag must agree with that mode, and setup errors on a mismatch.
+        kv_cache_mode = mcore_cfg["kv_cache_management_mode"]
+        if recompute_kv_cache_after_weight_updates != (kv_cache_mode == "recompute"):
+            raise ValueError(
+                "async_rl.recompute_kv_cache_after_weight_updates="
+                f"{recompute_kv_cache_after_weight_updates} conflicts with "
+                "policy.generation.mcore_generation_config."
+                f"kv_cache_management_mode={kv_cache_mode!r}: with "
+                "policy.generation.backend='megatron' the two must agree. Either "
+                "set the flag to true with kv_cache_management_mode='recompute', "
+                "or leave the flag false with 'persist'/'offload'."
+            )
+
+        if master_config.async_rl.generation_fleet_health.enabled:
+            raise NotImplementedError(
+                "async_rl.generation_fleet_health.enabled=true is not supported "
+                f"for the {cls.__name__} generation backend"
+            )
+
+    @classmethod
+    def verify_served_address(
+        cls, served_urls: list[Optional[str]], reserved_url: str
+    ) -> None:
+        """Fail loud if the engine serves anywhere but the pre-published address."""
+        if served_urls != [reserved_url]:
+            raise RuntimeError(
+                "Megatron server came up at a different address than the one "
+                f"pre-published to NeMo Gym: reserved {reserved_url}, serving {served_urls}."
+            )
+
     def __init__(
         self,
         config: PolicyConfig,
@@ -223,6 +275,11 @@ def __init__(
         if not skip_weight_load:
             self.prepare_for_generation()
 
+    @property
+    def worker_group(self) -> "RayWorkerGroup":
+        """The underlying policy's worker group (fleet-health probes read dp_size)."""
+        return self._policy.worker_group
+
     def init_collective(
         self,
         ip: str,
diff --git a/nemo_rl/models/generation/megatron/megatron_worker.py b/nemo_rl/models/generation/megatron/megatron_worker.py
index a34d0635201..8c9ee1f7c61 100644
--- a/nemo_rl/models/generation/megatron/megatron_worker.py
+++ b/nemo_rl/models/generation/megatron/megatron_worker.py
@@ -928,6 +928,10 @@ def init_collective_mcore_generation(
                 nccl_store, global_rank, world_size, nccl_options
             )
             nccl_backend._set_sequence_number_for_group()
+            # Create the group-wide NCCL communicator now, on every rank.
+            nccl_backend.eager_connect_single_device(
+                torch.device("cuda", torch.cuda.current_device())
+            )
             pg._register_backend(
                 torch.device("cuda"),
                 ProcessGroup.BackendType.NCCL,
diff --git a/tests/functional/L1_Functional_Tests_SingleController.sh b/tests/functional/L1_Functional_Tests_SingleController.sh
index aa259e47b89..751bf5a0ca0 100755
--- a/tests/functional/L1_Functional_Tests_SingleController.sh
+++ b/tests/functional/L1_Functional_Tests_SingleController.sh
@@ -37,6 +37,7 @@ run_test() {
 run_test fast uv run --no-sync bash ./tests/functional/grpo_dp_single_controller.sh
 run_test fast uv run --no-sync bash ./tests/functional/ppo_async_single_controller.sh
 run_test fast uv run --no-sync bash ./tests/functional/grpo_async_gym_single_controller.sh
+run_test fast uv run --no-sync bash ./tests/functional/grpo_megatron_generation_gym_single_controller.sh
 # Full mode only (~10 min): SIGKILLs a generation worker and asserts the job fails fast
 # and attributably instead of wedging. This is the ONLY end-to-end check of the
 # containment behaviour -- without it, a regression that restores the silent wedge is
diff --git a/tests/functional/grpo_megatron_generation_gym_single_controller.sh b/tests/functional/grpo_megatron_generation_gym_single_controller.sh
new file mode 100755
index 00000000000..e69c2c8017d
--- /dev/null
+++ b/tests/functional/grpo_megatron_generation_gym_single_controller.sh
@@ -0,0 +1,79 @@
+#!/bin/bash
+# SingleController + NeMo-Gym + Megatron generation e2e smoke.
+# Mirrors grpo_async_gym_single_controller.sh with the generation backend
+# swapped to non-colocated Megatron Inference. The config is loaded from the
+# exemplar YAML, which resolves value-equal to that test's CLI monolith
+# everywhere but the generation backend.
+
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd)
+PROJECT_ROOT=$(realpath $SCRIPT_DIR/../..)
+# Mark the current repo as safe, since wandb fetches metadata about the repo
+git config --global --add safe.directory $PROJECT_ROOT
+
+set -eou pipefail
+
+EXP_NAME=$(basename $0 .sh)
+EXP_DIR=$SCRIPT_DIR/$EXP_NAME
+LOG_DIR=$EXP_DIR/logs
+JSON_METRICS=$EXP_DIR/metrics.json
+RUN_LOG=$EXP_DIR/run.log
+CHECKPOINT_DIR=$EXP_DIR/checkpoints
+DATA_DIR=$EXP_DIR/data
+export PYTHONPATH=${PROJECT_ROOT}:${PYTHONPATH:-}
+
+rm -rf $EXP_DIR $LOG_DIR
+mkdir -p $EXP_DIR $LOG_DIR $CHECKPOINT_DIR $DATA_DIR
+
+# clean up checkpoint directory on exit
+trap "rm -rf $CHECKPOINT_DIR" EXIT
+
+cd $PROJECT_ROOT
+
+# Follow nemo-gym instructions here to get this data:
+# https://docs.nvidia.com/nemo/gym/0.1.0/tutorials/nemo-rl-grpo/setup.html#training-nemo-rl-grpo-setup
+cd 3rdparty/Gym-workspace/Gym
+
+# We need HF_TOKEN to download the data from huggingface
+if [[ ! -f env.yaml ]]; then
+    if [[ -z "${HF_TOKEN:-}" ]]; then
+        echo "[ERROR] HF_TOKEN is not set"
+        exit 1
+    fi
+    echo "hf_token: $HF_TOKEN" >> env.yaml
+fi
+
+uv run ng_prepare_data "+config_paths=[resources_servers/workplace_assistant/configs/workplace_assistant.yaml]" \
+    +output_dirpath=data/workplace_assistant \
+    +mode=train_preparation \
+    +should_download=true \
+    +data_source=huggingface
+cd -
+
+# This trimming of the workplace assistant dataset is necessary b/c with all the tools the first prompt is >4000 tokens
+# which will cause the generation engine to return nothing on the first prompt and crash RL. Since we want to keep this test short to
+# smoke test, we trim all but the first tool
+TRAIN_PATH=$DATA_DIR/workplace_assistant_train.jsonl
+VALIDATION_PATH=$DATA_DIR/workplace_assistant_validation.jsonl
+jq -c '.responses_create_params.tools |= (.[0:1])' 3rdparty/Gym-workspace/Gym/data/workplace_assistant/train.jsonl > $TRAIN_PATH
+jq -c '.responses_create_params.tools |= (.[0:1])' 3rdparty/Gym-workspace/Gym/data/workplace_assistant/validation.jsonl > $VALIDATION_PATH
+
+uv run coverage run -a --data-file=$PROJECT_ROOT/tests/.coverage --source=$PROJECT_ROOT/nemo_rl \
+    $PROJECT_ROOT/examples/run_grpo_single_controller.py \
+    --config $PROJECT_ROOT/examples/nemo_gym/grpo_qwen3_0_6b_megatron_generation_single_controller.yaml \
+    policy.generation.max_new_tokens=128 \
+    policy.max_total_sequence_length=512 \
+    logger.tensorboard_enabled=true \
+    logger.log_dir=$LOG_DIR \
+    logger.wandb_enabled=false \
+    logger.monitor_gpus=true \
+    data.train.data_path=$TRAIN_PATH \
+    data.validation.data_path=$VALIDATION_PATH \
+    $@ \
+    2>&1 | tee $RUN_LOG
+
+uv run tests/json_dump_tb_logs.py $LOG_DIR --output_path $JSON_METRICS
+
+# Observed to be between 0.8-1.3
+uv run tests/check_metrics.py $JSON_METRICS \
+    'median(data["train/gen_kl_error"]) < 1.3' \
+    'max(data["train/reward"]) > 0'
diff --git a/tests/test_suites/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync.sh b/tests/test_suites/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync.sh
new file mode 100755
index 00000000000..7b6b3f2c124
--- /dev/null
+++ b/tests/test_suites/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync.sh
@@ -0,0 +1,46 @@
+#!/bin/bash
+SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd)
+source $SCRIPT_DIR/common.env
+
+# ===== BEGIN CONFIG =====
+NUM_NODES=1
+# Megatron Inference decodes slower than the vLLM twin (which runs 450 steps
+# in 240 minutes); size like the classic megatron_generation nightlies.
+STEPS_PER_RUN=50
+MAX_STEPS=50
+NUM_RUNS=$(( (MAX_STEPS + STEPS_PER_RUN - 1) / STEPS_PER_RUN ))  # Round up
+NUM_MINUTES=180
+# ===== END CONFIG =====
+
+exit_if_max_steps_reached
+
+# Run the experiment
+cd $PROJECT_ROOT
+uv run examples/run_grpo_single_controller.py \
+    --config $CONFIG_PATH \
+    grpo.max_num_steps=$MAX_STEPS \
+    logger.log_dir=$LOG_DIR \
+    logger.wandb_enabled=True \
+    logger.wandb.project=nemo-rl \
+    logger.wandb.name=$EXP_NAME \
+    logger.monitor_gpus=True \
+    logger.tensorboard_enabled=True \
+    checkpointing.enabled=False \
+    checkpointing.checkpoint_dir=$CKPT_DIR \
+    $@ \
+    2>&1 | tee $RUN_LOG
+
+uv run tests/json_dump_tb_logs.py $LOG_DIR --output_path $JSON_METRICS
+
+# Only run metrics if the target step is reached
+if [[ $(jq 'to_entries | .[] | select(.key == "train/loss") | .value | keys | map(tonumber) | max' $JSON_METRICS) -ge $MAX_STEPS ]]; then
+    # Same logprob-health gate as the vLLM twin; no step-time assertion until
+    # a few runs calibrate Megatron Inference's cadence at this scale.
+    uv run tests/check_metrics.py $JSON_METRICS \
+        'median(data["train/token_mult_prob_error"]) < 1.1' \
+        'data["train/token_mult_prob_error"]["50"] < 1.1' \
+        'max(data["train/reward"]) > 0'
+
+    # Clean up checkpoint directory after successful run to save space.
+    rm -rf "$CKPT_DIR"
+fi
diff --git a/tests/test_suites/nightly.txt b/tests/test_suites/nightly.txt
index c396529e99e..19edb80e753 100644
--- a/tests/test_suites/nightly.txt
+++ b/tests/test_suites/nightly.txt
@@ -176,6 +176,7 @@ tests/test_suites/llm/prorlv2-qwen2.5-math-1.5b-instruct-1n8g-fsdp2tp1.v2-tq_moo
 # Single Controller (SC)
 tests/test_suites/llm/grpo-llama3.1-8b-instruct-2n8g-async-1off-single-controller-streaming2.sh
 tests/test_suites/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron-single-controller-sync.sh
+tests/test_suites/llm/grpo-qwen2.5-math-1.5b-instruct-1n8g-megatron_generation-noncolocated-single-controller-sync.sh
 tests/test_suites/llm/ppo-qwen2.5-1.5b-gsm8k-2n8g-megatron-valuetp2sp-dynbatch-noncolocated-async-single-controller.sh
 
 ########
diff --git a/tests/unit/algorithms/test_grpo.py b/tests/unit/algorithms/test_grpo.py
index bc5d4999577..f7388ad6ac2 100644
--- a/tests/unit/algorithms/test_grpo.py
+++ b/tests/unit/algorithms/test_grpo.py
@@ -3237,7 +3237,10 @@ def spinup_nemo_gym_actor(**kwargs):
             "enabled": False,
             "resources": {"gpus_per_node": 1, "num_nodes": 1},
         },
-        "mcore_generation_config": {"expose_http_server": True},
+        "mcore_generation_config": {
+            "expose_http_server": True,
+            "kv_cache_management_mode": "persist",
+        },
     }
     master_config.env = {"should_use_nemo_gym": True}
     master_config.loss_fn = ClippedPGLossConfig(reference_policy_kl_penalty=0.0)
diff --git a/tests/unit/reference_configs/grpo_math_1B.yaml b/tests/unit/reference_configs/grpo_math_1B.yaml
index c51186f9aea..f921ed76ec9 100644
--- a/tests/unit/reference_configs/grpo_math_1B.yaml
+++ b/tests/unit/reference_configs/grpo_math_1B.yaml
@@ -68,7 +68,7 @@ grpo:
     # when transient generation errors are expected and acceptable to drop.
     max_generation_failures: 0
     in_flight_weight_updates: false # Set to true to enable in-flight weight updates
-    recompute_kv_cache_after_weight_updates: false # Set to true to recompute kv cache after weight updates
+    recompute_kv_cache_after_weight_updates: false # Set to true to recompute kv cache after weight updates.
 
 # Reward-zeroing penalties applied to NeMo-Gym rollout results.
 reward_penalties:
@@ -395,7 +395,7 @@ policy:
       enable_chunked_prefill: true  # Split long prefills into chunks for better memory management
       enable_prefix_caching: false  # Reuse KV blocks across requests sharing a prompt prefix.
       max_tokens: 16384 # Maximum number of tokens to use in a single step. Analogous to vllm's max_num_batched_tokens
-      kv_cache_management_mode: "persist"  # KV cache lifecycle across suspend/resume. Options: "persist", "offload". To select "recompute", set grpo.async_grpo.recompute_kv_cache_after_weight_updates=true.
+      kv_cache_management_mode: "persist"  # KV cache lifecycle across suspend/resume. Options: "persist", "offload", "recompute".
       materialize_only_last_token_logits: true  # Materialize logits only for the last token of each sequence during decode.
       num_speculative_tokens: 0
       logprobs_mode: processed_logprobs  # Return log-probs after sampling processors. Use raw_logprobs for parity with policy recomputation.
diff --git a/tests/unit/single_controller/test_checkpointing.py b/tests/unit/single_controller/test_checkpointing.py
index 481df1ee695..1a2d6d5e263 100644
--- a/tests/unit/single_controller/test_checkpointing.py
+++ b/tests/unit/single_controller/test_checkpointing.py
@@ -230,6 +230,11 @@ def __init__(self) -> None:
         self.sync_count = 0
         self.shutdown_count = 0
 
+    @property
+    def is_stale(self) -> bool:
+        # WeightSynchronizer contract: stale until the first successful sync.
+        return self.sync_count == 0
+
     def sync_weights(self, *, kv_scales: Any = None) -> None:
         self.sync_count += 1
 
diff --git a/tests/unit/single_controller/test_setup.py b/tests/unit/single_controller/test_setup.py
index cd56bcf7653..110b07f421d 100644
--- a/tests/unit/single_controller/test_setup.py
+++ b/tests/unit/single_controller/test_setup.py
@@ -16,6 +16,8 @@
 
 from __future__ import annotations
 
+import contextlib
+import threading
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -39,8 +41,12 @@
 )
 from nemo_rl.data_plane.schema import SC_ROLLOUT_SCHEMA_FIELDS
 from nemo_rl.experience.rollouts import EffortLevelsConfig
+from nemo_rl.models.generation.megatron.megatron_generation import MegatronGeneration
 from nemo_rl.utils.config import load_config, register_omegaconf_resolvers
 
+# Captured at import, before the patched_factories fixture swaps it for a mock.
+_REAL_BUILD_GENERATION = sc_setup_mod._build_generation
+
 
 def _make_master_config(
     *,
@@ -62,6 +68,24 @@ def _make_master_config(
     normal load but unused here — model_construct skips validation, and we hand-fill
     only the dict-shaped fields setup reads.
     """
+    generation_config: dict = {
+        "backend": backend,
+        "colocated": {"enabled": colocated, "resources": {}},
+    }
+    policy_config: dict = {
+        "train_global_batch_size": num_prompts_per_step * 2,
+        "max_total_sequence_length": 32,
+        "tokenizer": {"use_fastokens": False},
+        "megatron_cfg": {"enabled": megatron_enabled},
+        "generation": generation_config,
+    }
+    if backend == "megatron":
+        # The megatron build path reads these before any generation factory runs.
+        generation_config["mcore_generation_config"] = {
+            "expose_http_server": False,
+            "kv_cache_management_mode": "persist",
+        }
+        policy_config["model_name"] = "test-model"
     return MasterConfig.model_construct(
         data_plane={"enabled": dp_enabled, "impl": "transfer_queue"},
         data={
@@ -81,16 +105,7 @@ def _make_master_config(
             val_at_start=False,
             val_at_end=False,
         ),
-        policy={
-            "train_global_batch_size": num_prompts_per_step * 2,
-            "max_total_sequence_length": 32,
-            "tokenizer": {"use_fastokens": False},
-            "megatron_cfg": {"enabled": megatron_enabled},
-            "generation": {
-                "backend": backend,
-                "colocated": {"enabled": colocated, "resources": {}},
-            },
-        },
+        policy=policy_config,
         # Full block: setup builds a CheckpointManager unconditionally (resume
         # lookup), which indexes these keys directly. Nothing is written while
         # enabled=False and the dir doesn't exist.
@@ -219,18 +234,6 @@ def test_build_generation_passes_sglang_config():
     generation.finish_generation.assert_called_once_with()
 
 
-def test_build_clusters_rejects_non_colocated_megatron_generation():
-    """The topology guard identifies Megatron as the generation backend."""
-    master_config = _make_master_config(colocated=False, backend="megatron")
-    master_config.cluster = {"num_nodes": 2, "gpus_per_node": 8}
-
-    with pytest.raises(
-        AssertionError,
-        match="Megatron generation backend.*non-colocated inference",
-    ):
-        sc_setup_mod._build_clusters(master_config)
-
-
 def test_build_clusters_rejects_unsupported_topology_backend(monkeypatch):
     """Topology planning reports the supported SC backends instead of KeyError."""
     master_config = _make_master_config(colocated=False, backend="trtllm")
@@ -254,7 +257,7 @@ def test_build_clusters_rejects_unsupported_topology_backend(monkeypatch):
 
     with pytest.raises(
         ValueError,
-        match="only supports vllm or sglang generation; got 'trtllm'",
+        match="only supports vllm, sglang, or megatron generation; got 'trtllm'",
     ):
         sc_setup_mod._build_clusters(master_config)
 
@@ -487,36 +490,70 @@ def create_teachers(*args, **kwargs):
         assert timings.teacher_model_init_time_s is not None
 
     @pytest.mark.parametrize(
-        ("invalid_case", "match"),
+        ("invalid_case", "expected_error", "match"),
         [
-            ("min_groups", "must be >="),
-            ("global_batch_size", "must equal policy.train_global_batch_size"),
-            ("buffer_capacity", "required capacity"),
+            ("min_groups", ValueError, "must be >="),
+            (
+                "global_batch_size",
+                ValueError,
+                "must equal policy.train_global_batch_size",
+            ),
+            ("buffer_capacity", ValueError, "required capacity"),
+            ("megatron_dtensor_trainer", ValueError, "megatron_cfg.enabled"),
+            ("megatron_recompute_mismatch", ValueError, "kv_cache_management_mode"),
+            ("megatron_fleet_health", NotImplementedError, "generation_fleet_health"),
+            ("gym_on_sglang", NotImplementedError, "vllm and megatron"),
         ],
     )
     def test_invalid_config_fails_before_setup_factories(
         self,
         invalid_case: str,
+        expected_error: type[Exception],
         match: str,
         patched_factories,
     ):
-        mc = _make_master_config()
+        use_gym = invalid_case == "gym_on_sglang"
         if invalid_case == "min_groups":
+            mc = _make_master_config()
             mc.async_rl.min_groups_for_streaming_train = 5
         elif invalid_case == "global_batch_size":
+            mc = _make_master_config()
             mc.policy["train_global_batch_size"] = 7
         elif invalid_case == "buffer_capacity":
+            mc = _make_master_config()
             mc.async_rl.max_buffered_rollouts = 7
+        elif invalid_case == "megatron_dtensor_trainer":
+            mc = _make_master_config(
+                colocated=False, backend="megatron", megatron_enabled=False
+            )
+        elif invalid_case == "megatron_recompute_mismatch":
+            # Flag says recompute; the engine mode (fixture default "persist") disagrees.
+            mc = _make_master_config(
+                colocated=False, backend="megatron", megatron_enabled=True
+            )
+            mc.async_rl.recompute_kv_cache_after_weight_updates = True
+        elif invalid_case == "megatron_fleet_health":
+            mc = _make_master_config(
+                colocated=False, backend="megatron", megatron_enabled=True
+            )
+            mc.async_rl.generation_fleet_health.enabled = True
+        elif invalid_case == "gym_on_sglang":
+            mc = _make_master_config(colocated=False, backend="sglang")
         else:  # pragma: no cover
             raise AssertionError(f"unknown test case {invalid_case}")
 
-        with pytest.raises(ValueError, match=match):
+        with (
+            patch.object(sc_setup_mod, "should_use_nemo_gym", return_value=use_gym),
+            patch.object(sc_setup_mod, "spinup_nemo_gym_actor") as mock_spinup,
+            pytest.raises(expected_error, match=match),
+        ):
             setup_single_controller(mc, MagicMock(pad_token_id=0))
 
         patched_factories["setup_response_data"].assert_not_called()
         patched_factories["_build_clusters"].assert_not_called()
         patched_factories["_build_generation"].assert_not_called()
         patched_factories["_build_trainer"].assert_not_called()
+        mock_spinup.assert_not_called()
 
     @pytest.mark.parametrize(
         ("loss_overrides", "match"),
@@ -972,9 +1009,197 @@ def test_nemo_gym_generation_init_time_includes_reserve_time(
         assert metrics.generation_init_reserve_time_s == 3.0
         assert metrics.generation_init_load_time_s is not None
 
-    @pytest.mark.parametrize("backend", ["sglang", "megatron"])
+    def _make_gym_megatron_config(self) -> MasterConfig:
+        mc = _make_master_config(
+            colocated=False, backend="megatron", megatron_enabled=True
+        )
+        mc.policy["generation"]["mcore_generation_config"]["expose_http_server"] = True
+        mc.policy["generation"]["stop_strings"] = None
+        mc.policy["generation"]["stop_token_ids"] = None
+        mc.policy["generation"]["top_k"] = None
+        return mc
+
+    @pytest.mark.parametrize(
+        ("scenario", "error_match"),
+        [
+            ("gym", None),
+            ("gym_served_mismatch", "different address"),
+            ("gym_router_failure", "router boom"),
+            ("native", None),
+        ],
+        ids=["gym", "gym_served_mismatch", "gym_router_failure", "native"],
+    )
+    def test_megatron_setup(
+        self, patched_factories, scenario: str, error_match: str | None
+    ):
+        """Non-colocated Megatron generation setup, gym and native legs.
+
+        gym: reserve rank-0's URL, spin Gym up on it, build trainer and engine
+        in parallel (the engine through _build_generation with the reserved
+        port), run the initial refit while Gym is still waiting -- the
+        skip-load engine only starts serving then -- cross-check the served
+        address, reap the port holder.
+        gym_served_mismatch: the served-vs-reserved cross-check fires after the
+        builds when the engine comes up on a different address.
+        gym_router_failure: the holder is created before the executor
+        try/finally that normally reaps it; a router-startup failure inside
+        that window must not leak the held socket.
+        native: expose_http_server=false and no Gym, so nothing reserves a URL,
+        no port holder is created, the cross-check is skipped, and the initial
+        refit is left to the actor.
+        """
+        gym = scenario != "native"
+        if gym:
+            mc = self._make_gym_megatron_config()
+            patched_factories["setup_response_data"].return_value = (
+                list(range(8)),
+                None,
+            )
+        else:
+            mc = _make_master_config(
+                colocated=False, backend="megatron", megatron_enabled=True
+            )
+        if scenario == "gym_router_failure":
+            mc.async_rl.generation_router.enabled = True
+        tokenizer = MagicMock(pad_token_id=0)
+        reserved_url = "http://10.0.0.1:5555/v1"
+        served_url = (
+            "http://10.0.0.9:7/v1"
+            if scenario == "gym_served_mismatch"
+            else reserved_url
+        )
+        port_holder = MagicMock(name="port_holder")
+        fake_gym_actor = MagicMock(name="nemo_gym_actor")
+        weight_sync = patched_factories["create_weight_synchronizer"].return_value
+        # Run the real _build_generation (MegatronGeneration is mocked below) so its
+        # Megatron branch is exercised, while the fixture mock still records the call.
+        patched_factories["_build_generation"].side_effect = _REAL_BUILD_GENERATION
+        # Gym's spinup only returns once the pre-published endpoint answers, and
+        # that endpoint comes up in the initial refit: block it on sync_weights so
+        # a setup that consumed the Gym task before refitting would hang here.
+        endpoint_up = threading.Event()
+        weight_sync.sync_weights.side_effect = lambda **_: endpoint_up.set()
+
+        def _spinup_gym(**_):
+            if not endpoint_up.wait(timeout=5):
+                raise TimeoutError("Gym was awaited before the initial refit")
+            return fake_gym_actor
+
+        # Real (disabled -> None) router startup on every leg but the failure one.
+        router_patch = (
+            patch.object(
+                sc_setup_mod,
+                "_maybe_start_generation_router",
+                side_effect=RuntimeError("router boom"),
+            )
+            if scenario == "gym_router_failure"
+            else contextlib.nullcontext()
+        )
+
+        with (
+            patch.object(sc_setup_mod, "should_use_nemo_gym", return_value=gym),
+            patch.object(
+                sc_setup_mod, "spinup_nemo_gym_actor", side_effect=_spinup_gym
+            ) as mock_spinup,
+            patch.object(sc_setup_mod, "router_replay_enabled", return_value=False),
+            patch.object(sc_setup_mod, "MegatronGeneration") as mock_megatron,
+            patch.object(sc_setup_mod, "ray") as mock_ray,
+            router_patch,
+        ):
+            mock_megatron.reserve_http_server_address.return_value = (
+                reserved_url,
+                5555,
+                port_holder,
+            )
+            # Wire the real check through the class mock so the
+            # served-vs-reserved legs exercise the genuine logic.
+            mock_megatron.verify_served_address = (
+                MegatronGeneration.verify_served_address
+            )
+            mock_megatron.return_value.dp_openai_server_base_urls = [served_url]
+            if error_match is None:
+                actor_args, metrics = setup_single_controller(mc, tokenizer)
+            else:
+                with pytest.raises(RuntimeError, match=error_match):
+                    setup_single_controller(mc, tokenizer)
+
+        inference_cluster = patched_factories["_build_clusters"].return_value[1]
+        assert mc.policy["generation"]["model_name"] == "test-model"
+        # Reservation + holder lifecycle exist on the gym legs only; every gym
+        # leg — success or either failure — reaps the holder exactly once.
+        if gym:
+            mock_megatron.reserve_http_server_address.assert_called_once_with(
+                inference_cluster,
+                mc.policy,
+            )
+            mock_ray.kill.assert_called_once_with(port_holder)
+        else:
+            mock_megatron.reserve_http_server_address.assert_not_called()
+            mock_ray.kill.assert_not_called()
+
+        if scenario == "gym_router_failure":
+            # Failed inside the reservation window: nothing downstream runs.
+            mock_spinup.assert_not_called()
+            patched_factories["_build_trainer"].assert_not_called()
+            patched_factories["_build_generation"].assert_not_called()
+            return
+
+        # Construction: trainer and generation are independent build tasks; the
+        # dedicated Megatron policy is built by _build_generation with the weight
+        # load skipped and the reserved port adopted (gym) or absent (native).
+        patched_factories["_build_trainer"].assert_called_once()
+        patched_factories["_build_generation"].assert_called_once()
+        mock_megatron.assert_called_once_with(
+            config=mc.policy,
+            tokenizer=tokenizer,
+            cluster=inference_cluster,
+            reserved_http_server_port=5555 if gym else None,
+            processor=None,
+            skip_weight_load=True,
+        )
+        # Stood down like every other backend; before the first refit this is a
+        # cache clear on the non-colocated Megatron workers.
+        mock_megatron.return_value.finish_generation.assert_called_once_with()
+        if gym:
+            # Gym spins up on the reserved URL, before the served-address
+            # cross-check — so the mismatch leg sees it too.
+            _, spinup_kwargs = mock_spinup.call_args
+            assert spinup_kwargs["base_urls"] == [reserved_url]
+            # The initial refit ran in setup, against the collective brought up
+            # there; the served-address check reads the URLs it populated.
+            weight_sync.init_communicator.assert_called_once_with()
+            weight_sync.sync_weights.assert_called_once_with()
+        else:
+            mock_spinup.assert_not_called()
+            # Native: the actor's startup sync performs the initial refit.
+            weight_sync.sync_weights.assert_not_called()
+        if scenario == "gym_served_mismatch":
+            return  # raised at the cross-check; no actor_args/metrics exist
+
+        assert actor_args.gen_handle is mock_megatron.return_value
+        assert actor_args.trainer_handle is patched_factories["fake_policy"]
+        assert actor_args.weight_synchronizer is weight_sync
+        assert metrics.generation_init_time_s is not None
+        assert metrics.policy_init_time_s is not None
+        assert metrics.collective_init_time_s is not None
+        patched_factories["create_weight_synchronizer"].assert_called_once()
+        _, factory_kwargs = patched_factories["create_weight_synchronizer"].call_args
+        assert factory_kwargs["generation_backend"] == "megatron"
+        assert factory_kwargs["colocated"] is False
+        assert factory_kwargs["inference_cluster"] is inference_cluster
+        if gym:
+            assert actor_args.env_handles["nemo_gym"] is fake_gym_actor
+            assert metrics.nemo_gym_init_time_s is not None
+            assert metrics.generation_init_reserve_time_s is not None
+            assert metrics.weight_sync_time_s is not None
+        else:
+            # Reserve/load split and setup-time sync exist on the gym-on path only.
+            assert metrics.generation_init_reserve_time_s is None
+            assert metrics.weight_sync_time_s is None
+
+    @pytest.mark.parametrize("backend", ["sglang"])
     def test_nemo_gym_rejects_non_vllm_backend(self, patched_factories, backend):
-        """SC nemo-gym wiring only supports vLLM; every other backend must raise."""
+        """SC nemo-gym wiring supports vllm and megatron; every other backend must raise."""
         mc = _make_master_config(backend=backend)
         patched_factories["setup_response_data"].return_value = (
             list(range(8)),
@@ -988,3 +1213,30 @@ def test_nemo_gym_rejects_non_vllm_backend(self, patched_factories, backend):
         ):
             setup_single_controller(mc, MagicMock(pad_token_id=0))
         mock_spinup.assert_not_called()
+
+    def test_megatron_fleet_health_rejected_with_clean_backend_error(self):
+        """megatron + generation_fleet_health fails naming the backend.
+
+        MegatronGeneration forwards ``worker_group`` to its policy, so
+        _maybe_attach_fleet_health survives its shard-count read and reaches
+        attach_fleet_health, whose base implementation rejects the backend by
+        name -- not an AttributeError on the monitor's constructor args.
+        """
+        mc = _make_master_config(
+            colocated=False, backend="megatron", megatron_enabled=True
+        )
+        mc.async_rl.generation_fleet_health.enabled = True
+        policy = MagicMock(name="policy")
+        policy.worker_group.dp_size = 2
+        generation = MegatronGeneration(
+            config=mc.policy,
+            tokenizer=MagicMock(),
+            policy=policy,
+        )
+        assert generation.worker_group is policy.worker_group
+
+        with pytest.raises(
+            NotImplementedError,
+            match="not supported for the MegatronGeneration generation backend",
+        ):
+            sc_setup_mod._maybe_attach_fleet_health(generation, mc)
diff --git a/tests/unit/test_recipes_and_test_suites.py b/tests/unit/test_recipes_and_test_suites.py
index 175942e5d51..220cef1dd03 100644
--- a/tests/unit/test_recipes_and_test_suites.py
+++ b/tests/unit/test_recipes_and_test_suites.py
@@ -256,7 +256,7 @@ def test_all_recipe_yamls_accounted_for_in_test_suites(
     )
 
 
-def test_nightly_compute_stays_below_4157_hours(nightly_test_suite, tracker):
+def test_nightly_compute_stays_below_4181_hours(nightly_test_suite, tracker):
     command = f"DRYRUN=1 HF_HOME=... HF_DATASETS_CACHE=... CONTAINER= ACCOUNT= PARTITION= ./tools/launch {' '.join(nightly_test_suite)}"
 
     print(f"Running command: {command}")
@@ -288,10 +288,8 @@ def test_nightly_compute_stays_below_4157_hours(nightly_test_suite, tracker):
         f"Last line of output was not as expected: '{last_line}'"
     )
     total_gpu_hours = float(last_line.split(":")[-1].strip())
-    # Reserve 8 GPU-hours for the SingleController MOPD nightly on top of
-    # main's 4149-hour limit.
-    assert total_gpu_hours <= 4157, (
-        f"Total GPU hours exceeded 4157: {last_line}. We should revisit the test suites to reduce the total GPU hours."
+    assert total_gpu_hours <= 4181, (
+        f"Total GPU hours exceeded 4181: {last_line}. We should revisit the test suites to reduce the total GPU hours."
     )
     tracker.track("total_nightly_gpu_hours", total_gpu_hours)