diff --git a/miles/backends/megatron_utils/lora/actor.py b/miles/backends/megatron_utils/lora/actor.py
index f809aa2108..3b48b02bb3 100644
--- a/miles/backends/megatron_utils/lora/actor.py
+++ b/miles/backends/megatron_utils/lora/actor.py
@@ -1,12 +1,9 @@
 from contextlib import ExitStack
 
-import ray
-
 from miles.backends.megatron_utils.actor import MegatronTrainRayActor
 from miles.backends.megatron_utils.lora import model as lora_model
 from miles.backends.megatron_utils.lora.optimizer import SlotOptimizer
-from miles.backends.training_utils.data import get_data_iterator, get_rollout_data
-from miles.backends.training_utils.parallel import get_parallel_state
+from miles.backends.training_utils.data import get_rollout_data
 from miles.utils.ray_utils import Box
 from miles.utils.tracking_utils.structured_log import with_logs
 
@@ -24,21 +21,19 @@ def forward_backward(self, batch_id: int, rollout_data_ref: Box) -> dict:
             return lora_model.run_forward_backward(self.args, batch_id, self.model, rollout_data)
 
     @with_logs
-    def optim_step(self, adam_params_by_slot: dict[int, dict]) -> dict[int, float]:
+    def optim_step(self, adam_params_by_slot: dict[int, dict]) -> dict[int, dict]:
         self._heartbeat.bump()
-        return lora_model.optim_step(self.args, self.slot_optimizers, adam_params_by_slot)
+        return lora_model.optim_step(self.slot_optimizers, adam_params_by_slot)
 
     @with_logs
-    def forward_only_logprobs(self, batch_id: int, rollout_data_ref: Box) -> Box | None:
+    def forward_only(self, batch_id: int, rollout_data_ref: Box) -> dict:
+        """Same loss pass as forward_backward, without the backward: the Tinker
+        forward() contract returns the requested loss per datum."""
         self._heartbeat.bump()
         with ExitStack() as stack:
             rollout_data, store_get_result = get_rollout_data(self.args, rollout_data_ref)
             stack.enter_context(store_get_result)
-            data_iterator, num_microbatches = get_data_iterator(self.args, self.model, rollout_data)
-            outputs = self.compute_log_prob(data_iterator, num_microbatches, rollout_id=batch_id)
-        if not get_parallel_state().is_pp_last_stage:
-            return None
-        return Box(ray.put({key: [t.cpu() for t in tensors] for key, tensors in outputs.items()}))
+            return lora_model.run_forward_backward(self.args, batch_id, self.model, rollout_data, forward_only=True)
 
     @with_logs
     def load_slot(self, slot: int, rank: int, alpha: float) -> None:
diff --git a/miles/backends/megatron_utils/lora/model.py b/miles/backends/megatron_utils/lora/model.py
index 1c8ddc13a1..4935f2db62 100644
--- a/miles/backends/megatron_utils/lora/model.py
+++ b/miles/backends/megatron_utils/lora/model.py
@@ -23,6 +23,8 @@ def run_forward_backward(
     batch_id: int,
     model: Sequence[DDP],
     rollout_data: RolloutBatch,
+    *,
+    forward_only: bool = False,
 ) -> dict:
     data_iterator, num_microbatches = get_data_iterator(args, model, rollout_data)
     assert len(num_microbatches) == 1, "a work unit is a single forward/backward pass"
@@ -33,27 +35,33 @@ def run_forward_backward(
         model_chunk.train()
     # disable_optimizer: bf16 loss scaling is the identity, so the pass needs no optimizer
     setup_train_iteration_config(args, model, None, disable_optimizer=True)
-    reset_grad_metadata_keep_grads(model)
+    if not forward_only:
+        reset_grad_metadata_keep_grads(model)
 
     dumper_phase_util = DumperMegatronUtil(args, model, DumperPhase.FWD_BWD, rollout_id=batch_id)
     losses_reduced = run_forward_backward_pass(
-        args, dumper_phase_util, data_iterator, model, num_microbatches[0], num_rollouts=None
+        args,
+        dumper_phase_util,
+        data_iterator,
+        model,
+        num_microbatches[0],
+        num_rollouts=None,
+        forward_only=forward_only,
     )
+    per_datum_outputs = [output for microbatch in losses_reduced for output in microbatch["per_datum"]]
     dumper_phase_util.finalize(model)
 
     if get_parallel_state().is_pp_last_stage:
-        return aggregate_train_losses(losses_reduced, None)
-    return {}
+        return {"metrics": aggregate_train_losses(losses_reduced, None), "per_datum": per_datum_outputs}
+    return {"metrics": {}, "per_datum": per_datum_outputs}
 
 
 def optim_step(
-    args: Namespace,
     slot_optimizers: dict[int, SlotOptimizer],
     adam_params_by_slot: dict[int, dict],
 ) -> dict[int, dict]:
-    # batch size 1: grads step as accumulated; normalization is the client's loss weights
     stepped = {slot: slot_optimizers[slot] for slot in adam_params_by_slot}
-    return step_slot_optimizers(stepped, adam_params_by_slot, clip_grad=args.clip_grad)
+    return step_slot_optimizers(stepped, adam_params_by_slot)
 
 
 def load_slot(args: Namespace, model: Sequence[DDP], slot: int, rank: int, alpha: float) -> SlotOptimizer:
diff --git a/miles/backends/megatron_utils/lora/optimizer.py b/miles/backends/megatron_utils/lora/optimizer.py
index 1c4da4365e..77ebd1f70a 100644
--- a/miles/backends/megatron_utils/lora/optimizer.py
+++ b/miles/backends/megatron_utils/lora/optimizer.py
@@ -192,8 +192,6 @@ def reset_grad_metadata_keep_grads(model_chunks) -> None:
 def step_slot_optimizers(
     slot_optimizers: dict[int, SlotOptimizer],
     adam_params_by_slot: dict[int, dict],
-    *,
-    clip_grad: float,
 ) -> dict[int, dict]:
     """Step slots in the same order on every rank; execution failures invalidate the cell."""
     slots = sorted(adam_params_by_slot)
@@ -201,7 +199,9 @@ def step_slot_optimizers(
         slot_optimizers[slot].apply_adam_params(adam_params_by_slot[slot])
         slot_optimizers[slot].prepare_grads()
 
-    outcomes = {slot: slot_optimizers[slot].clip_and_step(clip_grad) for slot in slots}
+    outcomes = {
+        slot: slot_optimizers[slot].clip_and_step(adam_params_by_slot[slot]["grad_clip_norm"]) for slot in slots
+    }
     for slot in slots:
         slot_optimizers[slot].zero_grads()
     for slot in slots:
diff --git a/miles/backends/megatron_utils/model.py b/miles/backends/megatron_utils/model.py
index 278fbbb466..45dd432c08 100644
--- a/miles/backends/megatron_utils/model.py
+++ b/miles/backends/megatron_utils/model.py
@@ -302,21 +302,8 @@ def forward_only(
     def forward_step(
         data_iterator: DataIterator, model: GPTModel, return_schedule_plan: bool = False
     ) -> tuple[torch.Tensor, Callable[[torch.Tensor], dict[str, list[torch.Tensor]]]]:
-        """Forward step used by Megatron's pipeline engine.
+        """Return the model output and its batch-bound loss callback."""
 
-        Args:
-            data_iterator (DataIterator): Input data iterator.
-            model (GPTModel): The GPT model chunk to execute.
-
-        Returns:
-            tuple[torch.Tensor, Callable[[torch.Tensor], dict[str, list[torch.Tensor]]]]:
-            Output tensor(s) and a callable that computes and packages results
-            to be collected by the engine.
-        """
-
-        assert not return_schedule_plan, "forward_only step should never return schedule plan"
-
-        # Get the batch.
         batch = get_batch(
             data_iterator,
             [
@@ -415,13 +402,15 @@ def _zero_grads(model: Sequence[DDP], optimizer: MegatronOptimizer | None, disab
         optimizer.zero_grad()
 
 
-def run_forward_backward_pass(args, dumper_phase_util, data_iterator, model, num_microbatches, num_rollouts):
+def run_forward_backward_pass(
+    args, dumper_phase_util, data_iterator, model, num_microbatches, num_rollouts, forward_only=False
+):
     """One pipeline forward/backward pass over the microbatches; no optimizer interaction."""
 
     @dumper_phase_util.wrap_forward_step
     def forward_step(data_iterator: DataIterator, model: GPTModel, return_schedule_plan: bool = False) -> tuple[
         torch.Tensor,
-        Callable[[torch.Tensor], tuple[torch.Tensor, int, dict[str, torch.Tensor | list[str]]]],
+        Callable[[torch.Tensor], tuple[torch.Tensor, int, dict]],
     ]:
         """Forward step used by Megatron's pipeline engine during training.
 
@@ -430,7 +419,7 @@ def forward_step(data_iterator: DataIterator, model: GPTModel, return_schedule_p
             model (GPTModel): The GPT model chunk to execute.
 
         Returns:
-            tuple[torch.Tensor, Callable[[torch.Tensor], tuple[torch.Tensor, int, dict[str, torch.Tensor | list[str]]]]]:
+            tuple[torch.Tensor, Callable[[torch.Tensor], tuple[torch.Tensor, int, dict]]]:
             Output tensor(s) and the loss function, which returns
             (loss, num_elems, {"keys": list[str], "values": torch.Tensor}).
         """
@@ -455,6 +444,9 @@ def forward_step(data_iterator: DataIterator, model: GPTModel, return_schedule_p
                 "witness_ids",
                 "opd_reverse_kl",
                 "rollout_mask_sums",
+                "loss_weights",
+                "target_tokens",
+                "sample_indices",
             ],
             args.data_pad_size_multiplier,
             args.qkv_format,
@@ -515,16 +507,17 @@ def forward_step(data_iterator: DataIterator, model: GPTModel, return_schedule_p
 
     # Forward pass.
     forward_backward_func = get_forward_backward_func()
-    return forward_backward_func(
-        forward_step_func=forward_step,
-        data_iterator=data_iterator,
-        model=model,
-        num_microbatches=num_microbatches,
-        seq_length=args.seq_length,
-        micro_batch_size=args.micro_batch_size,
-        decoder_seq_length=args.decoder_seq_length,
-        forward_only=False,
-    )
+    with torch.no_grad() if forward_only else nullcontext():
+        return forward_backward_func(
+            forward_step_func=forward_step,
+            data_iterator=data_iterator,
+            model=model,
+            num_microbatches=num_microbatches,
+            seq_length=args.seq_length,
+            micro_batch_size=args.micro_batch_size,
+            decoder_seq_length=args.decoder_seq_length,
+            forward_only=forward_only,
+        )
 
 
 def train_one_step(
diff --git a/miles/backends/training_utils/data.py b/miles/backends/training_utils/data.py
index 0716074c76..45d0c049ef 100644
--- a/miles/backends/training_utils/data.py
+++ b/miles/backends/training_utils/data.py
@@ -210,6 +210,10 @@ def get_batch(
     if "dynamic_global_batch_size" in data_iterator.rollout_data:
         batch["dynamic_global_batch_size"] = data_iterator.rollout_data["dynamic_global_batch_size"]
 
+    for unit_key in ("loss_fn", "loss_fn_config"):
+        if unit_key in data_iterator.rollout_data:
+            batch[unit_key] = data_iterator.rollout_data[unit_key]
+
     # No-op safety net if batches reach get_batch without rollout-level preprocessing.
     expand_multimodal_rollout_data_in_place(batch, qkv_format=qkv_format)
 
diff --git a/miles/backends/training_utils/loss.py b/miles/backends/training_utils/loss.py
index 92e89d9940..e68a253afc 100644
--- a/miles/backends/training_utils/loss.py
+++ b/miles/backends/training_utils/loss.py
@@ -132,11 +132,12 @@ def loss_function(
     logits: torch.Tensor,
     apply_megatron_loss_scaling: bool = False,
     num_rollouts: int | None = None,
-) -> tuple[torch.Tensor, int | torch.Tensor, dict[str, list[str] | torch.Tensor]]:
+) -> tuple[torch.Tensor, int | torch.Tensor, dict]:
     """Dispatch to the configured loss and rescale for Megatron integration.
 
     Selects one of "policy_loss", "value_loss", "sft_loss", or a custom loss
-    function based on `args.loss_type`, computes the loss and metrics, then
+    function based on `args.loss_type`, or the Tinker loss specified by
+    `batch["loss_fn"]`. Computes the loss and metrics, then
     rescales the loss by micro-batch and parallelism factors to integrate with
     Megatron's gradient accumulation.
 
@@ -157,6 +158,7 @@ def loss_function(
           `args.calculate_per_token_loss` is True, else `1` (int).
         - `logging_dict` has keys "keys" (list of str metric names) and
           "values" (1D tensor: [count, metric1, metric2, ...]).
+          Tinker losses may also return "per_datum" outputs.
     """
     parallel_state = get_parallel_state()
     num_tokens = sum([torch.clamp_min(loss_mask.sum(), 1) for loss_mask in batch["loss_masks"]])
@@ -172,7 +174,7 @@ def loss_function(
         denominators=batch.get("rollout_mask_sums", None),
     )
 
-    func = get_loss_function(args)
+    func = get_loss_function(args, batch.get("loss_fn"))
 
     if args.recompute_loss_function:
         loss, log = checkpoint(
@@ -196,11 +198,8 @@ def loss_function(
     else:
         assert args.use_dynamic_global_batch_size == ("dynamic_global_batch_size" in batch)
         global_batch_size = batch.get("dynamic_global_batch_size", args.global_batch_size)
-    # Multi-LoRA: samples enter the gradient buffers with weight 1; per-adapter
-    # normalization (1/adapter_global_batch_size, a constant known in advance)
-    # is applied to the accumulated slot gradient at optimizer-step time.
-    if is_multi_lora_enabled(args):
-        global_batch_size = 1
+    # Tinker losses already carry the client's normalization
+    loss_normalizer = 1 if is_multi_lora_enabled(args) else global_batch_size
     if not args.calculate_per_token_loss:
         if apply_megatron_loss_scaling:
             loss_parallel_size = (
@@ -208,17 +207,19 @@ def loss_function(
                 if args.true_on_policy_mode and parallel_state.is_ulysses_cp
                 else parallel_state.intra_dp_cp.size
             )
-            loss = loss * num_microbatches / global_batch_size * loss_parallel_size
+            loss = loss * num_microbatches / loss_normalizer * loss_parallel_size
         else:
-            loss = loss / global_batch_size * parallel_state.intra_dp.size
+            loss = loss / loss_normalizer * parallel_state.intra_dp.size
     else:
         if apply_megatron_loss_scaling:
             loss = loss * parallel_state.cp.size
 
+    per_datum = log.pop("per_datum", None) if batch.get("loss_fn") is not None else None
     return (
         loss,
         torch.tensor(num_tokens if args.calculate_per_token_loss else 1, device=logits.device),
         {
+            **({"per_datum": per_datum} if per_datum is not None else {}),
             "keys": list(log.keys()),
             "values": torch.tensor(
                 [num_samples if not args.calculate_per_token_loss else num_tokens] + list(log.values()),
diff --git a/miles/backends/training_utils/loss_hub/losses.py b/miles/backends/training_utils/loss_hub/losses.py
index 2d5a207738..d2d40b0693 100644
--- a/miles/backends/training_utils/loss_hub/losses.py
+++ b/miles/backends/training_utils/loss_hub/losses.py
@@ -505,7 +505,11 @@ def sft_loss_function(
     )
 
 
-def get_loss_function(args: Namespace) -> LossFunction:
+def get_loss_function(args: Namespace, loss_fn: str | None = None) -> LossFunction:
+    if loss_fn is not None:
+        from miles.backends.training_utils.loss_hub.tinker_losses import TINKER_LOSS_FUNCTIONS
+
+        return TINKER_LOSS_FUNCTIONS[loss_fn]
     match args.loss_type:
         case "policy_loss":
             return policy_loss_function
diff --git a/miles/backends/training_utils/loss_hub/tinker_losses.py b/miles/backends/training_utils/loss_hub/tinker_losses.py
new file mode 100644
index 0000000000..27f2fade1c
--- /dev/null
+++ b/miles/backends/training_utils/loss_hub/tinker_losses.py
@@ -0,0 +1,163 @@
+"""Sum Tinker losses over datums without server-side normalization.
+
+Client loss inputs carry normalization; the trainer accumulates raw sums.
+The SDK represents custom-loss gradients as `weights = -dL/dlogprob` with `cross_entropy`.
+"""
+
+from argparse import Namespace
+from collections.abc import Callable
+
+import torch
+
+from miles.backends.training_utils.loss_hub.logit_processors import get_log_probs_and_entropy
+from miles.utils.types import RolloutBatch
+
+PPO_DEFAULTS = {"clip_low_threshold": 0.8, "clip_high_threshold": 1.2}
+CISPO_DEFAULTS = {"clip_low_threshold": 0.0, "clip_high_threshold": 4.0}
+DRO_DEFAULTS = {"beta": 0.05}
+
+
+def _target_logprobs(args: Namespace, batch: RolloutBatch, logits: torch.Tensor) -> list[torch.Tensor]:
+    # Tinker targets are explicit labels: splice them over the response region of the gather sequence
+    label_tokens = [
+        torch.cat([tokens[: len(tokens) - len(targets)], _as_tensor_like(targets, tokens)])
+        for tokens, targets in zip(batch["unconcat_tokens"], batch["target_tokens"], strict=True)
+    ]
+    outputs = get_log_probs_and_entropy(
+        logits,
+        args=args,
+        unconcat_tokens=label_tokens,
+        total_lengths=batch["total_lengths"],
+        response_lengths=batch["response_lengths"],
+        with_entropy=False,
+        max_seq_lens=batch.get("max_seq_lens", None),
+    )
+    return outputs["log_probs"]
+
+
+def _as_tensor_like(values, reference: torch.Tensor) -> torch.Tensor:
+    return torch.as_tensor(values, dtype=reference.dtype, device=reference.device)
+
+
+def _response_masks(batch: RolloutBatch, log_probs: list[torch.Tensor]) -> list[torch.Tensor]:
+    """Per-datum loss masks; a DP-padding datum is all zeros and must not reach the objective."""
+    return [_as_tensor_like(mask, log_prob) for mask, log_prob in zip(batch["loss_masks"], log_probs, strict=True)]
+
+
+def _sum_loss_and_outputs(
+    batch: RolloutBatch,
+    logits: torch.Tensor,
+    log_probs: list[torch.Tensor],
+    per_datum_losses: list[torch.Tensor],
+) -> tuple[torch.Tensor, dict]:
+    if per_datum_losses:
+        loss = torch.stack(per_datum_losses).sum()
+    else:
+        # a microbatch with no supervised tokens still needs the graph alive; fp32 sum avoids fp16 inf -> nan
+        loss = logits.sum(dtype=torch.float32) * 0
+    per_datum = [
+        {"sample_index": index, "logprobs": log_prob.detach().cpu(), "loss": sample_loss.detach().cpu()}
+        for index, log_prob, sample_loss in zip(batch["sample_indices"], log_probs, per_datum_losses, strict=True)
+    ]
+    return loss, {"loss": loss.detach(), "per_datum": per_datum}
+
+
+def cross_entropy_loss_function(
+    args: Namespace,
+    batch: RolloutBatch,
+    logits: torch.Tensor,
+    sum_of_sample_mean: Callable[[torch.Tensor], torch.Tensor],
+) -> tuple[torch.Tensor, dict]:
+    log_probs = _target_logprobs(args, batch, logits)
+    per_datum_losses = [
+        -(_as_tensor_like(weights, log_prob) * log_prob * mask).sum()
+        for log_prob, weights, mask in zip(
+            log_probs, batch["loss_weights"], _response_masks(batch, log_probs), strict=True
+        )
+    ]
+    return _sum_loss_and_outputs(batch, logits, log_probs, per_datum_losses)
+
+
+def importance_sampling_loss_function(
+    args: Namespace,
+    batch: RolloutBatch,
+    logits: torch.Tensor,
+    sum_of_sample_mean: Callable[[torch.Tensor], torch.Tensor],
+) -> tuple[torch.Tensor, dict]:
+    log_probs = _target_logprobs(args, batch, logits)
+    per_datum_losses = []
+    for log_prob, sampling_log_prob, advantage, mask in zip(
+        log_probs, batch["rollout_log_probs"], batch["advantages"], _response_masks(batch, log_probs), strict=True
+    ):
+        ratio = torch.exp(log_prob - _as_tensor_like(sampling_log_prob, log_prob))
+        per_datum_losses.append(-(ratio * _as_tensor_like(advantage, log_prob) * mask).sum())
+    return _sum_loss_and_outputs(batch, logits, log_probs, per_datum_losses)
+
+
+def ppo_loss_function(
+    args: Namespace,
+    batch: RolloutBatch,
+    logits: torch.Tensor,
+    sum_of_sample_mean: Callable[[torch.Tensor], torch.Tensor],
+) -> tuple[torch.Tensor, dict]:
+    config = batch.get("loss_fn_config") or {}
+    clip_low = config.get("clip_low_threshold", PPO_DEFAULTS["clip_low_threshold"])
+    clip_high = config.get("clip_high_threshold", PPO_DEFAULTS["clip_high_threshold"])
+    log_probs = _target_logprobs(args, batch, logits)
+    per_datum_losses = []
+    for log_prob, sampling_log_prob, advantage, mask in zip(
+        log_probs, batch["rollout_log_probs"], batch["advantages"], _response_masks(batch, log_probs), strict=True
+    ):
+        ratio = torch.exp(log_prob - _as_tensor_like(sampling_log_prob, log_prob))
+        advantages = _as_tensor_like(advantage, log_prob)
+        objective = torch.minimum(ratio * advantages, torch.clamp(ratio, clip_low, clip_high) * advantages)
+        per_datum_losses.append(-(objective * mask).sum())
+    return _sum_loss_and_outputs(batch, logits, log_probs, per_datum_losses)
+
+
+def cispo_loss_function(
+    args: Namespace,
+    batch: RolloutBatch,
+    logits: torch.Tensor,
+    sum_of_sample_mean: Callable[[torch.Tensor], torch.Tensor],
+) -> tuple[torch.Tensor, dict]:
+    config = batch.get("loss_fn_config") or {}
+    clip_low = config.get("clip_low_threshold", CISPO_DEFAULTS["clip_low_threshold"])
+    clip_high = config.get("clip_high_threshold", CISPO_DEFAULTS["clip_high_threshold"])
+    log_probs = _target_logprobs(args, batch, logits)
+    per_datum_losses = []
+    for log_prob, sampling_log_prob, advantage, mask in zip(
+        log_probs, batch["rollout_log_probs"], batch["advantages"], _response_masks(batch, log_probs), strict=True
+    ):
+        ratio = torch.exp(log_prob - _as_tensor_like(sampling_log_prob, log_prob))
+        coefficient = torch.clamp(ratio, clip_low, clip_high).detach()
+        per_datum_losses.append(-(coefficient * log_prob * _as_tensor_like(advantage, log_prob) * mask).sum())
+    return _sum_loss_and_outputs(batch, logits, log_probs, per_datum_losses)
+
+
+def dro_loss_function(
+    args: Namespace,
+    batch: RolloutBatch,
+    logits: torch.Tensor,
+    sum_of_sample_mean: Callable[[torch.Tensor], torch.Tensor],
+) -> tuple[torch.Tensor, dict]:
+    config = batch.get("loss_fn_config") or {}
+    beta = config.get("beta", DRO_DEFAULTS["beta"])
+    log_probs = _target_logprobs(args, batch, logits)
+    per_datum_losses = []
+    for log_prob, sampling_log_prob, advantage, mask in zip(
+        log_probs, batch["rollout_log_probs"], batch["advantages"], _response_masks(batch, log_probs), strict=True
+    ):
+        divergence = log_prob - _as_tensor_like(sampling_log_prob, log_prob)
+        objective = log_prob * _as_tensor_like(advantage, log_prob) - 0.5 * beta * divergence**2
+        per_datum_losses.append(-(objective * mask).sum())
+    return _sum_loss_and_outputs(batch, logits, log_probs, per_datum_losses)
+
+
+TINKER_LOSS_FUNCTIONS = {
+    "cross_entropy": cross_entropy_loss_function,
+    "importance_sampling": importance_sampling_loss_function,
+    "ppo": ppo_loss_function,
+    "cispo": cispo_loss_function,
+    "dro": dro_loss_function,
+}
diff --git a/miles/ray/rollout/train_data_conversion.py b/miles/ray/rollout/train_data_conversion.py
index 9b3e513294..a647bb9d97 100644
--- a/miles/ray/rollout/train_data_conversion.py
+++ b/miles/ray/rollout/train_data_conversion.py
@@ -15,6 +15,7 @@
 
 ROLLOUT_DATA_TENSOR_DTYPES = {
     "tokens": "int32",
+    "target_tokens": "int32",
     "loss_masks": "int32",
     "rollout_log_probs": "float32",
     "rollout_sampling_mask_ids": "int32",
@@ -382,6 +383,9 @@ def _package_shards(args, data: dict[str, Any], partitions) -> list[dict[str, An
             "seq_witness_ids",
             "weight_versions",
             "adapter_slots",
+            "loss_weights",
+            "advantages",
+            "target_tokens",
         ]:
             if key not in data:
                 continue
@@ -393,6 +397,8 @@ def _package_shards(args, data: dict[str, Any], partitions) -> list[dict[str, An
             "total_lengths",
             "dynamic_global_batch_size",
             "prompt_group_sizes",
+            "loss_fn",
+            "loss_fn_config",
         ]:
             if key not in data:
                 continue
diff --git a/miles/ray/train/group.py b/miles/ray/train/group.py
index bac83a9a39..0aca8616ef 100644
--- a/miles/ray/train/group.py
+++ b/miles/ray/train/group.py
@@ -401,8 +401,8 @@ async def forward_backward(self, batch_id: int, data_ref) -> list:
     async def optim_step(self, adam_params_by_slot: dict[int, dict]) -> list:
         return await self._execute_slots("optim_step", adam_params_by_slot=adam_params_by_slot)
 
-    async def forward_only_logprobs(self, batch_id: int, data_ref) -> list:
-        return await self._execute_slots("forward_only_logprobs", batch_id=batch_id, rollout_data_ref=data_ref)
+    async def forward_only(self, batch_id: int, data_ref) -> list:
+        return await self._execute_slots("forward_only", batch_id=batch_id, rollout_data_ref=data_ref)
 
     async def load_slot(self, slot: int, rank: int, alpha: float) -> None:
         await self._execute_slots("load_slot", slot=slot, rank=rank, alpha=alpha)
diff --git a/miles/utils/multi_lora.py b/miles/utils/multi_lora.py
index c05da92436..018a71c556 100644
--- a/miles/utils/multi_lora.py
+++ b/miles/utils/multi_lora.py
@@ -60,6 +60,10 @@ def validate_multi_lora_args(args: Any) -> None:
     assert args.target_modules is not None, "--target-modules must be set when --multi-lora-n-adapters > 0"
     assert args.train_backend == "megatron", "Multi-LoRA currently requires --train-backend megatron"
     # Adapter routing is only recompute-safe without pipelining; enforce at launch.
+    assert getattr(args, "context_parallel_size", 1) == 1, (
+        "multi-LoRA requires --context-parallel-size 1: the Tinker losses zip "
+        "full-length per-datum vectors against log_probs, which CP would shard"
+    )
     assert getattr(args, "pipeline_model_parallel_size", 1) == 1, (
         "Multi-LoRA requires --pipeline-model-parallel-size 1: no single rank holds a "
         "complete adapter to push to the rollout engines, and a pipelined schedule would "
diff --git a/tests/fast/backends/megatron_utils/test_step_adapter_slots.py b/tests/fast/backends/megatron_utils/test_step_adapter_slots.py
index 932d1fda38..87f25a8206 100644
--- a/tests/fast/backends/megatron_utils/test_step_adapter_slots.py
+++ b/tests/fast/backends/megatron_utils/test_step_adapter_slots.py
@@ -38,8 +38,8 @@ def allgather_params(self):
 
 
 def _step(slot_optimizers):
-    adam = {slot: {"learning_rate": 1e-4} for slot in slot_optimizers}
-    return step_slot_optimizers(slot_optimizers, adam, clip_grad=1.0)
+    adam = {slot: {"learning_rate": 1e-4, "grad_clip_norm": 1.0} for slot in slot_optimizers}
+    return step_slot_optimizers(slot_optimizers, adam)
 
 
 def test_an_unknown_step_failure_escapes_before_success_is_reported():
diff --git a/tests/fast/backends/training_utils/loss/test_tinker_explicit_targets.py b/tests/fast/backends/training_utils/loss/test_tinker_explicit_targets.py
new file mode 100644
index 0000000000..959fe9ba14
--- /dev/null
+++ b/tests/fast/backends/training_utils/loss/test_tinker_explicit_targets.py
@@ -0,0 +1,58 @@
+"""target_tokens are explicit gather labels; shifted targets reproduce next-token scoring."""
+
+import torch
+from tests.fast.backends.training_utils.loss.loss_test_utils import make_args, make_parallel_state
+
+from miles.backends.training_utils.loss_hub import math_utils, tinker_losses
+
+VOCAB = 32
+
+
+def _naive_compute_log_probs(logits, tokens, _tp_group, *, sampling_mask=None):
+    return torch.log_softmax(logits, dim=-1).gather(-1, tokens.unsqueeze(-1))
+
+
+def _target_logprobs(batch, logits):
+    args = make_args(true_on_policy_mode=False, rollout_temperature=1.0, log_probs_chunk_size=64)
+    return tinker_losses._target_logprobs(args, batch, logits)
+
+
+def _batch(sequences: list[list[int]], targets: list[list[int]]) -> dict:
+    return {
+        "unconcat_tokens": [torch.tensor(sequence) for sequence in sequences],
+        "target_tokens": targets,
+        "total_lengths": [len(sequence) for sequence in sequences],
+        "response_lengths": [len(target) for target in targets],
+    }
+
+
+def test_targets_replace_the_sequence_as_gather_labels(monkeypatch):
+    monkeypatch.setattr(math_utils, "compute_log_probs", _naive_compute_log_probs)
+    make_parallel_state()
+    generator = torch.Generator().manual_seed(0)
+    sequences = [[1, 2, 3, 4, 5], [6, 7, 8]]
+    targets = [[9, 10, 11], [12, 13]]  # rl_loop-style: labels differ from the sequence
+    logits = torch.randn((1, 8, VOCAB), generator=generator)
+
+    log_probs = _target_logprobs(_batch(sequences, targets), logits)
+
+    flat = torch.log_softmax(logits.squeeze(0), dim=-1)
+    offset = 0
+    for sequence, target, log_prob in zip(sequences, targets, log_probs, strict=True):
+        start = offset + len(sequence) - len(target) - 1
+        expected = [flat[start + i, label] for i, label in enumerate(target)]
+        torch.testing.assert_close(log_prob, torch.stack(expected))
+        offset += len(sequence)
+
+
+def test_shifted_targets_degenerate_to_next_token_scoring(monkeypatch):
+    monkeypatch.setattr(math_utils, "compute_log_probs", _naive_compute_log_probs)
+    make_parallel_state()
+    generator = torch.Generator().manual_seed(1)
+    sequence = torch.randint(0, VOCAB, (6,), generator=generator).tolist()
+    logits = torch.randn((1, 6, VOCAB), generator=generator)
+
+    (log_probs,) = _target_logprobs(_batch([sequence], [sequence[1:]]), logits)
+
+    expected = torch.log_softmax(logits.squeeze(0)[:-1], dim=-1).gather(-1, torch.tensor(sequence[1:]).unsqueeze(-1))
+    torch.testing.assert_close(log_probs, expected.squeeze(-1))
diff --git a/tests/fast/backends/training_utils/loss/test_tinker_loss_outputs.py b/tests/fast/backends/training_utils/loss/test_tinker_loss_outputs.py
new file mode 100644
index 0000000000..58d7563c13
--- /dev/null
+++ b/tests/fast/backends/training_utils/loss/test_tinker_loss_outputs.py
@@ -0,0 +1,142 @@
+"""Per-datum outputs belong to their loss pass, including when backward recomputes the loss."""
+
+from argparse import Namespace
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+from miles.backends.training_utils import loss as loss_module
+from miles.backends.training_utils.loss_hub import tinker_losses
+
+
+@pytest.mark.parametrize("recompute", [False, True], ids=["direct", "recomputed"])
+def test_loss_passes_return_independent_detached_outputs(monkeypatch, recompute):
+    parallel = SimpleNamespace(cp=SimpleNamespace(size=1), intra_dp=SimpleNamespace(size=1))
+    monkeypatch.setattr(loss_module, "get_parallel_state", lambda: parallel)
+    monkeypatch.setattr(loss_module, "get_sum_of_sample_mean", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(tinker_losses, "_target_logprobs", lambda _args, _batch, logits: [logits])
+    args = Namespace(
+        calculate_per_token_loss=False,
+        qkv_format="thd",
+        recompute_loss_function=recompute,
+        use_dynamic_global_batch_size=True,
+        global_batch_size=1,
+        multi_lora=True,
+    )
+    completed = []
+    for sample_index in (7, 11):
+        logprobs = torch.tensor([-0.5, -0.25], requires_grad=True)
+        batch = {
+            "loss_fn": "cross_entropy",
+            "loss_weights": [[2.0, 3.0]],
+            "loss_masks": [torch.ones(2)],
+            "total_lengths": [3],
+            "response_lengths": [2],
+            "sample_indices": [sample_index],
+            "dynamic_global_batch_size": 1,
+        }
+        loss, _, logging = loss_module.loss_function(args, batch, 1, logprobs)
+        loss.backward()
+        assert logprobs.grad.tolist() == [-2.0, -3.0]
+        assert logging["keys"] == ["loss"]
+        assert logging["values"].tolist() == [1.0, 1.75]
+        completed.append(logging["per_datum"])
+
+    assert [[output["sample_index"] for output in outputs] for outputs in completed] == [[7], [11]]
+    for outputs in completed:
+        assert len(outputs) == 1, "loss recomputation must not append another datum output"
+        assert outputs[0]["loss"].item() == 1.75
+        assert outputs[0]["logprobs"].tolist() == [-0.5, -0.25]
+        assert not outputs[0]["loss"].requires_grad
+        assert not outputs[0]["logprobs"].requires_grad
+
+
+@pytest.mark.parametrize("loss_fn", sorted(tinker_losses.TINKER_LOSS_FUNCTIONS))
+def test_a_zero_loss_mask_removes_the_datum_from_every_objective(monkeypatch, loss_fn):
+    logprobs = torch.tensor([-0.5, -0.25], requires_grad=True)
+    monkeypatch.setattr(tinker_losses, "_target_logprobs", lambda _args, _batch, logits: [logits])
+    batch = {
+        "loss_fn": loss_fn,
+        "loss_weights": [[2.0, 3.0]],
+        "advantages": [[1.0, 1.0]],
+        "rollout_log_probs": [[-1.5, -1.25]],  # differs from logprobs so the DRO divergence term is nonzero
+        "loss_masks": [torch.zeros(2)],
+        "total_lengths": [3],
+        "response_lengths": [2],
+        "sample_indices": [0],
+    }
+    loss, _ = tinker_losses.TINKER_LOSS_FUNCTIONS[loss_fn](Namespace(), batch, logprobs, None)
+    loss.backward()
+    assert loss.item() == 0.0
+    assert logprobs.grad.abs().sum().item() == 0.0, "a DP-padding datum must contribute no gradient"
+
+
+@pytest.mark.parametrize("recompute", [False, True], ids=["direct", "recomputed"])
+@pytest.mark.parametrize(
+    "loss_fn, config, token_losses, gradients",
+    [
+        ("importance_sampling", {}, [-1, 1.5, -2, 3, -10, 15], [-1, 1.5, -2, 3, -10, 15]),
+        ("ppo", {}, [-1, 2.4, -2, 3, -2.4, 15], [-1, 0, -2, 3, 0, 15]),
+        (
+            "ppo",
+            {"clip_low_threshold": 0.4, "clip_high_threshold": 6},
+            [-1, 1.5, -2, 3, -10, 15],
+            [-1, 1.5, -2, 3, -10, 15],
+        ),
+        ("cispo", {}, [1, -1.5, 2, -3, 8, -12], [-1, 1.5, -2, 3, -8, 12]),
+        (
+            "cispo",
+            {"clip_low_threshold": 0.75, "clip_high_threshold": 2},
+            [1.5, -2.25, 2, -3, 4, -6],
+            [-1.5, 2.25, -2, 3, -4, 6],
+        ),
+        ("dro", {}, None, None),
+        ("dro", {"beta": 0.5}, None, None),
+    ],
+    ids=["is", "ppo-default", "ppo-override", "cispo-default", "cispo-override", "dro-default", "dro-override"],
+)
+def test_nonzero_objectives_and_gradients(monkeypatch, recompute, loss_fn, config, token_losses, gradients):
+    parallel = SimpleNamespace(cp=SimpleNamespace(size=1), intra_dp=SimpleNamespace(size=1))
+    monkeypatch.setattr(loss_module, "get_parallel_state", lambda: parallel)
+    monkeypatch.setattr(loss_module, "get_sum_of_sample_mean", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(tinker_losses, "_target_logprobs", lambda _args, _batch, logits: [logits[:3], logits[3:]])
+    args = Namespace(
+        calculate_per_token_loss=False,
+        qkv_format="thd",
+        recompute_loss_function=recompute,
+        use_dynamic_global_batch_size=True,
+        global_batch_size=4,
+        multi_lora=True,
+    )
+    # Both advantage signs cross each clipping boundary; the last two tokens are masked.
+    ratios = torch.tensor([0.5, 0.5, 1, 1, 5, 5, 0.5, 5], dtype=torch.float64)
+    logprobs = torch.full((8,), -1.0, dtype=torch.float64, requires_grad=True)
+    sampling_logprobs = -1 - ratios.log()
+    batch = {
+        "loss_fn": loss_fn,
+        "loss_fn_config": config,
+        "advantages": [[2, -3, 2], [-3, 2, -3, 11, -13]],
+        "rollout_log_probs": [sampling_logprobs[:3], sampling_logprobs[3:]],
+        "loss_masks": [torch.ones(3), torch.tensor([1, 1, 1, 0, 0])],
+        "total_lengths": [4, 6],
+        "response_lengths": [3, 5],
+        "sample_indices": [7, 11],
+        "dynamic_global_batch_size": 4,
+    }
+    if loss_fn == "dro":
+        beta = 0.05 if not config else 0.5
+        advantages = torch.tensor([2, -3, 2, -3, 2, -3], dtype=torch.float64)
+        divergence = ratios[:6].log()
+        token_losses = (advantages + beta / 2 * divergence.square()).tolist()
+        gradients = (-advantages + beta * divergence).tolist()
+    expected_losses = torch.tensor([sum(token_losses[:3]), sum(token_losses[3:])], dtype=torch.float64)
+    loss, _, logging = loss_module.loss_function(args, batch, 1, logprobs)
+    loss.backward()
+    torch.testing.assert_close(loss, expected_losses.sum())
+    torch.testing.assert_close(logprobs.grad, torch.tensor(gradients + [0, 0], dtype=torch.float64))
+    outputs = logging["per_datum"]
+    assert [output["sample_index"] for output in outputs] == [7, 11]
+    torch.testing.assert_close(torch.stack([output["loss"] for output in outputs]), expected_losses)
+    torch.testing.assert_close(torch.cat([output["logprobs"] for output in outputs]), logprobs.detach())
+    assert all(not output["loss"].requires_grad and not output["logprobs"].requires_grad for output in outputs)
diff --git a/tests/fast/backends/training_utils/loss/test_training_logprob_reuse.py b/tests/fast/backends/training_utils/loss/test_training_logprob_reuse.py
index d403aeacdc..093580ee15 100644
--- a/tests/fast/backends/training_utils/loss/test_training_logprob_reuse.py
+++ b/tests/fast/backends/training_utils/loss/test_training_logprob_reuse.py
@@ -276,7 +276,7 @@ def test_loss_dispatcher_uses_args_without_extra_policy_kwargs(monkeypatch, skip
     batch = make_batch(inputs, "policy_loss")
     logits = deep_clone(inputs["policy_logits"])
     policy_loss = Mock(return_value=(logits.sum(), {"loss": logits.new_zeros(())}))
-    monkeypatch.setattr(loss_module, "get_loss_function", lambda _args: policy_loss)
+    monkeypatch.setattr(loss_module, "get_loss_function", lambda _args, _loss_fn=None: policy_loss)
 
     loss_module.loss_function(
         args,
@@ -296,7 +296,7 @@ def test_loss_dispatcher_does_not_apply_actor_skip_to_value_loss(monkeypatch):
     batch = make_batch(inputs, "value_loss")
     logits = deep_clone(inputs["value_logits"])
     value_loss = Mock(return_value=(logits.sum(), {"loss": logits.new_zeros(())}))
-    monkeypatch.setattr(loss_module, "get_loss_function", lambda _args: value_loss)
+    monkeypatch.setattr(loss_module, "get_loss_function", lambda _args, _loss_fn=None: value_loss)
 
     loss_module.loss_function(args, batch, 1, logits)
 
diff --git a/tests/fast/ray/rollout/test_train_data_conversion.py b/tests/fast/ray/rollout/test_train_data_conversion.py
index 1e839dc086..ff143ac1bc 100644
--- a/tests/fast/ray/rollout/test_train_data_conversion.py
+++ b/tests/fast/ray/rollout/test_train_data_conversion.py
@@ -960,3 +960,21 @@ def test_dynamic_gbs_multi_step(self):
         assert shards[0]["num_microbatches"] == [2, 2]
         assert shards[0]["dynamic_global_batch_size"] == 8
         assert shards[0]["num_rollouts"] == [8, 8]
+
+
+def test_delayed_dp_split_preserves_the_tinker_loss_vectors():
+    """The tinker losses zip loss_weights / advantages / rollout_log_probs per
+    datum; a key missing from the shard whitelist disappears silently and only
+    fails inside the trainer."""
+    from miles.ray.rollout.train_data_conversion import _package_shards
+
+    data = {
+        "tokens": [[1], [2], [3], [4]],
+        "loss_weights": [[1.0]] * 4,
+        "advantages": [[0.5]] * 4,
+        "rollout_log_probs": [[-0.1]] * 4,
+    }
+    shards = _package_shards(None, data, [[0, 2], [1, 3]])
+    for shard in shards:
+        assert shard["loss_weights"] and shard["advantages"] and shard["rollout_log_probs"]
+    assert shards[0]["advantages"] == [[0.5], [0.5]]