[cfg, recipe, tests] feat: wire DAPO overlong reward buffer on V1 omni loop

examples/dapo_trainer/README.md

@@ -19,6 +19,25 @@ policy loss with asymmetric clipping, GRPO advantages, and no KL penalty. The
 registered naive reward manager calls the AVQA `choice_reward`; the reward
 manager name alone does not select the optimization algorithm.
 
+**Overlong reward buffer.** Overlong shaping penalizes responses that run past
+`reward.reward_kwargs.max_resp_len`, tapering the reward to zero (then to a
+full penalty) over the trailing `reward.reward_kwargs.overlong_buffer_cfg.len`
+tokens. It is wired through `reward.reward_kwargs` and only applies with
+`reward.reward_manager.name=dapo` (`source=register`) — it is a no-op under
+the `naive` manager this example uses. Enable it with:
+
+```text
+reward.reward_kwargs.overlong_buffer_cfg.enable=true
+reward.reward_kwargs.overlong_buffer_cfg.len=<buffer_len>
+reward.reward_kwargs.overlong_buffer_cfg.penalty_factor=<factor>
+reward.reward_kwargs.max_resp_len=<max_response_length>
+```
+
+See `tests/special_e2e/run_dapo_qwen3_omni_thinker_lora_v1_smoke.sh` for a
+working `name=dapo` recipe with overlong shaping enabled, and
+`tests/utils/test_dapo_overlong_reward_on_cpu.py` for the reward-shape
+contract.
+
 ## Run
 
 Download and extract the AVQA-R1-6K data, then convert it from the repository

tests/special_e2e/run_dapo_qwen3_omni_thinker_lora_v1_smoke.sh

@@ -0,0 +1,118 @@
+#!/usr/bin/env bash
+# Qwen3-Omni Thinker DAPO + LoRA V1 smoke without dynamic sampling.
+
+set -xeuo pipefail
+
+if [[ "${SKIP_COMPAT_DEPS_INSTALL:-0}" != "1" ]]; then
+    uv pip install --system --break-system-packages transformers==5.12.1 accelerate==1.14.0 peft==0.19.1
+fi
+
+export NCCL_IB_DISABLE=1
+export CPATH=/usr/include${CPATH:+:$CPATH}
+export RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO=0
+export VERL_USE_EXTERNAL_MODULES=verl_omni
+
+NUM_GPUS=${NUM_GPUS:-2}
+MODEL_PATH=${MODEL_PATH:-}
+DATA_DIR=${DATA_DIR:-${HOME}/data/gsm8k}
+TOTAL_TRAIN_STEPS=${TOTAL_TRAIN_STEPS:-2}
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+EXCLUDE_MODULES=".*talker.*|.*code2wav.*|.*code_predictor.*|.*visual.*|.*audio_tower.*"
+
+MODEL_PATH="${MODEL_PATH:-${HOME}/models/tiny-random/Qwen3-Omni}"
+python3 "${REPO_ROOT}/tests/special_e2e/build_qwen3_omni_tiny_random.py" \
+    --output-dir "${MODEL_PATH}" --force
+
+if [ ! -f "${DATA_DIR}/train.parquet" ]; then
+    python3 "${REPO_ROOT}/tests/special_e2e/create_dummy_math_data.py" \
+        --local_save_dir "${DATA_DIR}"
+fi
+
+python3 -m verl_omni.trainer.main_omni \
+    data.train_files="${DATA_DIR}/train.parquet" \
+    data.val_files="${DATA_DIR}/test.parquet" \
+    data.train_batch_size=4 \
+    data.max_prompt_length=256 \
+    data.max_response_length=512 \
+    data.val_max_samples=4 \
+    data.truncation='error' \
+    data.filter_overlong_prompts=true \
+    actor_rollout_ref.model.path="${MODEL_PATH}" \
+    +actor_rollout_ref.model.override_config.attn_implementation=sdpa \
+    actor_rollout_ref.model.lora_rank=8 \
+    actor_rollout_ref.model.lora_alpha=16 \
+    actor_rollout_ref.model.lora_dtype=float32 \
+    actor_rollout_ref.model.lora.merge=true \
+    actor_rollout_ref.model.enable_gradient_checkpointing=true \
+    actor_rollout_ref.model.use_remove_padding=true \
+    actor_rollout_ref.model.exclude_modules="${EXCLUDE_MODULES}" \
+    actor_rollout_ref.model.target_modules="['q_proj','k_proj','v_proj','o_proj']" \
+    actor_rollout_ref.actor.freeze_vision_tower=true \
+    actor_rollout_ref.actor.strategy=fsdp2 \
+    actor_rollout_ref.actor.optim.lr=3e-6 \
+    actor_rollout_ref.actor.optim.weight_decay=0.01 \
+    actor_rollout_ref.actor.optim.clip_grad=1.0 \
+    actor_rollout_ref.actor.ppo_mini_batch_size=4 \
+    actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=1 \
+    actor_rollout_ref.actor.use_dynamic_bsz=true \
+    actor_rollout_ref.actor.ppo_max_token_len_per_gpu=20480 \
+    actor_rollout_ref.actor.use_kl_loss=false \
+    actor_rollout_ref.actor.entropy_coeff=0 \
+    actor_rollout_ref.actor.policy_loss.loss_mode=vanilla \
+    actor_rollout_ref.actor.clip_ratio_low=0.2 \
+    actor_rollout_ref.actor.clip_ratio_high=0.28 \
+    actor_rollout_ref.actor.clip_ratio_c=10.0 \
+    actor_rollout_ref.actor.loss_agg_mode=token-mean \
+    actor_rollout_ref.actor.fsdp_config.model_dtype=bfloat16 \
+    actor_rollout_ref.actor.fsdp_config.param_offload=true \
+    actor_rollout_ref.actor.fsdp_config.optimizer_offload=true \
+    actor_rollout_ref.rollout.name=vllm_omni \
+    actor_rollout_ref.rollout.n=2 \
+    actor_rollout_ref.rollout.temperature=0.8 \
+    actor_rollout_ref.rollout.tensor_model_parallel_size="${NUM_GPUS}" \
+    actor_rollout_ref.rollout.gpu_memory_utilization=0.4 \
+    actor_rollout_ref.rollout.max_num_seqs=16 \
+    actor_rollout_ref.rollout.load_format=safetensors \
+    actor_rollout_ref.rollout.log_prob_use_dynamic_bsz=true \
+    actor_rollout_ref.rollout.log_prob_max_token_len_per_gpu=20480 \
+    actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=2 \
+    actor_rollout_ref.rollout.enable_prefix_caching=false \
+    +actor_rollout_ref.rollout.engine_kwargs.vllm_omni.output_mode="ar" \
+    +actor_rollout_ref.rollout.engine_kwargs.vllm_omni.pipeline_name="qwen3_omni_moe" \
+    actor_rollout_ref.rollout.val_kwargs.n=1 \
+    actor_rollout_ref.rollout.val_kwargs.temperature=1.0 \
+    actor_rollout_ref.rollout.val_kwargs.top_p=0.7 \
+    actor_rollout_ref.ref.strategy=fsdp2 \
+    actor_rollout_ref.ref.log_prob_use_dynamic_bsz=true \
+    actor_rollout_ref.ref.log_prob_max_token_len_per_gpu=20480 \
+    actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=2 \
+    actor_rollout_ref.ref.fsdp_config.param_offload=true \
+    actor_rollout_ref.ref.fsdp_config.model_dtype=bfloat16 \
+    algorithm.trainer_type=policy_gradient \
+    algorithm.sample_source=online \
+    algorithm.adv_estimator=grpo \
+    algorithm.use_kl_in_reward=false \
+    algorithm.filter_groups.enable=false \
+    reward.reward_manager.source=register \
+    reward.reward_manager.name=dapo \
+    reward.reward_kwargs.max_resp_len=512 \
+    reward.reward_kwargs.overlong_buffer_cfg.enable=true \
+    reward.reward_kwargs.overlong_buffer_cfg.len=128 \
+    reward.reward_kwargs.overlong_buffer_cfg.penalty_factor=1.0 \
+    reward.reward_kwargs.overlong_buffer_cfg.log=true \
+    trainer.val_before_train=false \
+    trainer.balance_batch=true \
+    trainer.critic_warmup=0 \
+    trainer.logger=console \
+    trainer.project_name=verl-test \
+    trainer.experiment_name=dapo-qwen3-omni-thinker-lora-e2e-v1-wo-dynamic-sampling \
+    trainer.n_gpus_per_node="${NUM_GPUS}" \
+    trainer.nnodes=1 \
+    trainer.test_freq=1 \
+    trainer.save_freq=-1 \
+    trainer.resume_mode=disable \
+    trainer.total_training_steps="${TOTAL_TRAIN_STEPS}" \
+    "$@"
+
+echo "Qwen3-Omni Thinker DAPO+LoRA e2e V1 smoke without dynamic sampling passed."

tests/utils/test_dapo_overlong_reward_on_cpu.py

@@ -0,0 +1,96 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Overlong penalty must change the reward on a truncated response (#446 Phase 2)."""
+
+import numpy as np
+import torch
+from omegaconf import OmegaConf
+from tokenizers import Tokenizer
+from tokenizers.models import WordLevel
+from tokenizers.pre_tokenizers import Whitespace
+from transformers import AutoTokenizer, PreTrainedTokenizerFast
+from verl import DataProto
+from verl.experimental.reward_loop.reward_manager.dapo import DAPORewardManager
+
+MAX_RESP_LEN = 16
+OVERLONG_BUFFER_LEN = 4
+OVERLONG_PENALTY_FACTOR = 1.0
+
+
+def _build_local_tokenizer(tmp_path) -> AutoTokenizer:
+    # A tiny in-memory tokenizer, so this test needs no network access or
+    # downloaded artifacts (the DAPO manager only calls tokenizer.decode()).
+    vocab = {"[UNK]": 0, **{str(i): i + 1 for i in range(100)}}
+    tokenizer = Tokenizer(WordLevel(vocab=vocab, unk_token="[UNK]"))
+    tokenizer.pre_tokenizer = Whitespace()
+    fast_tokenizer = PreTrainedTokenizerFast(tokenizer_object=tokenizer, unk_token="[UNK]")
+    fast_tokenizer.save_pretrained(tmp_path)
+    return AutoTokenizer.from_pretrained(tmp_path)
+
+
+def _compute_score(data_source, solution_str, ground_truth, extra_info=None):
+    return 1.0
+
+
+def _build_manager(overlong_enable: bool, tokenizer) -> DAPORewardManager:
+    config = OmegaConf.create(
+        {
+            "reward": {
+                "reward_kwargs": {
+                    "overlong_buffer_cfg": {
+                        "enable": overlong_enable,
+                        "len": OVERLONG_BUFFER_LEN,
+                        "penalty_factor": OVERLONG_PENALTY_FACTOR,
+                        "log": True,
+                    },
+                    "max_resp_len": MAX_RESP_LEN,
+                }
+            }
+        }
+    )
+    return DAPORewardManager(config, tokenizer, _compute_score)
+
+
+def _make_truncated_response() -> DataProto:
+    # valid_len == MAX_RESP_LEN, i.e. exceed_len == OVERLONG_BUFFER_LEN -> full penalty.
+    response_ids = torch.randint(0, 100, (1, MAX_RESP_LEN))
+    attention_mask = torch.ones(1, MAX_RESP_LEN, dtype=torch.long)
+    non_tensors = {
+        "data_source": np.array(["dummy"], dtype=object),
+        "reward_model": np.array([{"ground_truth": "x"}], dtype=object),
+        "extra_info": np.array([{}], dtype=object),
+    }
+    return DataProto.from_dict(
+        tensors={"responses": response_ids, "attention_mask": attention_mask},
+        non_tensors=non_tensors,
+    )
+
+
+def test_overlong_penalty_changes_reward_on_truncated_response(tmp_path):
+    batch = _make_truncated_response()
+    tokenizer = _build_local_tokenizer(tmp_path)
+
+    disabled = _build_manager(overlong_enable=False, tokenizer=tokenizer)
+    enabled = _build_manager(overlong_enable=True, tokenizer=tokenizer)
+
+    result_disabled = disabled.loop.run_until_complete(disabled.run_single(batch))
+    result_enabled = enabled.loop.run_until_complete(enabled.run_single(batch))
+
+    assert result_disabled["reward_score"] == 1.0
+    assert "overlong" not in result_disabled["reward_extra_info"]
+
+    assert result_enabled["reward_score"] < result_disabled["reward_score"]
+    assert result_enabled["reward_extra_info"]["overlong"]
+    assert result_enabled["reward_extra_info"]["overlong_reward"] < 0

tests/utils/test_qwen3_omni_dapo_launcher_on_cpu.py

@@ -42,7 +42,6 @@ def _assert_dapo_without_dynamic_sampling_contract(script: str) -> None:
     assert set(DAPO_WITHOUT_DYNAMIC_SAMPLING_SETTINGS) <= settings
     assert "actor_rollout_ref.actor.policy_loss.loss_mode=gspo" not in settings
     assert "algorithm.filter_groups.enable=true" not in settings
-    assert "overlong_buffer_cfg" not in script
 
 
 def test_dapo_example_launcher_has_phase_one_contract():
@@ -78,3 +77,25 @@ def test_dapo_example_launcher_has_phase_one_contract():
     assert "data.val_max_samples=4" not in settings
     assert "data.validation_shuffle=true" not in settings
     assert "trainer.val_before_train=false" not in settings
+    assert "overlong_buffer_cfg" not in launcher
+
+
+def test_dapo_tiny_random_smoke_matches_example_contract():
+    repo_root = Path(__file__).parents[2]
+    smoke = (repo_root / "tests/special_e2e/run_dapo_qwen3_omni_thinker_lora_v1_smoke.sh").read_text(encoding="utf-8")
+
+    _assert_dapo_without_dynamic_sampling_contract(smoke)
+    settings = _script_settings(smoke)
+    assert "reward.reward_manager.name=dapo" in settings
+    assert "build_qwen3_omni_tiny_random.py" in smoke
+    assert "SKIP_COMPAT_DEPS_INSTALL:-0" in smoke
+    assert 'trainer.total_training_steps="${TOTAL_TRAIN_STEPS}"' in smoke
+
+    assert "data.max_response_length=512" in settings
+    assert {
+        "reward.reward_kwargs.max_resp_len=512",
+        "reward.reward_kwargs.overlong_buffer_cfg.enable=true",
+        "reward.reward_kwargs.overlong_buffer_cfg.len=128",
+        "reward.reward_kwargs.overlong_buffer_cfg.penalty_factor=1.0",
+        "reward.reward_kwargs.overlong_buffer_cfg.log=true",
+    } <= settings

verl_omni/trainer/config/_generated_diffusion_trainer.yaml

@@ -474,6 +474,13 @@ reward:
     use_accelerator: false
   reward_functions: {}
   aggregation: weighted_sum
+  reward_kwargs:
+    overlong_buffer_cfg:
+      enable: false
+      len: 0
+      penalty_factor: 0.0
+      log: true
+    max_resp_len: null
   reward_manager:
     _target_: verl.workers.config.reward_model.RewardManagerConfig
     source: importlib

verl_omni/trainer/config/_generated_diffusion_veomni_trainer.yaml

@@ -515,6 +515,13 @@ reward:
     use_accelerator: false
   reward_functions: {}
   aggregation: weighted_sum
+  reward_kwargs:
+    overlong_buffer_cfg:
+      enable: false
+      len: 0
+      penalty_factor: 0.0
+      log: true
+    max_resp_len: null
   reward_manager:
     _target_: verl.workers.config.reward_model.RewardManagerConfig
     source: importlib

verl_omni/trainer/config/_generated_omni_megatron_trainer.yaml

@@ -913,6 +913,13 @@ reward:
     use_accelerator: false
   reward_functions: {}
   aggregation: weighted_sum
+  reward_kwargs:
+    overlong_buffer_cfg:
+      enable: false
+      len: 0
+      penalty_factor: 0.0
+      log: true
+    max_resp_len: null
   reward_manager:
     _target_: verl.workers.config.reward_model.RewardManagerConfig
     source: register

verl_omni/trainer/config/_generated_omni_trainer.yaml

@@ -873,6 +873,13 @@ reward:
     use_accelerator: false
   reward_functions: {}
   aggregation: weighted_sum
+  reward_kwargs:
+    overlong_buffer_cfg:
+      enable: false
+      len: 0
+      penalty_factor: 0.0
+      log: true
+    max_resp_len: null
   reward_manager:
     _target_: verl.workers.config.reward_model.RewardManagerConfig
     source: importlib

verl_omni/trainer/config/reward/reward.yaml

@@ -31,6 +31,16 @@ reward_functions: {}
 # Aggregation method for multi-reward. Only "weighted_sum" is supported.
 aggregation: weighted_sum
 
+# Extra kwargs read directly by the reward manager (e.g. verl's "dapo" manager).
+reward_kwargs:
+  # max_resp_len must exceed overlong_buffer_cfg.len when enable is true.
+  overlong_buffer_cfg:
+    enable: false
+    len: 0
+    penalty_factor: 0.0
+    log: true
+  max_resp_len: null
+
 # reward manager configuration
 reward_manager:
   _target_: verl.workers.config.reward_model.RewardManagerConfig