diff --git a/src/prime_rl/trainer/batch.py b/src/prime_rl/trainer/batch.py
index 2c42fc2e73..52571d6ed8 100644
--- a/src/prime_rl/trainer/batch.py
+++ b/src/prime_rl/trainer/batch.py
@@ -196,6 +196,8 @@ def prepare_sample(training_example: TrainingSample, seq_len: int) -> MicroBatch
         assert len(mm_token_type_ids) == len(input_ids), (
             f"mm_token_type_ids: {len(mm_token_type_ids)}, input_ids: {len(input_ids)}"
         )
+    if mm_kwargs is not None and "image_grid_thw" in mm_kwargs and mm_token_type_ids is None:
+        raise ValueError("image_grid_thw multimodal samples require mm_token_type_ids")
     assert len(env_names) == len(input_ids), f"env_names: {len(env_names)}, input_ids: {len(input_ids)}"
 
     return MicroBatch(
@@ -214,6 +216,7 @@ def prepare_sample(training_example: TrainingSample, seq_len: int) -> MicroBatch
         rl_weights=rl_weights,
         ce_weights=ce_weights,
         ref_kl_weights=ref_kl_weights,
+        seq_lens=[len(input_ids)],
     )
 
 
@@ -237,15 +240,14 @@ def first_sample(self) -> MicroBatch:
 
     def can_add(self, sample: MicroBatch, max_seq_len: int) -> bool:
         # Loss routing is per token (component weight streams), so samples of
-        # different loss types pack together freely — only modality, length and
-        # routed-experts presence constrain packing.
+        # different loss types pack together freely. The packer groups by
+        # run/LoRA before this point.
         first_sample = self.first_sample
-        return (
-            not _is_multimodal_sample(first_sample)
-            and not _is_multimodal_sample(sample)
-            and self.length + len(sample.input_ids) <= max_seq_len
-            and (first_sample.routed_experts is None) == (sample.routed_experts is None)
-        )
+        if self.length + len(sample.input_ids) > max_seq_len:
+            return False
+        if (first_sample.routed_experts is None) != (sample.routed_experts is None):
+            return False
+        return True
 
     def add(self, lora_idx: int, sample: MicroBatch) -> None:
         self.samples.append((lora_idx, sample))
@@ -290,6 +292,8 @@ def _materialize_bin(bin_content: _MicroBatchBin, num_loras: int) -> MicroBatch:
     env_names: list[str] = []
     ref_logprobs: list[float] | None = [] if has_ref_logprobs else None
     mm_token_type_ids: list[int] | None = [] if has_mm_token_type_ids else None
+    mm_kwargs: dict[str, EncodedTensor] | None = None
+    seq_lens: list[int] = []
     streams: dict[str, list[float] | None] = {name: ([] if has_stream[name] else None) for name in STREAM_FILL}
     routed_experts: RoutedExperts | None = None
     lora_num_tokens = [0] * num_loras
@@ -322,11 +326,19 @@ def _materialize_bin(bin_content: _MicroBatchBin, num_loras: int) -> MicroBatch:
                 assert routed_experts.shape[1:] == sample.routed_experts.shape[1:]
                 routed_experts.data += sample.routed_experts.data
                 routed_experts.shape[0] += sample.routed_experts.shape[0]
+        if sample.mm_kwargs is not None:
+            if mm_kwargs is None:
+                mm_kwargs = copy.deepcopy(sample.mm_kwargs)
+            else:
+                for key in mm_kwargs:
+                    mm_kwargs[key].data += sample.mm_kwargs[key].data
+                    mm_kwargs[key].shape[0] += sample.mm_kwargs[key].shape[0]
+        seq_lens.extend(sample.seq_lens)
         lora_num_tokens[lora_idx] += sample_len
 
     sequence_lengths = [len(sample.input_ids) for _, sample in bin_content.samples]
     assert sum(sequence_lengths) == len(input_ids), (sequence_lengths, len(input_ids))
-    first_sample = bin_content.first_sample
+    assert sum(seq_lens) == len(input_ids), (seq_lens, len(input_ids))
 
     return MicroBatch(
         input_ids=input_ids,
@@ -341,10 +353,11 @@ def _materialize_bin(bin_content: _MicroBatchBin, num_loras: int) -> MicroBatch:
         routed_experts=routed_experts,
         mm_token_type_ids=mm_token_type_ids,
         env_names=env_names,
-        mm_kwargs=first_sample.mm_kwargs if _is_multimodal_sample(first_sample) else None,
+        mm_kwargs=mm_kwargs,
         rl_weights=streams["rl_weights"],
         ce_weights=streams["ce_weights"],
         ref_kl_weights=streams["ref_kl_weights"],
+        seq_lens=seq_lens,
     )
 
 
@@ -374,11 +387,9 @@ def packed_samples_into_micro_bs(
 ) -> list[MicroBatch]:
     """
     Pack samples into micro_batch efficiently.
-    We follow the First Fit Decreasing algorithm to pack the samples into bins and minimize potential padding while never truncating.
-    With per-token temperatures, samples can be packed together regardless of their temperature values.
-
-    NOTE: Multimodal samples (with mm_kwargs) are NOT packed together as they have variable-sized
-    vision data that doesn't pack well. Each multimodal sample becomes its own micro batch.
+    We follow the First Fit Decreasing algorithm to pack samples into bins and
+    minimize potential padding while never truncating. Packed batches preserve
+    sample boundaries in ``seq_lens``.
     """
     # Sort by (lora_idx, -length) for packing efficiency
     samples.sort(key=lambda x: (x[0], -len(x[1].input_ids)))
@@ -386,7 +397,7 @@ def packed_samples_into_micro_bs(
     bins: list[_MicroBatchBin] = []
 
     for idx, sample in samples:
-        # Try to find a bin that can fit this sequence (only pack text-only samples)
+        # Try to find a bin that can fit this sequence.
         for bin_content in bins:
             if bin_content.can_add(sample, max_seq_len):
                 bin_content.add(idx, sample)
@@ -464,6 +475,7 @@ def pad_micro_batch(micro_batch: MicroBatch, pad_to_multiple_of: int) -> MicroBa
         )
     if micro_batch.mm_token_type_ids is not None:
         micro_batch.mm_token_type_ids.extend([0] * padding_size)
+    micro_batch.seq_lens.append(padding_size)
     if micro_batch.routed_experts is not None:
         _pad_routed_experts(micro_batch, padding_size)
     micro_batch.env_names.extend([""] * padding_size)
@@ -538,32 +550,24 @@ def prepare_batch(
     Prepare a batch of problems for each GPU. Each batch is a list of micro batches.
     Each micro batch is shape [1, seq_len], the number of samples is not fixed per micro batch.
 
-    FSDP requires all ranks to execute the same operations at each step. If one rank
-    processes a multimodal batch (triggering the vision encoder) while another processes
-    a text-only batch, the all-gather will hang. We separate micro batches by modality
-    and distribute them so that at each step index, all ranks see the same modality.
+    VLM models handle text-only samples with a synthetic dummy vision path, so
+    microbatches can be distributed through one generic packing path.
     """
     all_samples = [(idx, prepare_sample(rollout, seq_len)) for idx, rollout in zip(idxs, rollouts)]
 
-    micro_batches = packed_samples_into_micro_bs(all_samples, seq_len, num_loras, num_train_workers, bin_cost)
+    micro_batches = packed_samples_into_micro_bs(
+        all_samples,
+        seq_len,
+        num_loras,
+        num_train_workers,
+        bin_cost,
+    )
     micro_batches = [pad_micro_batch(micro_batch, pad_to_multiple_of) for micro_batch in micro_batches]
 
-    # Separate by modality so each step index has uniform modality across all ranks
-    mm_batches = [b for b in micro_batches if _is_multimodal_sample(b)]
-    text_batches = [b for b in micro_batches if not _is_multimodal_sample(b)]
+    micro_batches = _pad_group_for_distribution(micro_batches, num_train_workers)
 
-    # Pad each group independently so its count is divisible by num_train_workers
-    mm_batches = _pad_group_for_distribution(mm_batches, num_train_workers)
-    text_batches = _pad_group_for_distribution(text_batches, num_train_workers)
-
-    # Alignment check after distribution padding so the dummy batches are covered too
-    for micro_batch in (*mm_batches, *text_batches):
+    # Alignment check after distribution padding so the dummy batches are covered too.
+    for micro_batch in micro_batches:
         _assert_token_arrays_aligned(micro_batch)
 
-    batches_per_gpu: list[list[MicroBatch]] = [[] for _ in range(num_train_workers)]
-    for group in (mm_batches, text_batches):
-        group_batches_per_gpu = _distribute_group(group, num_train_workers, bin_cost)
-        for worker_idx, worker_batches in enumerate(group_batches_per_gpu):
-            batches_per_gpu[worker_idx].extend(worker_batches)
-
-    return batches_per_gpu
+    return _distribute_group(micro_batches, num_train_workers, bin_cost)
diff --git a/src/prime_rl/trainer/model.py b/src/prime_rl/trainer/model.py
index 2cdf8bbef8..8a08d4397f 100644
--- a/src/prime_rl/trainer/model.py
+++ b/src/prime_rl/trainer/model.py
@@ -56,7 +56,11 @@
 from prime_rl.utils.logger import get_logger
 from prime_rl.utils.sequence import get_cu_seqlens_from_position_ids
 from prime_rl.utils.utils import format_time
-from prime_rl.utils.vlm import get_language_model, get_vision_encoder, is_vlm_architecture
+from prime_rl.utils.vlm import (
+    get_language_model,
+    get_vision_encoder,
+    is_vlm_architecture,
+)
 
 
 def pre_download_model(model_name: str) -> None:
@@ -476,6 +480,12 @@ def get_model(
             raise ValueError(
                 "VLM models must use optimization_dtype='bfloat16' and reduce_dtype='bfloat16' to match vLLM inference."
             )
+        # Context parallelism patches the global flash-attention dispatch to do the
+        # Ulysses all-to-all, which is causal- and sharded-sequence-only.
+        vision_config = getattr(model_config, "vision_config", None)
+        if config.cp > 1 and vision_config is not None:
+            vision_config._attn_implementation = "sdpa"
+            logger.info("CP enabled for VLM: pinning vision encoder attention to 'sdpa' (excluded from Ulysses CP).")
 
     # GPT-OSS only supports FlashAttention via kernels-community/vllm-flash-attn3, which requires Hopper (SM 90).
     # On other architectures (e.g. Blackwell), users must fall back to eager attention.
@@ -583,6 +593,12 @@ def get_model(
     else:
         impl_to_use = config.impl
 
+    if config.vlm is not None and config.cp > 1 and not (is_vlm_arch and impl_to_use == "custom" and custom_vlm_cls):
+        raise ValueError(
+            "VLM context parallelism requires a custom PrimeRL VLM implementation; "
+            f"{getattr(model_config, 'model_type', config.name)!r} is only supported with cp=1."
+        )
+
     with device:
         if impl_to_use == "custom" and custom_vlm_cls is not None:
             model_cls = custom_vlm_cls
@@ -663,13 +679,31 @@ def setup_fsdp(model: nn.Module, config: ModelConfig, parallel_dims: ParallelDim
         dp_mod_ep_mesh = parallel_dims.world_mesh[tuple(dp_mod_ep_mesh_dim_names)]
 
     is_vlm_training = config.vlm is not None
+    cp_vlm = is_vlm_training and config.cp > 1
+    # Params kept out of FSDP entirely (full/replicated on every rank).
+    fsdp_ignored_params: set[nn.Parameter] = set()
+
+    def _all_params_frozen(module: nn.Module) -> bool:
+        params = list(module.parameters())
+        return bool(params) and not any(p.requires_grad for p in params)
+
+    def _ignore_params(module: nn.Module, name: str) -> None:
+        params = list(module.parameters())
+        if not params:
+            return
+        fsdp_ignored_params.update(params)
+        get_logger().info(f"CP+VLM: keeping frozen {name} out of FSDP (ignored_params).")
+
     if is_vlm_training:
         vision_encoder = get_vision_encoder(model, override=config.vlm.vision_encoder_attr)
         if vision_encoder is None:
             raise ValueError(f"VLM model {config.name} has no recognized vision encoder")
 
-        fully_shard(vision_encoder, mesh=hsdp_mesh, **fsdp_config)
-        get_logger().info(f"Applied FSDP to vision encoder (frozen={config.vlm.freeze_vision_encoder})")
+        if cp_vlm and _all_params_frozen(vision_encoder):
+            _ignore_params(vision_encoder, "vision encoder")
+        else:
+            fully_shard(vision_encoder, mesh=hsdp_mesh, **fsdp_config)
+            get_logger().info(f"Applied FSDP to vision encoder (frozen={config.vlm.freeze_vision_encoder})")
 
     language_model = get_language_model(model, override=config.vlm.language_model_attr if is_vlm_training else None)
     transformer_layers = language_model.layers
@@ -692,11 +726,12 @@ def setup_fsdp(model: nn.Module, config: ModelConfig, parallel_dims: ParallelDim
     if shard_norm_and_lm_head:
         # This optimization breaks weight tying
         embed_module = getattr(language_model, "embed_tokens", None) or getattr(language_model, "embeddings", None)
-        fully_shard(
-            embed_module,
-            mesh=hsdp_mesh,
-            **fsdp_config,
-        )
+        if embed_module is not None:
+            fully_shard(
+                embed_module,
+                mesh=hsdp_mesh,
+                **fsdp_config,
+            )
         norm_module = getattr(language_model, "norm", None) or language_model.norm_f
         fully_shard(
             [model.lm_head, norm_module],
@@ -714,6 +749,7 @@ def setup_fsdp(model: nn.Module, config: ModelConfig, parallel_dims: ParallelDim
         mp_policy=mp_policy,
         offload_policy=offload_policy,
         reshard_after_forward=config.reshard_after_forward,
+        ignored_params=fsdp_ignored_params or None,
     )
 
     if not parallel_dims.ep_enabled:
@@ -1166,26 +1202,35 @@ def setup_model(
 
 def forward(
     model: nn.Module,
-    input_ids: Int[Tensor, "batch seq"],
-    position_ids: Int[Tensor, "batch seq"],
+    input_ids: Int[Tensor, "batch seq"] | None,
+    position_ids: Int[Tensor, "batch seq"] | None,
     labels: Int[Tensor, "batch seq"] | None = None,
     temperature: Tensor | None = None,
     routed_experts: Int[Tensor, "batch seq layers topk"] | None = None,
+    inputs_embeds: Tensor | None = None,
     # Generic multimodal kwargs (e.g. {"pixel_values": ...,
     # "image_grid_thw": ...} for Qwen3-VL; just {"pixel_values": ...}
     # for Gemma3). Passed straight through to ``model(**kwargs)`` so
     # the model's HF forward signature is the schema. ``mm_token_type_ids``
-    # is split out because it's prime-rl-computed (from token ids),
-    # not a renderer/processor output.
+    # is split out because it's prime-rl-computed (from token ids).
     mm_kwargs: dict[str, Tensor] | None = None,
     mm_token_type_ids: Int[Tensor, "batch seq"] | None = None,
+    seq_lens: Int[Tensor, "segments"] | None = None,
+    seq_lens_are_global: bool = False,
+    cp_group: object | None = None,
+    cp_rank: int | None = None,
+    cp_world_size: int | None = None,
+    cp_style: str | None = None,
 ) -> PrimeLmOutput:
     # Build kwargs for model forward
     kwargs = {
-        "input_ids": input_ids,
         "labels": labels,
         "temperature": temperature,
     }
+    if inputs_embeds is None:
+        kwargs["input_ids"] = input_ids
+    else:
+        kwargs["inputs_embeds"] = inputs_embeds
 
     if mm_kwargs:
         # Forward the per-model multimodal tensors verbatim, plus the
@@ -1194,18 +1239,27 @@ def forward(
         kwargs.update(mm_kwargs)
         if mm_token_type_ids is not None:
             kwargs["mm_token_type_ids"] = mm_token_type_ids
-        # ``position_ids`` for MRoPE families: Qwen3-VL's HF forward
-        # recomputes 3D positions from ``image_grid_thw`` and breaks if
-        # given the trainer's pre-computed 1D ``position_ids``. Detect
-        # via the mm_kwargs shape so we don't enumerate model_types.
         if "image_grid_thw" not in mm_kwargs:
             kwargs["position_ids"] = position_ids
     else:
         kwargs["position_ids"] = position_ids
 
+    if isinstance(model, PreTrainedModelPrimeRL):
+        kwargs.update(model.prime_forward_kwargs(seq_lens=seq_lens, seq_lens_are_global=seq_lens_are_global))
+
     if routed_experts is not None:
         kwargs["routed_experts"] = routed_experts
 
+    if cp_group is not None:
+        kwargs.update(
+            {
+                "cp_group": cp_group,
+                "cp_rank": cp_rank,
+                "cp_world_size": cp_world_size,
+                "cp_style": cp_style,
+            }
+        )
+
     out = model(**kwargs)
 
     # PrimeLmOutput is a TypedDict (dict at runtime), HF outputs are dataclass-like objects
diff --git a/src/prime_rl/trainer/models/base.py b/src/prime_rl/trainer/models/base.py
index df66984847..db859f197e 100644
--- a/src/prime_rl/trainer/models/base.py
+++ b/src/prime_rl/trainer/models/base.py
@@ -21,6 +21,15 @@ def _can_set_experts_implementation(cls) -> bool:
         """PrimeRL models use custom MoE implementations and don't support dynamic experts implementation."""
         return False
 
+    def prime_forward_kwargs(
+        self,
+        *,
+        seq_lens: Tensor | None = None,
+        seq_lens_are_global: bool = False,
+    ) -> dict[str, object]:
+        """Return PrimeRL-only kwargs needed by this model's forward path."""
+        return {}
+
     def get_correct_experts_implementation(self, requested_experts: str | None) -> str:
         """PrimeRL models always use eager experts implementation."""
         return "eager"
diff --git a/src/prime_rl/trainer/models/qwen3_5_moe/modeling_qwen3_5_moe.py b/src/prime_rl/trainer/models/qwen3_5_moe/modeling_qwen3_5_moe.py
index f0d3b2a7cf..30047e74e1 100644
--- a/src/prime_rl/trainer/models/qwen3_5_moe/modeling_qwen3_5_moe.py
+++ b/src/prime_rl/trainer/models/qwen3_5_moe/modeling_qwen3_5_moe.py
@@ -19,6 +19,7 @@
 from prime_rl.trainer.models.layers.lm_head import PrimeLmOutput
 from prime_rl.trainer.models.layers.moe import FeedForward, MoE, MoEArgs
 from prime_rl.trainer.models.layers.rotary_emb import apply_rotary_pos_emb
+from prime_rl.utils.cp import setup_cp_attention_params, shard_for_cp, shard_position_ids_for_cp
 from prime_rl.utils.sequence import get_cu_seqlens_from_position_ids, get_cu_seqlens_from_seq_lens
 
 from .configuration_qwen3_5_moe import Qwen3_5MoeConfig
@@ -52,6 +53,11 @@
 except ImportError:
     causal_conv1d_fn = None  # type: ignore
 
+try:
+    from fla.modules.convolution import causal_conv1d as fla_causal_conv1d
+except ImportError:
+    fla_causal_conv1d = None  # type: ignore
+
 try:
     from fla.modules import FusedRMSNormGated
     from fla.ops.cp import FLACPContext, build_cp_context
@@ -244,14 +250,23 @@ def __init__(self, config: Qwen3_5MoeConfig):
         self._causal_conv1d_fn = causal_conv1d_fn
         self._chunk_gated_delta_rule = chunk_gated_delta_rule or torch_chunk_gated_delta_rule
 
-    def _build_cp_context(self, local_seq_len: int, device: torch.device) -> "FLACPContext | None":
+    def _build_cp_context(
+        self,
+        local_seq_len: int,
+        device: torch.device,
+        cu_seqlens: torch.LongTensor | None = None,
+    ) -> "FLACPContext | None":
         """Build fla CP context from the local (sharded) sequence length."""
         cp_group = getattr(self, "cp_group", None)
         if cp_group is None or build_cp_context is None:
             return None
-        # Reconstruct global cu_seqlens: single contiguous sequence across all CP ranks
         global_seq_len = local_seq_len * self.cp_world_size
-        global_cu_seqlens = torch.tensor([0, global_seq_len], dtype=torch.int32, device=device)
+        # VLM CP passes global seq_lens; text CP may pass local cu_seqlens
+        # recomputed from already-sharded position_ids.
+        if cu_seqlens is not None and int(cu_seqlens[-1].item()) == global_seq_len:
+            global_cu_seqlens = cu_seqlens.to(device=device, dtype=torch.int32)
+        else:
+            global_cu_seqlens = torch.tensor([0, global_seq_len], dtype=torch.int32, device=device)
         return build_cp_context(
             cu_seqlens=global_cu_seqlens,
             group=cp_group,
@@ -264,6 +279,9 @@ def forward(
         cu_seqlens: torch.LongTensor | None = None,
     ) -> torch.Tensor:
         batch_size, seq_len, _ = hidden_states.shape
+        cp_context = self._build_cp_context(seq_len, hidden_states.device, cu_seqlens=cu_seqlens)
+        if cp_context is not None:
+            cu_seqlens = cp_context.cu_seqlens
 
         mixed_qkv = self.in_proj_qkv(hidden_states).transpose(1, 2)
         z = self.in_proj_z(hidden_states).reshape(batch_size, seq_len, -1, self.head_v_dim)
@@ -272,7 +290,16 @@ def forward(
 
         # Causal conv1d — must reset at sequence boundaries for packed batches,
         # otherwise the kernel-1 left pad leaks state across sequences.
-        if self._causal_conv1d_fn is not None:
+        if cp_context is not None:
+            mixed_qkv, _ = fla_causal_conv1d(
+                x=mixed_qkv.transpose(1, 2),
+                weight=self.conv1d.weight.squeeze(1),
+                bias=self.conv1d.bias,
+                activation=self.activation,
+                cp_context=cp_context,
+            )
+            mixed_qkv = mixed_qkv.transpose(1, 2)
+        elif self._causal_conv1d_fn is not None:
             seq_idx = None
             if cu_seqlens is not None:
                 seg_lens = cu_seqlens[1:] - cu_seqlens[:-1]
@@ -313,10 +340,7 @@ def forward(
             query = query.repeat_interleave(self.num_v_heads // self.num_k_heads, dim=2)
             key = key.repeat_interleave(self.num_v_heads // self.num_k_heads, dim=2)
 
-        # Use fla's native CP when available, otherwise fall back to PyTorch kernel
-        cp_context = self._build_cp_context(seq_len, hidden_states.device)
         if cp_context is not None:
-            cu_seqlens = cp_context.cu_seqlens
             core_attn_out, _ = self._chunk_gated_delta_rule(
                 query,
                 key,
@@ -851,6 +875,7 @@ def forward(
         inputs_embeds: Optional[torch.FloatTensor] = None,
         routed_experts: Optional[torch.LongTensor] = None,
         seq_lens: Optional[torch.LongTensor] = None,
+        seq_lens_are_global: bool = False,
     ) -> MoeModelOutputWithPast:
         if (input_ids is None) ^ (inputs_embeds is not None):
             raise ValueError("You must specify exactly one of input_ids or inputs_embeds")
@@ -863,7 +888,9 @@ def forward(
 
         flash_attn_enabled = self.config._attn_implementation in ("flash_attention_2", "flash_attention_3", "fa4")
         if flash_attn_enabled:
-            if position_ids.ndim == 3:
+            if seq_lens_are_global and seq_lens is not None:
+                cu_seqlens, max_seqlen = get_cu_seqlens_from_seq_lens(seq_lens.to(device=inputs_embeds.device))
+            elif position_ids.ndim == 3:
                 if inputs_embeds.shape[0] != 1:
                     raise ValueError("3D Qwen3.5 MRoPE positions require batch size 1 for varlen attention")
                 seq_len = inputs_embeds.shape[1]
@@ -882,7 +909,10 @@ def forward(
             seq_lens = seq_lens.to(device=inputs_embeds.device)
             if seq_lens.numel() > 1 and "full_attention" in self.config.layer_types:
                 raise ValueError("Packed Qwen3.5 MRoPE batches with full_attention layers require flash attention")
-            cu_seqlens, max_seqlen = get_cu_seqlens_from_seq_lens(seq_lens, total_tokens=inputs_embeds.shape[1])
+            cu_seqlens, max_seqlen = get_cu_seqlens_from_seq_lens(
+                seq_lens,
+                total_tokens=None if seq_lens_are_global else inputs_embeds.shape[1],
+            )
         else:
             max_seqlen = None
             cu_seqlens = None
@@ -948,7 +978,7 @@ def _dummy_vision_inputs(self, device: torch.device) -> tuple[torch.Tensor, torc
         grid_thw = torch.tensor([[1, m, m]], dtype=torch.long, device=device)
         return pixel_values, grid_thw
 
-    def forward(
+    def prepare_inputs_embeds_and_position_ids(
         self,
         input_ids: torch.LongTensor | None = None,
         position_ids: torch.LongTensor | None = None,
@@ -956,11 +986,11 @@ def forward(
         pixel_values: torch.Tensor | None = None,
         image_grid_thw: torch.LongTensor | None = None,
         mm_token_type_ids: torch.LongTensor | None = None,
-        routed_experts: torch.LongTensor | None = None,
         seq_lens: torch.LongTensor | None = None,
-        **kwargs,
-    ) -> MoeModelOutputWithPast:
+    ) -> tuple[torch.FloatTensor, torch.LongTensor]:
         if inputs_embeds is None:
+            if input_ids is None:
+                raise ValueError("input_ids are required when inputs_embeds are not provided")
             inputs_embeds = self.language_model.embed_tokens(input_ids)
 
         if image_grid_thw is not None and input_ids is None:
@@ -1005,6 +1035,8 @@ def forward(
                 )
             image_mask = image_mask.unsqueeze(-1).expand_as(inputs_embeds).to(inputs_embeds.device)
             inputs_embeds = inputs_embeds.masked_scatter(image_mask, image_embeds)
+        else:
+            inputs_embeds = inputs_embeds + image_embeds.sum() * 0.0
 
         if position_ids is None:
             if image_grid_thw is not None:
@@ -1018,11 +1050,53 @@ def forward(
             else:
                 position_ids = torch.arange(inputs_embeds.shape[1], device=inputs_embeds.device).unsqueeze(0)
 
+        return inputs_embeds, position_ids
+
+    def forward(
+        self,
+        input_ids: torch.LongTensor | None = None,
+        position_ids: torch.LongTensor | None = None,
+        inputs_embeds: torch.FloatTensor | None = None,
+        pixel_values: torch.Tensor | None = None,
+        image_grid_thw: torch.LongTensor | None = None,
+        mm_token_type_ids: torch.LongTensor | None = None,
+        routed_experts: torch.LongTensor | None = None,
+        seq_lens: torch.LongTensor | None = None,
+        seq_lens_are_global: bool = False,
+        cp_group: object | None = None,
+        cp_rank: int | None = None,
+        cp_world_size: int | None = None,
+        cp_style: str | None = None,
+        **kwargs,
+    ) -> MoeModelOutputWithPast:
+        cp_enabled = cp_group is not None or cp_rank is not None or cp_world_size is not None or cp_style is not None
+
+        inputs_embeds, position_ids = self.prepare_inputs_embeds_and_position_ids(
+            input_ids=input_ids,
+            position_ids=position_ids,
+            inputs_embeds=inputs_embeds,
+            pixel_values=pixel_values,
+            image_grid_thw=image_grid_thw,
+            mm_token_type_ids=mm_token_type_ids,
+            seq_lens=seq_lens,
+        )
+
+        if cp_enabled:
+            # VLM CP prep stays in root forward so FSDP enters the root before
+            # embed/vision children; merge real or dummy vision before sharding.
+            setup_cp_attention_params(position_ids, cp_group=cp_group, cp_style=cp_style, seq_lens=seq_lens)
+            inputs_embeds = shard_for_cp(inputs_embeds, cp_rank=cp_rank, cp_world_size=cp_world_size)
+            position_ids = shard_position_ids_for_cp(position_ids, cp_rank=cp_rank, cp_world_size=cp_world_size)
+            if routed_experts is not None:
+                routed_experts = shard_for_cp(routed_experts, cp_rank=cp_rank, cp_world_size=cp_world_size)
+            seq_lens_are_global = True
+
         return self.language_model(
             inputs_embeds=inputs_embeds,
             position_ids=position_ids,
             routed_experts=routed_experts,
             seq_lens=seq_lens,
+            seq_lens_are_global=seq_lens_are_global,
         )
 
 
@@ -1155,6 +1229,16 @@ def convert_layer_to_prime(cls, state_dict: dict[str, Tensor], layer_idx: int) -
     # Forward
     # ------------------------------------------------------------------
 
+    def prime_forward_kwargs(
+        self,
+        *,
+        seq_lens: Tensor | None = None,
+        seq_lens_are_global: bool = False,
+    ) -> dict[str, object]:
+        if seq_lens is None:
+            return {}
+        return {"seq_lens": seq_lens, "seq_lens_are_global": seq_lens_are_global}
+
     def forward(
         self,
         input_ids: Optional[torch.LongTensor] = None,
@@ -1171,6 +1255,11 @@ def forward(
         image_grid_thw: Optional[torch.LongTensor] = None,
         mm_token_type_ids: Optional[torch.LongTensor] = None,
         seq_lens: Optional[torch.LongTensor] = None,
+        seq_lens_are_global: bool = False,
+        cp_group: object | None = None,
+        cp_rank: int | None = None,
+        cp_world_size: int | None = None,
+        cp_style: str | None = None,
         **kwargs: Unpack[TransformersKwargs],
     ) -> PrimeLmOutput:
         assert use_cache is None, "use_cache is not supported for custom qwen3_5_moe for now"
@@ -1193,6 +1282,11 @@ def forward(
                 mm_token_type_ids=mm_token_type_ids,
                 routed_experts=routed_experts,
                 seq_lens=seq_lens,
+                seq_lens_are_global=seq_lens_are_global,
+                cp_group=cp_group,
+                cp_rank=cp_rank,
+                cp_world_size=cp_world_size,
+                cp_style=cp_style,
             )
         else:
             outputs = self.model(
@@ -1201,6 +1295,7 @@ def forward(
                 inputs_embeds=inputs_embeds,
                 routed_experts=routed_experts,
                 seq_lens=seq_lens,
+                seq_lens_are_global=seq_lens_are_global,
             )
 
         hidden_states = outputs.last_hidden_state
diff --git a/src/prime_rl/trainer/rl/data.py b/src/prime_rl/trainer/rl/data.py
index 27e031f496..c192854dd1 100644
--- a/src/prime_rl/trainer/rl/data.py
+++ b/src/prime_rl/trainer/rl/data.py
@@ -35,6 +35,7 @@ class TensorMicroBatch(TypedDict):
 
     # Batch level
     lora_num_tokens: Int[Tensor, "n_loras"]
+    seq_lens: Int[Tensor, "segments"] | None
 
     # MoE router replay
     routed_experts: Int[Tensor, "batch seq layers topk"] | None
@@ -130,6 +131,7 @@ def _get_sample_micro_batch(self, generator: torch.Generator) -> TensorMicroBatc
             "sequence_lengths": sequence_lengths,
             "loss_mask": loss_mask.unsqueeze(0),
             "lora_num_tokens": lora_num_tokens,
+            "seq_lens": torch.tensor(sequence_lengths, dtype=torch.long),
             "routed_experts": None,
             "mm_kwargs": None,
             "mm_token_type_ids": None,
@@ -162,6 +164,7 @@ def _get_micro_batch(self, generator: torch.Generator) -> TensorMicroBatch:
             "sequence_lengths": [self.seq_len],
             "loss_mask": torch.ones(self.seq_len, dtype=torch.bool).unsqueeze(0),
             "lora_num_tokens": lora_num_tokens,
+            "seq_lens": torch.tensor([self.seq_len], dtype=torch.long),
             "routed_experts": None,
             "mm_kwargs": None,
             "mm_token_type_ids": None,
@@ -259,6 +262,7 @@ def _micro_batch_to_tensor(self, micro_batch: MicroBatch) -> TensorMicroBatch:
             env_names=micro_batch.env_names,
             sequence_lengths=micro_batch.sequence_lengths,
             lora_num_tokens=torch.tensor(micro_batch.lora_num_tokens, dtype=torch.int32),
+            seq_lens=torch.tensor(micro_batch.seq_lens, dtype=torch.long) if micro_batch.seq_lens is not None else None,
             mm_kwargs=mm_kwargs,
             mm_token_type_ids=torch.tensor(micro_batch.mm_token_type_ids, dtype=torch.long).unsqueeze(0)
             if micro_batch.mm_token_type_ids is not None
diff --git a/src/prime_rl/trainer/rl/packer.py b/src/prime_rl/trainer/rl/packer.py
index fec0658767..bcf5dda3f3 100644
--- a/src/prime_rl/trainer/rl/packer.py
+++ b/src/prime_rl/trainer/rl/packer.py
@@ -90,7 +90,15 @@ def __init__(
         bin_cost: Callable[[Sequence[int]], int],
         start_step: int = 0,
     ):
-        super().__init__(dp_world_size, seq_len, pad_to_multiple_of, tokenizer, config, bin_cost, start_step)
+        super().__init__(
+            dp_world_size,
+            seq_len,
+            pad_to_multiple_of,
+            tokenizer,
+            config,
+            bin_cost,
+            start_step,
+        )
         assert self.multi_run_manager.max_runs == 1, "SinglePacker only supports one run"
 
     def pack(self):
@@ -137,7 +145,15 @@ def __init__(
         bin_cost: Callable[[Sequence[int]], int],
         start_step: int = 0,
     ):
-        super().__init__(dp_world_size, seq_len, pad_to_multiple_of, tokenizer, config, bin_cost, start_step)
+        super().__init__(
+            dp_world_size,
+            seq_len,
+            pad_to_multiple_of,
+            tokenizer,
+            config,
+            bin_cost,
+            start_step,
+        )
         # Per-run buffer: stores (TrainingSample, step) tuples
         self.buffers: list[deque[tuple[TrainingSample, int]]] = [
             deque() for _ in range(self.multi_run_manager.max_runs)
@@ -359,9 +375,21 @@ def setup_packer(
     multi_run_manager = get_multi_run_manager()
     if multi_run_manager.max_runs == 1:
         return SinglePacker(
-            dp_world_size, seq_len, pad_to_multiple_of, tokenizer, transport_config, bin_cost, start_step
+            dp_world_size,
+            seq_len,
+            pad_to_multiple_of,
+            tokenizer,
+            transport_config,
+            bin_cost,
+            start_step,
         )
     else:
         return MultiPacker(
-            dp_world_size, seq_len, pad_to_multiple_of, tokenizer, transport_config, bin_cost, start_step
+            dp_world_size,
+            seq_len,
+            pad_to_multiple_of,
+            tokenizer,
+            transport_config,
+            bin_cost,
+            start_step,
         )
diff --git a/src/prime_rl/trainer/rl/train.py b/src/prime_rl/trainer/rl/train.py
index 6e79433065..44363b616f 100644
--- a/src/prime_rl/trainer/rl/train.py
+++ b/src/prime_rl/trainer/rl/train.py
@@ -209,7 +209,7 @@ def load_run_checkpoint(_optimizer, idx: int) -> None:
             setup_sparse_mla_cp,
         )
 
-        assert_cp_style_supports_model(config.model.cp_style, model)
+        assert_cp_style_supports_model(config.model.cp_style, model, cp_world_size=parallel_dims.cp)
         # sparse MLA is softmax (works with both ring and ulysses).
         setup_sparse_mla_cp(model, cp_group, cp_rank, parallel_dims.cp)
         # Linear-attn / Mamba layers are only configured under ulysses; with ring
@@ -421,19 +421,32 @@ def load_run_checkpoint(_optimizer, idx: int) -> None:
                 if micro_batch.get("mm_token_type_ids") is not None
                 else None
             )
+            seq_lens = micro_batch["seq_lens"].to("cuda") if micro_batch.get("seq_lens") is not None else None
 
             labels = shift_tensor_left(input_ids)
 
-            # VLM + CP is not supported: MRoPE requires global positions but CP shards the sequence
-            if cp_enabled and mm_kwargs is not None:
-                raise NotImplementedError("Context parallelism is not supported with VLM/multimodal training")
-
+            forward_mm_kwargs = mm_kwargs
+            forward_seq_lens = seq_lens
+            forward_seq_lens_are_global = False
+            forward_cp_group = None
+            forward_cp_rank = None
+            forward_cp_world_size = None
+            forward_cp_style = None
+            vlm_cp_enabled = cp_enabled and config.model.vlm is not None
             if cp_enabled:
-                input_ids, forward_position_ids = setup_cp_params(
-                    input_ids, position_ids, cp_rank, cp_size, cp_group, cp_style=config.model.cp_style
-                )
+                if vlm_cp_enabled:
+                    forward_position_ids = position_ids
+                    forward_cp_group = cp_group
+                    forward_cp_rank = cp_rank
+                    forward_cp_world_size = cp_size
+                    forward_cp_style = config.model.cp_style
+                else:
+                    input_ids, forward_position_ids = setup_cp_params(
+                        input_ids, position_ids, cp_rank, cp_size, cp_group, cp_style=config.model.cp_style
+                    )
+
                 labels = shard_for_cp(labels, cp_rank=cp_rank, cp_world_size=cp_size)
-                if routed_experts is not None:
+                if routed_experts is not None and not vlm_cp_enabled:
                     routed_experts = shard_for_cp(routed_experts, cp_rank=cp_rank, cp_world_size=cp_size)
             else:
                 forward_position_ids = position_ids
@@ -441,7 +454,10 @@ def load_run_checkpoint(_optimizer, idx: int) -> None:
             if config.model.lora:
                 lora_num_tokens = micro_batch["lora_num_tokens"].to("cuda")
                 if cp_enabled:
-                    chunk_size = input_ids.shape[1]
+                    if vlm_cp_enabled:
+                        chunk_size = input_ids.shape[1] // cp_size
+                    else:
+                        chunk_size = input_ids.shape[1]
                     # Convert to cumsum, adjust for CP chunk, convert back to num_tokens
                     cu_offsets = lora_num_tokens.cumsum(dim=0, dtype=torch.int32)
                     adjusted_cu = torch.clip(cu_offsets - chunk_size * cp_rank, min=0, max=chunk_size)
@@ -464,9 +480,15 @@ def load_run_checkpoint(_optimizer, idx: int) -> None:
                     forward_position_ids,
                     labels=labels,
                     temperature=temperatures,
-                    mm_kwargs=mm_kwargs,
+                    mm_kwargs=forward_mm_kwargs,
                     mm_token_type_ids=mm_token_type_ids,
+                    seq_lens=forward_seq_lens,
+                    seq_lens_are_global=forward_seq_lens_are_global,
                     routed_experts=routed_experts,
+                    cp_group=forward_cp_group,
+                    cp_rank=forward_cp_rank,
+                    cp_world_size=forward_cp_world_size,
+                    cp_style=forward_cp_style,
                 )
 
             if out.get("logprobs") is None:
diff --git a/src/prime_rl/trainer/sft/train.py b/src/prime_rl/trainer/sft/train.py
index 3a82ebe864..614ad0910f 100644
--- a/src/prime_rl/trainer/sft/train.py
+++ b/src/prime_rl/trainer/sft/train.py
@@ -145,7 +145,7 @@ def train(config: SFTConfig):
     if parallel_dims.cp_enabled:
         from prime_rl.utils.cp import assert_cp_style_supports_model
 
-        assert_cp_style_supports_model(config.model.cp_style, model)
+        assert_cp_style_supports_model(config.model.cp_style, model, cp_world_size=parallel_dims.cp)
         # sparse MLA is softmax (works with both ring and ulysses).
         setup_sparse_mla_cp(model, cp_group, cp_rank, parallel_dims.cp)
         # Linear-attn / Mamba layers are only configured under ulysses; with ring
diff --git a/src/prime_rl/transport/types.py b/src/prime_rl/transport/types.py
index 45727584fe..7e75826b6a 100644
--- a/src/prime_rl/transport/types.py
+++ b/src/prime_rl/transport/types.py
@@ -108,3 +108,6 @@ class MicroBatch(msgspec.Struct, array_like=True, gc=False, omit_defaults=True):
     # Packer-derived metadata used for run-local token exports.
     run_id: str | None = None
     run_step: int | None = None
+
+    # Packed sample boundaries. Sum equals len(input_ids) when present.
+    seq_lens: list[int] | None = None
diff --git a/src/prime_rl/utils/cp.py b/src/prime_rl/utils/cp.py
index 2ac9d45ad7..0907273c31 100644
--- a/src/prime_rl/utils/cp.py
+++ b/src/prime_rl/utils/cp.py
@@ -11,7 +11,7 @@
 import torch.nn as nn
 from ring_flash_attn import update_ring_flash_attn_params
 
-from prime_rl.utils.sequence import get_cu_seqlens_from_position_ids
+from prime_rl.utils.sequence import get_cu_seqlens_from_position_ids, get_cu_seqlens_from_seq_lens
 
 CPStyle = Literal["ring", "ulysses"]
 
@@ -34,7 +34,7 @@ def _has_linear_attn_layer(model: nn.Module) -> bool:
     return False
 
 
-def assert_cp_style_supports_model(cp_style: CPStyle, model: nn.Module) -> None:
+def assert_cp_style_supports_model(cp_style: CPStyle, model: nn.Module, cp_world_size: int | None = None) -> None:
     """Refuse `cp_style='ring'` on models that have linear/SSM attention layers.
 
     Ring CP is a softmax-attention algorithm (sequence ring all-gather of K/V).
@@ -50,6 +50,21 @@ def assert_cp_style_supports_model(cp_style: CPStyle, model: nn.Module) -> None:
             "cp_style='ulysses' instead — its all-to-all on Q/K/V works "
             "out-of-the-box with non-softmax kernels."
         )
+    if cp_style == "ulysses" and cp_world_size is not None:
+        config = getattr(model, "config", None)
+        config = getattr(config, "text_config", config)
+        num_attention_heads = getattr(config, "num_attention_heads", None)
+        num_key_value_heads = getattr(config, "num_key_value_heads", None)
+        if num_attention_heads is not None and num_attention_heads % cp_world_size != 0:
+            raise ValueError(
+                f"cp_style='ulysses' requires num_attention_heads ({num_attention_heads}) "
+                f"to be divisible by cp size ({cp_world_size})"
+            )
+        if num_key_value_heads is not None and num_key_value_heads % cp_world_size != 0:
+            raise ValueError(
+                f"cp_style='ulysses' requires num_key_value_heads ({num_key_value_heads}) "
+                f"to be divisible by cp size ({cp_world_size})"
+            )
 
 
 def setup_hybrid_cp(model: nn.Module, cp_group: dist.ProcessGroup, cp_rank: int, cp_world_size: int) -> None:
@@ -125,7 +140,7 @@ def setup_sparse_mla_cp(model: nn.Module, cp_group: dist.ProcessGroup, cp_rank:
         get_logger().info(f"Configured sparse MLA CP on {count} DSA layers")
 
 
-def shard_for_cp(t: torch.Tensor, cp_rank: int, cp_world_size: int) -> torch.Tensor:
+def shard_for_cp(t: torch.Tensor, cp_rank: int, cp_world_size: int, seq_dim: int = 1) -> torch.Tensor:
     """
     Shard a tensor for context parallelism.
     Args:
@@ -136,13 +151,25 @@ def shard_for_cp(t: torch.Tensor, cp_rank: int, cp_world_size: int) -> torch.Ten
         The shard of the tensor for the current rank.
     """
 
-    assert t.shape[0] == 1, "For CP, tensor must have batch dimension of 1"
+    if t.shape[seq_dim] % cp_world_size != 0:
+        raise ValueError(
+            f"CP requires sequence dimension {seq_dim} to be divisible by cp size: "
+            f"shape={tuple(t.shape)}, cp_size={cp_world_size}"
+        )
 
-    chunked_t = torch.chunk(t, cp_world_size, dim=1)
+    chunked_t = torch.chunk(t, cp_world_size, dim=seq_dim)
 
     return chunked_t[cp_rank]
 
 
+def shard_position_ids_for_cp(position_ids: torch.Tensor, cp_rank: int, cp_world_size: int) -> torch.Tensor:
+    # Qwen MRoPE positions are [3, batch, seq]; generic 3D tensors like
+    # inputs_embeds are [batch, seq, hidden] and should still shard on dim 1.
+    if position_ids.ndim == 3:
+        return shard_for_cp(position_ids, cp_rank=cp_rank, cp_world_size=cp_world_size, seq_dim=2)
+    return shard_for_cp(position_ids, cp_rank=cp_rank, cp_world_size=cp_world_size)
+
+
 def gather_for_cp(t: torch.Tensor, cp_group: dist.ProcessGroup) -> torch.Tensor:
     gathered_t = dist_nn.all_gather(t, group=cp_group)
 
@@ -202,7 +229,33 @@ def setup_cp_params(
     Returns the sequence-sharded input_ids and position_ids — the rest of the
     model still runs sequence-sharded; only attention sees the full sequence.
     """
-    cu_seqlens, max_seqlen = get_cu_seqlens_from_position_ids(position_ids)
+    setup_cp_attention_params(position_ids, cp_group=cp_group, cp_style=cp_style)
+
+    input_ids = shard_for_cp(input_ids, cp_rank=cp_rank, cp_world_size=cp_world_size)
+    position_ids = shard_position_ids_for_cp(position_ids, cp_rank=cp_rank, cp_world_size=cp_world_size)
+    return input_ids, position_ids
+
+
+def setup_cp_attention_params(
+    position_ids: torch.Tensor,
+    cp_group: dist.ProcessGroup,
+    cp_style: CPStyle = "ring",
+    seq_lens: torch.Tensor | None = None,
+) -> None:
+    if position_ids.ndim == 3:
+        # Qwen MRoPE positions are [3, batch, seq], so packed sample
+        # boundaries come from seq_lens instead of 2D position-id resets.
+        total_tokens = position_ids.shape[2]
+        if seq_lens is None:
+            cu_seqlens = torch.tensor([0, total_tokens], dtype=torch.int32, device=position_ids.device)
+            max_seqlen = total_tokens
+        else:
+            cu_seqlens, max_seqlen = get_cu_seqlens_from_seq_lens(
+                seq_lens.to(device=position_ids.device),
+                total_tokens=total_tokens,
+            )
+    else:
+        cu_seqlens, max_seqlen = get_cu_seqlens_from_position_ids(position_ids)
 
     if cp_style == "ring":
         update_ring_flash_attn_params(cu_seqlens, cp_group)
@@ -214,7 +267,3 @@ def setup_cp_params(
         update_ulysses_params(cu_seqlens, max_seqlen)
     else:
         raise ValueError(f"Unknown cp_style: {cp_style}")
-
-    input_ids = shard_for_cp(input_ids, cp_rank=cp_rank, cp_world_size=cp_world_size)
-    position_ids = shard_for_cp(position_ids, cp_rank=cp_rank, cp_world_size=cp_world_size)
-    return input_ids, position_ids
diff --git a/tests/unit/orchestrator/test_batch.py b/tests/unit/orchestrator/test_batch.py
index 6cc442b68c..2a6992e982 100644
--- a/tests/unit/orchestrator/test_batch.py
+++ b/tests/unit/orchestrator/test_batch.py
@@ -161,6 +161,7 @@ def test_randomized_packing_invariants():
         for batch in flat_batches:
             assert len(batch.input_ids) <= seq_len
             assert sum(batch.sequence_lengths) == len(batch.input_ids)
+            assert batch.seq_lens == batch.sequence_lengths
             assert sum(batch.lora_num_tokens) == len(batch.input_ids)
             assert len(batch.env_names) == len(batch.input_ids)
 
@@ -176,6 +177,7 @@ def test_pad_micro_batch_preserves_explicit_sequence_lengths():
 
     assert len(padded.input_ids) == 6
     assert padded.sequence_lengths == [4, 2]
+    assert padded.seq_lens == [4, 2]
     assert sum(padded.sequence_lengths) == len(padded.input_ids)
     assert padded.loss_mask[-2:] == [False, False]
 
@@ -271,6 +273,7 @@ def test_prepare_batch_packs_different_temperatures(make_training_example):
     # Second sample (4 tokens): all get temp 1.1
     assert flat_batches[0].temperatures[4:8] == [1.1, 1.1, 1.1, 1.1]
     assert flat_batches[0].env_names == ["env-a"] * 4 + ["env-b"] * 4
+    assert flat_batches[0].seq_lens == [4, 4]
 
 
 def test_prepare_sample_propagates_weight_streams(make_training_example):
@@ -445,6 +448,49 @@ def test_prepare_sample_truncates_mm_at_image_boundary():
     assert n_placeholders == mb.mm_kwargs["pixel_values"].shape[0]  # ppt == 1 here
 
 
+def test_prepare_batch_packs_multimodal_with_text():
+    mm_sample = TrainingSample(
+        token_ids=[10, 11, 12],
+        mask=[False, True, True],
+        logprobs=[0.0, -0.1, -0.2],
+        temperatures=[1.0, 1.0, 1.0],
+        advantages=[0.0, 1.0, 1.0],
+        env_name="mm-env",
+        mm_token_type_ids=[0, 1, 0],
+        mm_kwargs={
+            "pixel_values": _encoded(np.array([[1.0, 2.0]], dtype=np.float32)),
+            "image_grid_thw": _encoded(np.array([[1, 2, 2]], dtype=np.int64)),
+        },
+    )
+    text_sample = TrainingSample(
+        token_ids=[20, 21],
+        mask=[False, True],
+        logprobs=[0.0, -0.3],
+        temperatures=[0.7, 0.7],
+        advantages=[0.0, 1.0],
+        env_name="text-env",
+    )
+
+    batches_per_gpu = prepare_batch(
+        rollouts=[mm_sample, text_sample],
+        seq_len=8,
+        num_train_workers=1,
+        idxs=[0, 0],
+        num_loras=1,
+        bin_cost=build_bin_cost(None),
+    )
+
+    batch = batches_per_gpu[0][0]
+    assert batch.seq_lens == [3, 2]
+    assert batch.sequence_lengths == [3, 2]
+    assert batch.position_ids == [0, 1, 2, 0, 1]
+    assert batch.mm_token_type_ids == [0, 1, 0, 0, 0]
+    assert batch.mm_kwargs is not None
+    assert batch.mm_kwargs["pixel_values"].shape == [1, 2]
+    assert batch.mm_kwargs["image_grid_thw"].shape == [1, 3]
+    assert batch.env_names == ["mm-env"] * 3 + ["text-env"] * 2
+
+
 def test_prepare_sample_none_routed_experts():
     """When routed_experts is None, micro_batch.routed_experts is None."""
     sample = TrainingSample(
diff --git a/tests/unit/train/models/test_qwen3_5_moe_vlm.py b/tests/unit/train/models/test_qwen3_5_moe_vlm.py
index 7e6af33e7c..e24d216bb0 100644
--- a/tests/unit/train/models/test_qwen3_5_moe_vlm.py
+++ b/tests/unit/train/models/test_qwen3_5_moe_vlm.py
@@ -1,5 +1,10 @@
+import os
+import socket
+
 import pytest
 import torch
+import torch.distributed as dist
+import torch.multiprocessing as mp
 from transformers import AutoConfig
 from transformers.models.qwen3_5_moe.modeling_qwen3_5_moe import (
     Qwen3_5MoeForConditionalGeneration as HFQwen3_5MoeVLM,
@@ -13,11 +18,17 @@
 pytestmark = [pytest.mark.gpu]
 
 
-def _tiny_vlm_config():
+def _tiny_vlm_config(attn_implementation="sdpa"):
     """HF composite config shrunk for unit testing."""
-    config = AutoConfig.from_pretrained("Qwen/Qwen3.5-35B-A3B", trust_remote_code=True, attn_implementation="sdpa")
+    config = AutoConfig.from_pretrained(
+        "Qwen/Qwen3.5-35B-A3B",
+        trust_remote_code=True,
+        attn_implementation=attn_implementation,
+    )
     config.use_cache = False
+    config._attn_implementation = attn_implementation
     tc = config.text_config
+    tc._attn_implementation = attn_implementation
     tc.vocab_size = 256
     tc.hidden_size = 256
     tc.num_hidden_layers = 2
@@ -69,6 +80,99 @@ def _make_mm_token_type_ids(input_ids, image_token_id):
     return mm_token_type_ids
 
 
+def _free_tcp_port() -> int:
+    sock = socket.socket()
+    sock.bind(("127.0.0.1", 0))
+    port = sock.getsockname()[1]
+    sock.close()
+    return port
+
+
+def _make_packed_two_image_batch(config, *, dtype=torch.bfloat16):
+    device = "cuda"
+    vc = config.vision_config
+    patch_dim = vc.in_channels * vc.temporal_patch_size * vc.patch_size * vc.patch_size
+    image_grid_thw = torch.tensor([[1, 4, 4], [1, 4, 4]], device=device)
+    num_patches = int(image_grid_thw.prod(dim=1).sum().item())
+    pixel_values = torch.randn(num_patches, patch_dim, device=device, dtype=dtype)
+
+    image_tokens = torch.full((1, 4), config.image_token_id, device=device)
+    segment0 = torch.cat(
+        [torch.tensor([[10, 11]], device=device), image_tokens, torch.tensor([[12, 13]], device=device)], dim=1
+    )
+    segment1 = torch.cat(
+        [torch.tensor([[20, 21]], device=device), image_tokens, torch.tensor([[22, 23]], device=device)], dim=1
+    )
+    input_ids = torch.cat([segment0, segment1], dim=1)
+    return {
+        "input_ids": input_ids,
+        "pixel_values": pixel_values,
+        "image_grid_thw": image_grid_thw,
+        "mm_token_type_ids": _make_mm_token_type_ids(input_ids, config.image_token_id),
+        "seq_lens": torch.tensor([segment0.shape[1], segment1.shape[1]], device=device),
+        "temperatures": torch.ones_like(input_ids, dtype=torch.float32),
+    }
+
+
+def _qwen35_vlm_ulysses_equivalence_worker(rank: int, port: int):
+    os.environ["MASTER_ADDR"] = "127.0.0.1"
+    os.environ["MASTER_PORT"] = str(port)
+    dist.init_process_group("gloo", rank=rank, world_size=2)
+    try:
+        torch.cuda.set_device(0)
+        pytest.importorskip("flash_attn")
+        from prime_rl.trainer.models.layers.ulysses_attn import substitute_ulysses_attn
+
+        config = _tiny_vlm_config(attn_implementation="flash_attention_2")
+        config.text_config.layer_types = ["full_attention"] * config.text_config.num_hidden_layers
+
+        torch.manual_seed(2026)
+        batch = _make_packed_two_image_batch(config)
+        input_ids = batch["input_ids"]
+
+        torch.manual_seed(1234)
+        with torch.device("cuda"), default_dtype(torch.bfloat16):
+            baseline = Qwen3_5MoeForCausalLM(config)
+        inject_prime_lm_head(baseline)
+        baseline.eval()
+
+        with torch.no_grad():
+            baseline_logits = baseline(
+                input_ids=input_ids,
+                pixel_values=batch["pixel_values"],
+                image_grid_thw=batch["image_grid_thw"],
+                mm_token_type_ids=batch["mm_token_type_ids"],
+                seq_lens=batch["seq_lens"],
+            )["logits"].float()
+
+        torch.manual_seed(1234)
+        with torch.device("cuda"), default_dtype(torch.bfloat16):
+            cp_model = Qwen3_5MoeForCausalLM(config)
+        inject_prime_lm_head(cp_model)
+        cp_model.eval()
+
+        substitute_ulysses_attn(dist.group.WORLD, attn_impl="flash_attention_2")
+        with torch.no_grad():
+            local_logits = cp_model(
+                input_ids=input_ids,
+                pixel_values=batch["pixel_values"],
+                image_grid_thw=batch["image_grid_thw"],
+                mm_token_type_ids=batch["mm_token_type_ids"],
+                seq_lens=batch["seq_lens"],
+                cp_group=dist.group.WORLD,
+                cp_rank=rank,
+                cp_world_size=2,
+                cp_style="ulysses",
+            )["logits"].float()
+
+        gathered = [torch.empty_like(local_logits) for _ in range(2)]
+        dist.all_gather(gathered, local_logits)
+        cp_logits = torch.cat(gathered, dim=1)
+        torch.testing.assert_close(cp_logits, baseline_logits, rtol=5e-2, atol=5e-2)
+    finally:
+        dist.destroy_process_group()
+
+
 def test_vlm_forward():
     """Custom VLM produces logits for both text-only and multimodal inputs."""
     config = _tiny_vlm_config()
@@ -100,6 +204,32 @@ def test_vlm_forward():
     assert out_mm["logits"].shape == (1, input_ids_mm.shape[1], vocab)
 
 
+def test_vlm_text_only_forward_runs_dummy_vision(monkeypatch):
+    """Text-only VLM forward still enters the vision path with a zero-contribution dummy."""
+    config = _tiny_vlm_config()
+    with torch.device("cuda"), default_dtype(torch.float32):
+        model = Qwen3_5MoeForCausalLM(config)
+    inject_prime_lm_head(model)
+
+    calls = 0
+    original_forward = model.model.visual.forward
+
+    def wrapped_forward(*args, **kwargs):
+        nonlocal calls
+        calls += 1
+        return original_forward(*args, **kwargs)
+
+    monkeypatch.setattr(model.model.visual, "forward", wrapped_forward)
+
+    input_ids = torch.randint(0, 200, (1, 20), device="cuda")
+    seq_lens = torch.tensor([20], device="cuda")
+
+    out = model(input_ids=input_ids, seq_lens=seq_lens)
+
+    assert calls == 1
+    assert out["logits"].shape == (1, 20, config.text_config.vocab_size)
+
+
 def test_vlm_backward():
     """Gradients flow through both vision scatter and text model."""
     config = _tiny_vlm_config()
@@ -125,6 +255,19 @@ def test_vlm_backward():
     assert model.model.visual.patch_embed.proj.weight.grad is not None
 
 
+def test_vlm_ulysses_cp_matches_unsharded_packed_multimodal_forward():
+    pytest.importorskip("flash_attn")
+    if not torch.cuda.is_available():
+        pytest.skip("CUDA is required for Ulysses VLM CP equivalence")
+
+    mp.start_processes(
+        _qwen35_vlm_ulysses_equivalence_worker,
+        args=(_free_tcp_port(),),
+        nprocs=2,
+        start_method="spawn",
+    )
+
+
 def test_vlm_weight_load_from_hf():
     """Weights from HF VLM checkpoint load correctly into custom VLM after conversion.
 
diff --git a/tests/unit/train/rl/test_packer.py b/tests/unit/train/rl/test_packer.py
index c71dc2f58d..59721e86e3 100644
--- a/tests/unit/train/rl/test_packer.py
+++ b/tests/unit/train/rl/test_packer.py
@@ -1,6 +1,7 @@
 from pathlib import Path
 from typing import Generator
 
+import numpy as np
 import pytest
 import tomli_w
 import torch
@@ -12,7 +13,7 @@
 from prime_rl.trainer.runs import setup_multi_run_manager
 from prime_rl.trainer.utils import build_bin_cost
 from prime_rl.trainer.world import reset_world
-from prime_rl.transport.types import TrainingSample
+from prime_rl.transport.types import EncodedTensor, TrainingSample
 
 
 @pytest.fixture(autouse=True, scope="module")
@@ -52,6 +53,67 @@ def make_training_sample() -> TrainingSample:
     )
 
 
+def _encoded_tensor(data, dtype) -> EncodedTensor:
+    arr = np.asarray(data, dtype=dtype)
+    return EncodedTensor(dtype=str(arr.dtype), shape=list(arr.shape), data=arr.tobytes())
+
+
+def _decode_encoded_tensor(encoded: EncodedTensor):
+    return np.frombuffer(encoded.data, dtype=np.dtype(encoded.dtype)).reshape(encoded.shape).tolist()
+
+
+def _mm_sample(value: float, env_name: str = "test-env") -> TrainingSample:
+    return TrainingSample(
+        token_ids=[1, 250, 2],
+        mask=[False, True, True],
+        logprobs=[0.0, -0.1, -0.2],
+        temperatures=[1.0, 1.0, 1.0],
+        env_name=env_name,
+        advantages=[0.0, 1.0, 1.0],
+        mm_token_type_ids=[0, 1, 0],
+        mm_kwargs={
+            "pixel_values": _encoded_tensor([[value, value + 1]], np.float32),
+            "image_grid_thw": _encoded_tensor([[1, 2, 2]], np.int64),
+        },
+    )
+
+
+def _packer_with_two_runs(tmp_path, monkeypatch, dp_world_size, seq_len):
+    """Set up a MultiPacker over two discovered runs; capture sent grids."""
+    reset_world()
+    runs._MULTI_RUN_MANAGER = None
+    manager = setup_multi_run_manager(output_dir=tmp_path, max_runs=2, device=torch.device("cpu"))
+    create_run_with_config(tmp_path, "run_a")
+    create_run_with_config(tmp_path, "run_b")
+    manager.discover_runs()
+
+    sent: list = []
+
+    class DummyReceiver:
+        def receive(self):
+            return []
+
+        def reset_run(self, idx):
+            pass
+
+    class DummySender:
+        def send(self, micro_batch_grid):
+            sent.append(micro_batch_grid)
+
+    monkeypatch.setattr("prime_rl.trainer.rl.packer.setup_training_batch_receiver", lambda _c: DummyReceiver())
+    monkeypatch.setattr("prime_rl.trainer.rl.packer.setup_micro_batch_sender", lambda *a, **k: DummySender())
+    packer = MultiPacker(
+        dp_world_size=dp_world_size,
+        seq_len=seq_len,
+        pad_to_multiple_of=1,
+        tokenizer=None,
+        config=FileSystemTransportConfig(),
+        bin_cost=build_bin_cost(None),
+        start_step=0,
+    )
+    return manager, packer, sent
+
+
 def test_packer_progress_updates_once_per_run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     reset_world()
     runs._MULTI_RUN_MANAGER = None
@@ -114,3 +176,77 @@ def fake_sender(_output_dir, _data_world_size, _current_step, _config):
     micro_batch = sender.sent[0][0][0]
     assert micro_batch.run_id == "run_test123"
     assert micro_batch.run_step == 0
+
+
+def test_multipacker_pack_preserves_mm_kwargs_and_run_tagging(tmp_path, monkeypatch):
+    """MultiPacker packs each run separately and preserves multimodal sidecars."""
+    from prime_rl.trainer.batch import _is_multimodal_sample
+
+    manager, packer, sent = _packer_with_two_runs(tmp_path, monkeypatch, dp_world_size=2, seq_len=5)
+    a, b = manager.id_2_idx["run_a"], manager.id_2_idx["run_b"]
+    for idx, value in ((a, 1.0), (b, 10.0)):
+        packer.buffers[idx].append((_mm_sample(value), 0))
+        packer.buffers[idx].append((make_training_sample(), 0))
+
+    packer.pack()
+    assert sent, "pack() sent nothing"
+    grid = sent[-1]
+    assert len(grid) == 2
+
+    mm_mbs = [mb for rank in grid for mb in rank if _is_multimodal_sample(mb)]
+    assert mm_mbs, "no MM microbatches produced"
+    real_run_idxs = set()
+    for mb in mm_mbs:
+        assert mb.mm_kwargs is not None
+        if any(mb.loss_mask):
+            tagged = [i for i, n in enumerate(mb.lora_num_tokens) if n > 0]
+            assert len(tagged) == 1 and mb.lora_num_tokens[tagged[0]] == len(mb.input_ids)
+            real_run_idxs.add(tagged[0])
+    assert real_run_idxs == {a, b}, f"both runs' MM should be tagged; got {real_run_idxs}"
+
+
+def test_multipacker_pack_mm_padding_is_zero_loss(tmp_path, monkeypatch):
+    """A lone MM sample forces a dummy MM microbatch for rank padding; it must be zero-loss."""
+    from prime_rl.trainer.batch import _is_multimodal_sample
+
+    manager, packer, sent = _packer_with_two_runs(tmp_path, monkeypatch, dp_world_size=2, seq_len=3)
+    a, b = manager.id_2_idx["run_a"], manager.id_2_idx["run_b"]
+    packer.buffers[a].append((_mm_sample(1.0), 0))
+    packer.buffers[b].append((make_training_sample(), 0))
+
+    packer.pack()
+    assert sent
+    grid = sent[-1]
+    mm_mbs = [mb for rank in grid for mb in rank if _is_multimodal_sample(mb)]
+    dummies = [mb for mb in mm_mbs if not any(mb.loss_mask)]
+    assert dummies, "expected a zero-loss dummy MM padding microbatch"
+    for dummy in dummies:
+        assert all(advantage == 0.0 for advantage in dummy.advantages)
+
+
+def test_multipacker_pack_packs_mm_kwargs_within_each_run(tmp_path, monkeypatch):
+    """Eager multimodal samples pack within a run but never across runs."""
+    from prime_rl.trainer.batch import _is_multimodal_sample
+
+    manager, packer, sent = _packer_with_two_runs(tmp_path, monkeypatch, dp_world_size=1, seq_len=12)
+    a, b = manager.id_2_idx["run_a"], manager.id_2_idx["run_b"]
+    for idx, base in ((a, 1.0), (b, 10.0)):
+        packer.buffers[idx].append((_mm_sample(base), 0))
+        packer.buffers[idx].append((_mm_sample(base + 2), 0))
+
+    packer.pack()
+    assert sent
+    grid = sent[-1]
+    real_mm_mbs = [mb for rank in grid for mb in rank if _is_multimodal_sample(mb) and any(mb.loss_mask)]
+
+    assert len(real_mm_mbs) == 2
+    for mb in real_mm_mbs:
+        assert len(mb.input_ids) == 6
+        assert mb.position_ids == [0, 1, 2, 0, 1, 2]
+        assert mb.seq_lens == [3, 3]
+        assert mb.mm_kwargs is not None
+        assert mb.mm_kwargs["pixel_values"].shape == [2, 2]
+        assert mb.mm_kwargs["image_grid_thw"].shape == [2, 3]
+        assert len(_decode_encoded_tensor(mb.mm_kwargs["pixel_values"])) == 2
+        tagged = [i for i, n in enumerate(mb.lora_num_tokens) if n > 0]
+        assert len(tagged) == 1
diff --git a/tests/unit/train/test_model_forward.py b/tests/unit/train/test_model_forward.py
index 7baf7760a8..b1220244b2 100644
--- a/tests/unit/train/test_model_forward.py
+++ b/tests/unit/train/test_model_forward.py
@@ -2,8 +2,47 @@
 
 import torch
 import torch.nn as nn
+from transformers import PretrainedConfig
 
+import prime_rl.trainer.model as trainer_model
 from prime_rl.trainer.model import forward
+from prime_rl.trainer.models.base import PreTrainedModelPrimeRL
+
+
+class _TinyVLM(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.config = SimpleNamespace(tie_word_embeddings=False)
+        self.model = nn.Module()
+        self.model.visual = nn.Linear(4, 4)
+        self.model.language_model = nn.Module()
+        self.model.language_model.embed_tokens = nn.Embedding(8, 4)
+        self.model.language_model.layers = nn.ModuleList([nn.Linear(4, 4)])
+        self.model.language_model.norm = nn.LayerNorm(4)
+        self.lm_head = nn.Linear(4, 8, bias=False)
+
+
+class _FakeParallelDims:
+    ep_enabled = False
+
+    def get_mesh(self, name):
+        assert name == "hsdp"
+        return "hsdp-mesh"
+
+
+def _vlm_cp_config():
+    return SimpleNamespace(
+        reduce_dtype="bfloat16",
+        fsdp_cpu_offload=False,
+        reshard_after_forward=True,
+        cp=2,
+        name="tiny-vlm",
+        vlm=SimpleNamespace(
+            vision_encoder_attr="model.visual",
+            language_model_attr="model.language_model",
+            freeze_vision_encoder=True,
+        ),
+    )
 
 
 class _CaptureModel(nn.Module):
@@ -14,8 +53,79 @@ def __init__(self, config: SimpleNamespace):
 
     def forward(self, **kwargs):
         self.kwargs = kwargs
-        input_ids = kwargs["input_ids"]
-        return {"logits": torch.zeros(*input_ids.shape, 4)}
+        if "input_ids" in kwargs:
+            shape = kwargs["input_ids"].shape
+        else:
+            shape = kwargs["inputs_embeds"].shape[:2]
+        return {"logits": torch.zeros(*shape, 4)}
+
+
+class _PrimeCaptureModel(PreTrainedModelPrimeRL):
+    config_class = PretrainedConfig
+
+    def __init__(self):
+        super().__init__(PretrainedConfig())
+        self.kwargs = None
+
+    def prime_forward_kwargs(
+        self,
+        *,
+        seq_lens: torch.Tensor | None = None,
+        seq_lens_are_global: bool = False,
+    ) -> dict[str, object]:
+        return {
+            "seq_lens": seq_lens,
+            "seq_lens_are_global": seq_lens_are_global,
+            "hook_marker": True,
+        }
+
+    def forward(self, **kwargs):
+        self.kwargs = kwargs
+        shape = kwargs["input_ids"].shape
+        return {"logits": torch.zeros(*shape, 4)}
+
+
+def test_setup_fsdp_vlm_context_parallel_ignores_frozen_vision_encoder(monkeypatch):
+    model = _TinyVLM()
+    for param in model.model.visual.parameters():
+        param.requires_grad = False
+
+    calls = []
+
+    def fake_fully_shard(module, **kwargs):
+        calls.append((module, kwargs))
+
+    monkeypatch.setattr(trainer_model, "fully_shard", fake_fully_shard)
+
+    trainer_model.setup_fsdp(model, _vlm_cp_config(), _FakeParallelDims())
+
+    sharded_modules = [module for module, _ in calls]
+    assert model.model.language_model.embed_tokens in sharded_modules
+    assert model.model.visual not in sharded_modules
+
+    root_kwargs = calls[-1][1]
+    ignored_params = root_kwargs["ignored_params"]
+    assert set(model.model.visual.parameters()) <= ignored_params
+    assert model.model.language_model.embed_tokens.weight not in ignored_params
+
+
+def test_setup_fsdp_vlm_context_parallel_shards_trainable_vision_encoder(monkeypatch):
+    model = _TinyVLM()
+    calls = []
+
+    def fake_fully_shard(module, **kwargs):
+        calls.append((module, kwargs))
+
+    monkeypatch.setattr(trainer_model, "fully_shard", fake_fully_shard)
+
+    trainer_model.setup_fsdp(model, _vlm_cp_config(), _FakeParallelDims())
+
+    sharded_modules = [module for module, _ in calls]
+    assert model.model.visual in sharded_modules
+    assert model.model.language_model.embed_tokens in sharded_modules
+
+    root_kwargs = calls[-1][1]
+    assert root_kwargs["ignored_params"] is None
 
 
 def test_forward_passes_renderer_mm_token_type_ids_through():
@@ -81,3 +191,104 @@ def test_forward_keeps_position_ids_for_non_mrope_vlm():
 
     assert model.kwargs is not None
     torch.testing.assert_close(model.kwargs["position_ids"], position_ids)
+
+
+def test_forward_does_not_leak_seq_lens_to_generic_text_models():
+    model = _CaptureModel(SimpleNamespace(model_type="qwen3"))
+    input_ids = torch.tensor([[1, 2, 3, 4]])
+    position_ids = torch.arange(input_ids.shape[1]).unsqueeze(0)
+    seq_lens = torch.tensor([2, 2])
+
+    forward(model, input_ids, position_ids, seq_lens=seq_lens)
+
+    assert model.kwargs is not None
+    assert "seq_lens" not in model.kwargs
+    assert "seq_lens_are_global" not in model.kwargs
+
+
+def test_forward_merges_prime_forward_kwargs_for_custom_models():
+    model = _PrimeCaptureModel()
+    input_ids = torch.tensor([[1, 2, 3, 4]])
+    position_ids = torch.arange(input_ids.shape[1]).unsqueeze(0)
+    seq_lens = torch.tensor([2, 2])
+
+    forward(model, input_ids, position_ids, seq_lens=seq_lens, seq_lens_are_global=True)
+
+    assert model.kwargs is not None
+    torch.testing.assert_close(model.kwargs["seq_lens"], seq_lens)
+    assert model.kwargs["seq_lens_are_global"] is True
+    assert model.kwargs["hook_marker"] is True
+
+
+def test_forward_strips_position_ids_without_leaking_seq_lens_for_mrope_vlm():
+    """Generic VLMs do not receive PrimeRL-only packed-boundary kwargs."""
+    model = _CaptureModel(SimpleNamespace(model_type="qwen3_5_moe"))
+    input_ids = torch.tensor([[1, 10, 10, 2, 20, 20]])
+    position_ids = torch.tensor([[0, 1, 2, 0, 1, 2]])
+    seq_lens = torch.tensor([3, 3])
+
+    forward(
+        model,
+        input_ids,
+        position_ids,
+        mm_kwargs={"pixel_values": torch.ones(4, 3), "image_grid_thw": torch.tensor([[1, 1, 2], [1, 1, 2]])},
+        seq_lens=seq_lens,
+    )
+
+    assert model.kwargs is not None
+    assert "position_ids" not in model.kwargs
+    assert "seq_lens" not in model.kwargs
+
+
+def test_forward_accepts_premerged_inputs_embeds_without_cp_metadata():
+    model = _CaptureModel(SimpleNamespace(model_type="qwen3_5_moe"))
+    inputs_embeds = torch.randn(1, 4, 8)
+    position_ids = torch.arange(12).view(3, 1, 4)
+    seq_lens = torch.tensor([2, 2])
+
+    forward(
+        model,
+        None,
+        position_ids,
+        inputs_embeds=inputs_embeds,
+        seq_lens=seq_lens,
+        seq_lens_are_global=True,
+    )
+
+    assert model.kwargs is not None
+    assert "input_ids" not in model.kwargs
+    torch.testing.assert_close(model.kwargs["inputs_embeds"], inputs_embeds)
+    torch.testing.assert_close(model.kwargs["position_ids"], position_ids)
+    assert "seq_lens" not in model.kwargs
+    assert "seq_lens_are_global" not in model.kwargs
+
+
+def test_forward_passes_raw_vlm_inputs_with_context_parallel_metadata():
+    model = _CaptureModel(SimpleNamespace(model_type="qwen3_5_moe"))
+    input_ids = torch.tensor([[1, 10, 10, 2]])
+    position_ids = torch.arange(input_ids.shape[1]).unsqueeze(0)
+    seq_lens = torch.tensor([4])
+    cp_group = object()
+
+    forward(
+        model,
+        input_ids,
+        position_ids,
+        mm_kwargs={"pixel_values": torch.ones(2, 3), "image_grid_thw": torch.tensor([[1, 1, 2]])},
+        mm_token_type_ids=torch.tensor([[0, 1, 1, 0]]),
+        seq_lens=seq_lens,
+        cp_group=cp_group,
+        cp_rank=1,
+        cp_world_size=2,
+        cp_style="ulysses",
+    )
+
+    assert model.kwargs is not None
+    torch.testing.assert_close(model.kwargs["input_ids"], input_ids)
+    assert "inputs_embeds" not in model.kwargs
+    assert "position_ids" not in model.kwargs
+    assert "seq_lens" not in model.kwargs
+    assert model.kwargs["cp_group"] is cp_group
+    assert model.kwargs["cp_rank"] == 1
+    assert model.kwargs["cp_world_size"] == 2
+    assert model.kwargs["cp_style"] == "ulysses"
diff --git a/tests/unit/utils/test_cp.py b/tests/unit/utils/test_cp.py
new file mode 100644
index 0000000000..f2ca1be9b0
--- /dev/null
+++ b/tests/unit/utils/test_cp.py
@@ -0,0 +1,38 @@
+from types import SimpleNamespace
+
+import pytest
+import torch
+from torch import nn
+
+from prime_rl.utils.cp import assert_cp_style_supports_model, shard_for_cp, shard_position_ids_for_cp
+
+
+def test_shard_position_ids_for_cp_chunks_3d_mrope_positions_on_sequence_dim():
+    position_ids = torch.arange(24).view(3, 1, 8)
+
+    rank0 = shard_position_ids_for_cp(position_ids, cp_rank=0, cp_world_size=2)
+    rank1 = shard_position_ids_for_cp(position_ids, cp_rank=1, cp_world_size=2)
+
+    torch.testing.assert_close(rank0, position_ids[:, :, :4])
+    torch.testing.assert_close(rank1, position_ids[:, :, 4:])
+
+
+def test_shard_for_cp_rejects_non_divisible_sequence_dim():
+    with pytest.raises(ValueError, match="divisible by cp size"):
+        shard_for_cp(torch.arange(5).view(1, 5), cp_rank=0, cp_world_size=2)
+
+
+def test_ulysses_guard_requires_attention_heads_divisible_by_cp():
+    model = nn.Module()
+    model.config = SimpleNamespace(num_attention_heads=3, num_key_value_heads=2)
+
+    with pytest.raises(ValueError, match="num_attention_heads"):
+        assert_cp_style_supports_model("ulysses", model, cp_world_size=2)
+
+
+def test_ulysses_guard_requires_kv_heads_divisible_by_cp():
+    model = nn.Module()
+    model.config = SimpleNamespace(num_attention_heads=4, num_key_value_heads=3)
+
+    with pytest.raises(ValueError, match="num_key_value_heads"):
+        assert_cp_style_supports_model("ulysses", model, cp_world_size=2)