diff --git a/examples/nemo_gym/grpo_nanov3.yaml b/examples/nemo_gym/grpo_nanov3.yaml index 55ecacc1b3d..cb74c313730 100644 --- a/examples/nemo_gym/grpo_nanov3.yaml +++ b/examples/nemo_gym/grpo_nanov3.yaml @@ -305,6 +305,8 @@ env: should_use_nemo_gym: true # true: skip expensive train_data_step*.jsonl; false: write full jsonl. should_log_nemo_gym_responses: true + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: # Port range for Gym HTTP servers, kept below the OS ephemeral range # (below the 9000 ephemeral floor on some GB200 nodes) and non-overlapping with NeMo diff --git a/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe1.yaml b/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe1.yaml index be6da9634cf..48f284cb711 100644 --- a/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe1.yaml +++ b/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe1.yaml @@ -219,6 +219,8 @@ data: env: should_use_nemo_gym: true should_log_nemo_gym_responses: false + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: skip_venv_if_present: true port_range_low: 5000 diff --git a/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe2.yaml b/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe2.yaml index e29b839065c..6764b43126d 100644 --- a/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe2.yaml +++ b/examples/nemo_gym/grpo_qwen3_30ba3b_thinking_swe2.yaml @@ -218,6 +218,8 @@ data: env: should_use_nemo_gym: true should_log_nemo_gym_responses: false + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: skip_venv_if_present: true port_range_low: 5000 diff --git a/examples/nemo_gym/grpo_workplace_assistant_nemotron_nano_v2_9b.yaml b/examples/nemo_gym/grpo_workplace_assistant_nemotron_nano_v2_9b.yaml index c7fe9c903ce..01194e2b770 100644 --- a/examples/nemo_gym/grpo_workplace_assistant_nemotron_nano_v2_9b.yaml +++ b/examples/nemo_gym/grpo_workplace_assistant_nemotron_nano_v2_9b.yaml @@ -292,6 +292,8 @@ env: should_use_nemo_gym: true # true: skip expensive train_data_step*.jsonl; false: write jsonl (heavy). Prefer true if disk is tight. should_log_nemo_gym_responses: true + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: # This is passed into NeMo-Gym as the initial_global_config_dict # Port range for Gym HTTP servers, kept below the OS ephemeral range # (below the 9000 ephemeral floor on some GB200 nodes) and non-overlapping with NeMo diff --git a/examples/nemo_gym/nemotron-3-super/stage1_rlvr.yaml b/examples/nemo_gym/nemotron-3-super/stage1_rlvr.yaml index a98b0abbb42..f9f48694aa8 100644 --- a/examples/nemo_gym/nemotron-3-super/stage1_rlvr.yaml +++ b/examples/nemo_gym/nemotron-3-super/stage1_rlvr.yaml @@ -281,6 +281,8 @@ env: should_use_nemo_gym: true # true: skip expensive train_data_step*.jsonl (recommended for large Gym runs); false: write full jsonl. should_log_nemo_gym_responses: true + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: uv_venv_dir: "${oc.env:PERSISTENT_CACHE}/gym_venvs" skip_venv_if_present: true diff --git a/examples/nemo_gym/nemotron-3-super/stage2_swe1.yaml b/examples/nemo_gym/nemotron-3-super/stage2_swe1.yaml index 9c7f3cdf68c..941b6327021 100644 --- a/examples/nemo_gym/nemotron-3-super/stage2_swe1.yaml +++ b/examples/nemo_gym/nemotron-3-super/stage2_swe1.yaml @@ -283,6 +283,8 @@ data: env: should_use_nemo_gym: true + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: uv_venv_dir: "${oc.env:PERSISTENT_CACHE}/gym_venvs" # overridden by super_launch.sh to $GYM_VENV_DIR skip_venv_if_present: true diff --git a/examples/nemo_gym/nemotron-3-super/stage2_swe2.yaml b/examples/nemo_gym/nemotron-3-super/stage2_swe2.yaml index d14b0d6685c..e714c93b852 100644 --- a/examples/nemo_gym/nemotron-3-super/stage2_swe2.yaml +++ b/examples/nemo_gym/nemotron-3-super/stage2_swe2.yaml @@ -279,6 +279,8 @@ sif_dir: "/path/to/sif" env: should_use_nemo_gym: true should_log_nemo_gym_responses: false + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: uv_venv_dir: "${oc.env:PERSISTENT_CACHE}/gym_venvs" # overridden by super_launch.sh to $GYM_VENV_DIR skip_venv_if_present: true diff --git a/examples/nemo_gym/nemotron-3-super/stage3_rlhf.yaml b/examples/nemo_gym/nemotron-3-super/stage3_rlhf.yaml index 501cccd82a1..b4a257b4912 100644 --- a/examples/nemo_gym/nemotron-3-super/stage3_rlhf.yaml +++ b/examples/nemo_gym/nemotron-3-super/stage3_rlhf.yaml @@ -282,6 +282,8 @@ env: should_use_nemo_gym: true # true: skip expensive train_data_step*.jsonl (recommended for large Gym runs); false: enable full jsonl. should_log_nemo_gym_responses: true + # false: ignore env mask_sample flags so the loss trains on every sample. + should_mask_flagged_samples: true nemo_gym: uv_venv_dir: "${oc.env:PERSISTENT_CACHE}/gym_venvs" # overridden by super_launch.sh to $GYM_VENV_DIR skip_venv_if_present: true diff --git a/nemo_rl/algorithms/async_utils/trajectory_collector.py b/nemo_rl/algorithms/async_utils/trajectory_collector.py index 452b6a0463b..c4c30990499 100644 --- a/nemo_rl/algorithms/async_utils/trajectory_collector.py +++ b/nemo_rl/algorithms/async_utils/trajectory_collector.py @@ -752,6 +752,7 @@ async def _iter_rollout_groups( from nemo_rl.experience.rollouts import ( get_nemo_gym_thinking_tags, run_async_nemo_gym_rollout, + should_mask_flagged_samples, ) # NeMo-Gym owns stop criteria. Configuration fills the policy EOS token @@ -777,6 +778,9 @@ async def _iter_rollout_groups( greedy=False, reward_penalty_config=self.master_config.reward_penalties, thinking_tags=get_nemo_gym_thinking_tags(self.master_config.env), + mask_env_flagged_samples=should_mask_flagged_samples( + self.master_config.env + ), ): task_index = rollout_result.task_index if task_index is None: diff --git a/nemo_rl/algorithms/grpo.py b/nemo_rl/algorithms/grpo.py index 137c9dbd621..44231c71579 100644 --- a/nemo_rl/algorithms/grpo.py +++ b/nemo_rl/algorithms/grpo.py @@ -90,6 +90,7 @@ run_async_multi_turn_rollout, run_multi_turn_rollout, run_nemo_gym_rollout_sync, + should_mask_flagged_samples, ) from nemo_rl.models.generation.interfaces import ( GenerationConfig, @@ -2803,6 +2804,9 @@ def grpo_train( effort_config=_get_effort_config(master_config), reward_penalty_config=master_config.reward_penalties, thinking_tags=get_nemo_gym_thinking_tags(master_config.env), + mask_env_flagged_samples=should_mask_flagged_samples( + master_config.env + ), ) input_ids = nemo_gym_rollout_result.input_ids repeated_batch = nemo_gym_rollout_result.final_batch @@ -3654,6 +3658,9 @@ def validate( effort_config=_get_effort_config(master_config), reward_penalty_config=master_config.reward_penalties, thinking_tags=get_nemo_gym_thinking_tags(master_config.env), + mask_env_flagged_samples=should_mask_flagged_samples( + master_config.env + ), ) val_batch = nemo_gym_rollout_result.final_batch gen_metrics = nemo_gym_rollout_result.rollout_metrics diff --git a/nemo_rl/algorithms/single_controller_utils/setup.py b/nemo_rl/algorithms/single_controller_utils/setup.py index 6ad8d107a3f..eab9bdc4ef5 100644 --- a/nemo_rl/algorithms/single_controller_utils/setup.py +++ b/nemo_rl/algorithms/single_controller_utils/setup.py @@ -49,6 +49,7 @@ from nemo_rl.environments.interfaces import EnvironmentInterface from nemo_rl.environments.nemo_gym import spinup_nemo_gym_actor from nemo_rl.experience.rollout_manager import RolloutManager +from nemo_rl.experience.rollouts import should_mask_flagged_samples from nemo_rl.models.generation.interfaces import ( resolve_routed_experts_dtype_name_for_model, ) @@ -441,6 +442,7 @@ def setup_single_controller( policy_generation=generation, generation_config=generation_config, use_nemo_gym=use_nemo_gym, + mask_env_flagged_samples=should_mask_flagged_samples(master_config.env), tq_buffer=tq_buffer, ) diff --git a/nemo_rl/experience/rollout_manager.py b/nemo_rl/experience/rollout_manager.py index a1f485bf5eb..6f79b3d0da7 100644 --- a/nemo_rl/experience/rollout_manager.py +++ b/nemo_rl/experience/rollout_manager.py @@ -405,6 +405,7 @@ def __init__( max_seq_len: int, max_rollout_turns: int, generation_config: GenerationConfig, + mask_env_flagged_samples: bool = True, **kwargs: Any, ) -> None: self._tokenizer = tokenizer @@ -413,6 +414,7 @@ def __init__( self._max_seq_len = max_seq_len self._max_rollout_turns = max_rollout_turns self._generation_config = generation_config + self._mask_env_flagged_samples = mask_env_flagged_samples self._validate_init_params() @@ -553,6 +555,13 @@ def _result_to_completion(self, result: dict) -> Completion: sum(len(m["token_ids"]) for m in result["message_log"]) == self._max_seq_len ) + # Same gate as the batched path: when masking is off, drop the env + # mask flag so later batch building never sees it. + if not self._mask_env_flagged_samples: + (result["full_result"].get("instance_config") or {}).pop( + "mask_sample", None + ) + return Completion( message_log=result["message_log"], env_extras=result["full_result"], @@ -651,6 +660,7 @@ def __init__( policy_generation: Optional[GenerationInterface] = None, generation_config: Optional[GenerationConfig] = None, use_nemo_gym: bool = False, + mask_env_flagged_samples: bool = True, tq_buffer: Optional[TQReplayBuffer] = None, ) -> None: assert num_generations_per_prompt >= 1, ( @@ -676,6 +686,8 @@ def __init__( max_rollout_turns=max_rollout_turns, policy_generation=policy_generation, # type: ignore generation_config=generation_config, + # Only used by AsyncNemoGymRolloutImpl; AsyncRolloutImpl ignores it. + mask_env_flagged_samples=mask_env_flagged_samples, ) self._tokenizer = tokenizer self._num_generations_per_prompt = num_generations_per_prompt diff --git a/nemo_rl/experience/rollouts.py b/nemo_rl/experience/rollouts.py index d08d6af19e9..882711e838f 100644 --- a/nemo_rl/experience/rollouts.py +++ b/nemo_rl/experience/rollouts.py @@ -1558,6 +1558,21 @@ def get_nemo_gym_thinking_tags(env_config: dict[str, Any]) -> list[str]: return list(DEFAULT_THINKING_TAGS) +def should_mask_flagged_samples(env_config: dict[str, Any]) -> bool: + """Read ``env.should_mask_flagged_samples``; absent means True. + + True (the default): env-driven ``mask_sample`` flags are carried in the + rollout batch and flagged samples are dropped from the loss. + + Set false when the flags are too coarse to honor: for example, Gym flags + rollouts that hit max iterations even when they solve the task, and those + are samples worth training on. It also keeps batch composition + deterministic for controlled benchmark runs — how many samples get + flagged varies run to run. + """ + return env_config.get("should_mask_flagged_samples") is not False + + def _get_reward_penalty_config_value( reward_penalty_config: dict[str, Any] | BaseModel | None, key: str, @@ -1997,6 +2012,7 @@ async def run_async_nemo_gym_rollout( effort_config: Optional[EffortLevelsConfig] = None, reward_penalty_config: dict[str, Any] | BaseModel | None = None, thinking_tags: list[str] | tuple[str, ...] | None = None, + mask_env_flagged_samples: bool = True, returns_entire_batch: bool = False, ) -> AsyncGenerator[NemoGymRolloutResult, None]: """Stream complete NeMo-Gym prompt groups in group-completion order. @@ -2023,6 +2039,8 @@ async def run_async_nemo_gym_rollout( effort_config: Optional configuration for effort-based reward shaping. reward_penalty_config: Optional reward-penalty configuration. thinking_tags: Optional opening and closing tags used by thinking penalties. + mask_env_flagged_samples: Whether to carry env-driven ``mask_sample`` + flags in the rollout batch for loss masking. returns_entire_batch: Whether to treat the input as one potentially heterogeneous group. This requires ``num_generations`` to equal the batch size and is used by synchronous callers. @@ -2148,6 +2166,7 @@ async def run_async_nemo_gym_rollout( effort_config=effort_config, reward_penalty_config=reward_penalty_config, thinking_tags=thinking_tags, + mask_env_flagged_samples=mask_env_flagged_samples, ) if accumulator.is_complete: final_rollout_result = rollout_result @@ -2184,6 +2203,7 @@ def run_nemo_gym_rollout_sync( effort_config: Optional[EffortLevelsConfig] = None, reward_penalty_config: dict[str, Any] | BaseModel | None = None, thinking_tags: list[str] | tuple[str, ...] | None = None, + mask_env_flagged_samples: bool = True, ) -> NemoGymRolloutResult: """Run and return one complete NeMo-Gym batch synchronously. @@ -2206,6 +2226,8 @@ def run_nemo_gym_rollout_sync( effort_config: Optional configuration for effort-based reward shaping. reward_penalty_config: Optional reward-penalty configuration. thinking_tags: Optional opening and closing tags used by thinking penalties. + mask_env_flagged_samples: Whether to carry env-driven ``mask_sample`` + flags in the rollout batch for loss masking. Returns: The fully postprocessed NeMo-Gym rollout batch in input-row order. @@ -2235,6 +2257,7 @@ async def _consume_rollout() -> NemoGymRolloutResult: effort_config=effort_config, reward_penalty_config=reward_penalty_config, thinking_tags=thinking_tags, + mask_env_flagged_samples=mask_env_flagged_samples, returns_entire_batch=True, ): pass @@ -2257,6 +2280,7 @@ def _postprocess_single_nemo_gym_group( effort_config: Optional[EffortLevelsConfig] = None, reward_penalty_config: dict[str, Any] | BaseModel | None = None, thinking_tags: list[str] | tuple[str, ...] | None = None, + mask_env_flagged_samples: bool = True, ) -> NemoGymRolloutResult: """Postprocess one complete prompt group from the NeMo-Gym stream.""" # Length-based reward shaping for low-effort prompts @@ -2437,11 +2461,12 @@ def _postprocess_single_nemo_gym_group( "truncated": torch.tensor( [m["hit_max_tokens"] for m in all_sample_metrics], dtype=torch.bool ), - # Agent/env-driven mask flag — True means this sample should be masked - # from the GRPO gradient (kept for advantage computation). - "mask_sample": _extract_mask_sample_flags(results), } ) + # Env/agent mask flag: flagged samples are dropped from the loss but still + # count for advantages. env.should_mask_flagged_samples=false skips this. + if mask_env_flagged_samples: + final_batch["mask_sample"] = _extract_mask_sample_flags(results) if length_rewards_low: rollout_metrics["mean_length_reward_low"] = sum(length_rewards_low) / len( diff --git a/tests/unit/experience/test_reward_penalties.py b/tests/unit/experience/test_reward_penalties.py index 14db46dee5d..ea5f4921519 100644 --- a/tests/unit/experience/test_reward_penalties.py +++ b/tests/unit/experience/test_reward_penalties.py @@ -17,11 +17,15 @@ import pytest import torch +from nemo_rl.distributed.batched_data_dict import BatchedDataDict from nemo_rl.experience.rollouts import ( _extract_mask_sample_flags, + _postprocess_single_nemo_gym_group, apply_reward_penalties, resolve_reward_penalty_config, + should_mask_flagged_samples, ) +from nemo_rl.utils.timer import Timer # ---- Helpers to build minimal result dicts ---- @@ -79,6 +83,7 @@ def _msg(role, token_ids, **extra): class _FakeTokenizer: def __init__(self, eos_token_id=2, token_map=None): self.eos_token_id = eos_token_id + self.pad_token_id = 0 self.token_map = token_map or {"": [12], "": [13]} def encode(self, text, add_special_tokens=False): @@ -104,6 +109,66 @@ def test_reads_mask_sample_from_instance_config(self): ) +class TestShouldMaskFlaggedSamples: + def test_reads_env_should_mask_flagged_samples(self): + assert should_mask_flagged_samples({}) is True + assert ( + should_mask_flagged_samples({"should_mask_flagged_samples": None}) is True + ) + assert ( + should_mask_flagged_samples({"should_mask_flagged_samples": True}) is True + ) + assert ( + should_mask_flagged_samples({"should_mask_flagged_samples": False}) is False + ) + + +class _FakeGeneration: + cfg = {"max_total_sequence_length": 100} + + +def _gate_result(mask_sample): + return { + "full_result": { + "reward": 1.0, + "response": {"output": []}, + "instance_config": {"mask_sample": mask_sample}, + }, + "message_log": [ + {"role": "user", "token_ids": torch.tensor([3, 4])}, + {"role": "assistant", "token_ids": torch.tensor([1, 2])}, + ], + "input_message_log": [{"role": "user", "token_ids": torch.tensor([3, 4])}], + } + + +class TestMaskEnvFlaggedSamplesBatchedGate: + """The batched NeMo-Gym postprocess gate at _postprocess_single_nemo_gym_group.""" + + def _final_batch(self, mask_env_flagged_samples): + results = [_gate_result(True), _gate_result(False)] + nemo_gym_rows = [{"agent_ref": {"name": "agent"}} for _ in results] + rollout_result = _postprocess_single_nemo_gym_group( + nemo_gym_rows=nemo_gym_rows, + results=results, + timer=Timer(), + timer_prefix="timing/test", + policy_generation=_FakeGeneration(), + input_batch=BatchedDataDict({"loss_multiplier": torch.ones(2)}), + tokenizer=_FakeTokenizer(), + log_full_result_tables=False, + mask_env_flagged_samples=mask_env_flagged_samples, + ) + return rollout_result.final_batch + + def test_gate_on_carries_mask_sample(self): + final_batch = self._final_batch(True) + assert torch.equal(final_batch["mask_sample"], torch.tensor([True, False])) + + def test_gate_off_omits_mask_sample(self): + assert "mask_sample" not in self._final_batch(False) + + # ===================================================================== # Penalty 1: penalize_duplicated_reasoning # ===================================================================== diff --git a/tests/unit/experience/test_rollout_manager.py b/tests/unit/experience/test_rollout_manager.py index c4dfc5fe781..a0879ddb3bf 100644 --- a/tests/unit/experience/test_rollout_manager.py +++ b/tests/unit/experience/test_rollout_manager.py @@ -39,7 +39,10 @@ from nemo_rl.data.processors import nemo_gym_data_processor from nemo_rl.distributed.batched_data_dict import BatchedDataDict from nemo_rl.experience.interfaces import Completion, PromptGroupRecord -from nemo_rl.experience.rollout_manager import RolloutManager +from nemo_rl.experience.rollout_manager import ( + AsyncNemoGymRolloutImpl, + RolloutManager, +) from nemo_rl.experience.rollouts import ( run_async_multi_turn_rollout, run_async_nemo_gym_rollout, @@ -302,6 +305,69 @@ def test_rollout_manager_raises_without_impl_params(): RolloutManager(**common, use_nemo_gym=True) +def test_rollout_manager_forwards_mask_env_flagged_samples(): + """env.should_mask_flagged_samples reaches the NeMo-Gym impl through RolloutManager.""" + common = { + "tokenizer": None, + "task_to_env": {}, + "num_generations_per_prompt": 1, + "max_seq_len": 1, + "generation_config": { + "stop_strings": None, + "stop_token_ids": None, + "top_k": None, + }, + "use_nemo_gym": True, + } + + assert RolloutManager(**common)._impl._mask_env_flagged_samples is True + manager = RolloutManager(**common, mask_env_flagged_samples=False) + assert manager._impl._mask_env_flagged_samples is False + + +def _nemo_gym_impl(mask_env_flagged_samples): + return AsyncNemoGymRolloutImpl( + tokenizer=None, + task_to_env={}, + num_generations_per_prompt=1, + max_seq_len=100, + max_rollout_turns=1, + generation_config={ + "stop_strings": None, + "stop_token_ids": None, + "top_k": None, + }, + mask_env_flagged_samples=mask_env_flagged_samples, + ) + + +def _mask_gate_result(): + return { + "message_log": [ + { + "role": "assistant", + "token_ids": [1, 2], + "generation_logprobs": [0.0, 0.0], + } + ], + "full_result": { + "reward": 1.0, + "instance_config": {"mask_sample": True, "other_key": "kept"}, + }, + } + + +def test_result_to_completion_keeps_mask_flag_when_gate_on(): + completion = _nemo_gym_impl(True)._result_to_completion(_mask_gate_result()) + assert completion.env_extras["instance_config"]["mask_sample"] is True + + +def test_result_to_completion_drops_mask_flag_when_gate_off(): + completion = _nemo_gym_impl(False)._result_to_completion(_mask_gate_result()) + assert "mask_sample" not in completion.env_extras["instance_config"] + assert completion.env_extras["instance_config"]["other_key"] == "kept" + + # --------------------------------------------------------------------------- # Tests for AsyncRolloutManager (native async path) # ---------------------------------------------------------------------------