fix(async): gate rollout batches on validation pause

nemo_rl/algorithms/async_utils/trajectory_collector.py

@@ -547,6 +547,15 @@ def _run_collection_loop(
                     if not self.running:
                         break
 
+                # Refit and weight updates can wake this loop after validation
+                # pauses it. Check again before starting a batch.
+                if not self._manual_pause_cleared.is_set() and self.running:
+                    with (
+                        efficiency_span("idle/validation_pause", tracer=self._tracer),
+                        self._efficiency_timer.time("idle/validation_pause"),
+                    ):
+                        self._manual_pause_cleared.wait()
+
                 if not self.running:
                     break
 

nemo_rl/algorithms/grpo.py

@@ -5406,6 +5406,16 @@ def _flush_collector_telemetry() -> None:
                     and policy_generation.wake_carries_weight_updates()
                 )
 
+                should_run_validation = (
+                    val_period > 0
+                    and (step + 1) >= val_start_at
+                    and (step + 1) % val_period == 0
+                ) or (val_at_end and is_last_step)
+                if should_run_validation:
+                    # Stop dispatch before refit wakes the collector. This also
+                    # separates the training and validation payload metrics.
+                    ray.get(trajectory_collector.pause.remote())
+
                 print("🔄 Synchronizing policy weights to trajectory collector…")
                 if defer_wake_for_save:
                     # Wake-deferral (checkpoint scheduling, which the backend
@@ -5473,17 +5483,9 @@ def _flush_collector_telemetry() -> None:
 
                 # Validation
                 val_metrics, validation_timings = None, None
-                should_run_validation = (
-                    val_period > 0
-                    and (step + 1) >= val_start_at
-                    and (step + 1) % val_period == 0
-                ) or (val_at_end and is_last_step)
 
                 payload_metrics: dict[str, int | float] = {}
                 if should_run_validation:
-                    # Stop new dispatch before separating the training and
-                    # validation payload-metric intervals.
-                    ray.get(trajectory_collector.pause.remote())
                     if master_config.grpo.debug_payload_metrics:
                         payload_metrics = merge_multimodal_payload_metrics(
                             [

tests/unit/algorithms/test_async_utils.py

@@ -1377,6 +1377,53 @@ def test_collection_loop_marks_data_exhausted_on_natural_completion(self):
         assert status["errored"] is False
         assert status["running"] is False
 
+    def test_collection_loop_defers_new_batch_while_manually_paused(self):
+        """Keep a weight update from bypassing a manual pause."""
+        collector = self.create_local_collector()
+        collector.running = True
+        processed = []
+        batch_processed = threading.Event()
+        should_pause_calls = []
+
+        def _fake_should_pause():
+            # Park on the first check. The weight update below wakes the loop.
+            should_pause_calls.append(1)
+            return len(should_pause_calls) == 1
+
+        collector._should_pause_for_generation_limits = _fake_should_pause
+
+        def _record_batch(batch):
+            processed.append(batch)
+            batch_processed.set()
+
+        collector._process_batch = _record_batch
+        collector.dataloader = [{"b": 0}]
+
+        loop_thread = threading.Thread(target=collector._collection_loop, daemon=True)
+        loop_thread.start()
+
+        deadline = time.time() + 5.0
+        while collector._generation_limit_cleared.is_set():
+            assert time.time() < deadline, "loop never reached generation-limit wait"
+            time.sleep(0.01)
+
+        collector.pause()
+        collector.set_weight_version(1)
+
+        assert not batch_processed.wait(0.5), (
+            "_process_batch ran while the collector was paused for validation"
+        )
+        assert processed == []
+
+        collector.resume()
+        assert batch_processed.wait(5.0), "batch not processed after resume"
+        assert processed == [{"b": 0}]
+
+        loop_thread.join(5.0)
+        assert not loop_thread.is_alive()
+        assert collector.data_exhausted is True
+        assert collector.collection_failed is False
+
     @pytest.mark.asyncio
     async def test_drain_payload_metrics_returns_collector_interval(self, monkeypatch):
         collector = self.create_local_collector()