DaoyuanLi2816 · DaoyuanLi2816 · Aug 12, 2026 · Aug 12, 2026 · Aug 12, 2026 · Aug 12, 2026
diff --git a/.github/workflows/verl-bridge.yml b/.github/workflows/verl-bridge.yml
@@ -47,7 +47,7 @@ jobs:
           changed=$(git diff --name-only "origin/$BASE_REF"...HEAD)
           echo "changed files:"
           echo "$changed"
-          if echo "$changed" | grep -Eq '^(\.github/workflows/verl-bridge\.yml|pyproject\.toml|scripts/[^/]*verl_bridge[^/]*|src/miniverl/bridge/.*|tests/.*verl_bridge.*)$'; then
+          if echo "$changed" | grep -Eq '^(\.github/workflows/verl-bridge\.yml|pyproject\.toml|scripts/[^/]*verl_bridge[^/]*|src/miniverl/bridge/.*|src/miniverl/losses/verl_topk\.py|tests/.*verl_bridge.*|tests/conformance/test_verl_v08_loss\.py)$'; then
             echo "bridge=true" >> "$GITHUB_OUTPUT"
           else
             echo "bridge=false" >> "$GITHUB_OUTPUT"
@@ -68,11 +68,13 @@ jobs:
       - name: Install miniVERL bridge environment
         run: >-
           python -m pip install --upgrade pip &&
-          python -m pip install ".[bridge,train]" "hydra-core>=1.3,<2"
+          python -m pip install ".[bridge,train]" "hydra-core>=1.3,<2" pytest
       - name: Install the exact official verl source without its distributed stack
         run: >-
           python -m pip install --no-deps
           "git+https://github.com/verl-project/verl.git@7aed6b230776f963fa09509c10d9c3a767d1102c"
+      - name: Compare forward_kl_topk values, diagnostics and gradients
+        run: python -m pytest -q tests/conformance/test_verl_v08_loss.py -m verl_conformance
       - name: Generate and export the standards smoke bundle
         run: |
           python scripts/prepare_verl_bridge_smoke.py --out _verl-smoke-source

diff --git a/PROJECT_STATE.md b/PROJECT_STATE.md
@@ -46,6 +46,24 @@ The pure-OPD integration executes two Parquet prompts through rollout, teacher
 scoring, a padded actor update, checkpoint and manifest without an environment
 or reward.
 
+## v0.8.0 single-GPU verl OPD pivot — PR C
+
+The compatibility loss is now a separate `forward_kl_topk` implementation;
+native `bucketed_topk_tail` is unchanged. The supported profile requires
+forward KL, upstream-compatible `token-mean` aggregation, temperature 1.0 and
+no sampled-token NLL mixture. Deterministic tensor tests import the official
+verl v0.8.0 loss implementation at pinned commit
+`7aed6b230776f963fa09509c10d9c3a767d1102c` and compare per-token loss,
+teacher/student top-k mass, overlap diagnostics, the reduced scalar and student
+gradients at `rtol=1e-6`, `atol=1e-7`.
+
+Teacher cache entries now bind the prompt-row digest, exact actor response
+token IDs and policy version; the cache identity additionally binds teacher,
+tokenizer, top-k, temperature and score-implementation version. A mismatched
+response or implementation fails closed. The prompt-data integration exercises
+rollout, exact-state scoring, cache reload, one token-mean optimizer update and
+the emitted verl diagnostics.
+
 ## v0.7.1 Product correction — RELEASE CANDIDATE
 
 Branch `v0.7.1-product-correction` starts from synchronized main

diff --git a/pyproject.toml b/pyproject.toml
@@ -195,6 +195,7 @@ markers = [
   "gpu: requires a CUDA GPU (deselected in CPU CI)",
   "torch: requires the [train] extra (torch/transformers/peft)",
   "network: requires network access (deselected in CI)",
+  "verl_conformance: imports the exact pinned official verl source",
   "slow: takes more than a few seconds",
 ]
 

diff --git a/src/miniverl/cache/store.py b/src/miniverl/cache/store.py
@@ -63,6 +63,23 @@
 logger = get_logger("cache")
 
 
+def _binding_checksum(
+    *,
+    prompt_row_digest: str | None,
+    actor_response_token_ids: list[int] | None,
+    policy_version: int,
+    score_implementation_version: str | None,
+) -> str:
+    payload = {
+        "actor_response_token_ids": actor_response_token_ids,
+        "policy_version": policy_version,
+        "prompt_row_digest": prompt_row_digest,
+        "score_implementation_version": score_implementation_version,
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
 def _replace_shard_file(source: Path, target: Path) -> None:
     source.replace(target)
 
@@ -161,6 +178,7 @@ def create(
         top_k: int,
         temperature: float,
         loss_mode: str,
+        score_implementation_version: str | None = None,
         dtype: str = "float32",
         entries_per_shard: int = 32,
         overwrite: bool = False,
@@ -200,6 +218,7 @@ def create(
             top_k=top_k,
             temperature=temperature,
             loss_mode=loss_mode,
+            score_implementation_version=score_implementation_version,
             dtype=dtype,
             entries_per_shard=entries_per_shard,
         )
@@ -251,6 +270,7 @@ def assert_compatible(
         top_k: int,
         temperature: float,
         loss_mode: str,
+        score_implementation_version: str | None = None,
         dtype: str,
     ) -> None:
         """Reject reuse when any objective or teacher identity component changed."""
@@ -276,6 +296,8 @@ def assert_compatible(
             "loss_mode": loss_mode,
             "dtype": dtype,
         }
+        if score_implementation_version is not None:
+            expected["score_implementation_version"] = score_implementation_version
         if self.index.schema_version >= 2:
             expected["tokenizer_identity"] = dict(tokenizer_identity or {})
             expected["teacher_adapter_provenance"] = (
@@ -344,6 +366,12 @@ def write(
         span_counts: dict[str, int] = {}
         for name in batch.span_types:
             span_counts[name] = span_counts.get(name, 0) + 1
+        binding_checksum = _binding_checksum(
+            prompt_row_digest=batch.prompt_row_digest,
+            actor_response_token_ids=batch.actor_response_token_ids,
+            policy_version=batch.policy_version,
+            score_implementation_version=self.index.score_implementation_version,
+        )
 
         self._pending[batch.trajectory_id] = {
             "tensors": tensors,
@@ -355,6 +383,9 @@ def write(
                 "tail_is_exact_zero": tail_is_exact_zero,
                 "selected_span_types": span_counts,
                 "ordered_span_types": list(batch.span_types),
+                "prompt_row_digest": batch.prompt_row_digest,
+                "actor_response_token_ids": batch.actor_response_token_ids,
+                "binding_checksum": binding_checksum,
             },
         }
         self._pending_order.append(batch.trajectory_id)
@@ -375,6 +406,9 @@ def write(
             checksum=digest.hexdigest(),
             selected_span_types=span_counts,
             ordered_span_types=list(batch.span_types),
+            prompt_row_digest=batch.prompt_row_digest,
+            actor_response_token_ids=batch.actor_response_token_ids,
+            binding_checksum=binding_checksum,
         )
 
     def _next_shard_name(self) -> str:
@@ -430,6 +464,9 @@ def flush(self) -> None:
                 checksum=meta["checksum"],
                 selected_span_types=meta["selected_span_types"],
                 ordered_span_types=meta["ordered_span_types"],
+                prompt_row_digest=meta["prompt_row_digest"],
+                actor_response_token_ids=meta["actor_response_token_ids"],
+                binding_checksum=meta["binding_checksum"],
             )
         self._write_index(next_index)
         self.index = next_index
@@ -452,6 +489,8 @@ def read(
         trajectory_id: str,
         *,
         expect_policy_version: int | None = None,
+        expect_prompt_row_digest: str | None = None,
+        expect_actor_response_token_ids: list[int] | None = None,
         device: str = "cpu",
     ) -> TeacherTargetBatch:
         """Load one trajectory's targets, enforcing the policy-version contract."""
@@ -471,6 +510,34 @@ def read(
                 hint="that would make the update off-policy. Re-score the trajectory, "
                 "or switch to run.mode=offline_kd if fixed targets are intended.",
             )
+        if (
+            expect_prompt_row_digest is not None
+            and entry.prompt_row_digest != expect_prompt_row_digest
+        ):
+            raise StaleCacheError(
+                f"teacher targets for {trajectory_id!r} have prompt-row digest "
+                f"{entry.prompt_row_digest!r}, expected {expect_prompt_row_digest!r}"
+            )
+        if (
+            expect_actor_response_token_ids is not None
+            and entry.actor_response_token_ids != expect_actor_response_token_ids
+        ):
+            raise StaleCacheError(
+                f"teacher targets for {trajectory_id!r} do not match the exact actor "
+                "response token IDs"
+            )
+        if entry.binding_checksum is not None:
+            actual_binding = _binding_checksum(
+                prompt_row_digest=entry.prompt_row_digest,
+                actor_response_token_ids=entry.actor_response_token_ids,
+                policy_version=entry.policy_version,
+                score_implementation_version=self.index.score_implementation_version,
+            )
+            if actual_binding != entry.binding_checksum:
+                raise CacheCorruptionError(
+                    f"binding checksum mismatch for {trajectory_id!r}: expected "
+                    f"{entry.binding_checksum[:16]}..., got {actual_binding[:16]}..."
+                )
         shard_path = self.path / entry.shard
         if not shard_path.is_file():
             raise CacheCorruptionError(f"shard {entry.shard} referenced by the index is missing")
@@ -516,6 +583,8 @@ def read(
             temperature=entry.temperature,
             top_k=entry.top_k,
             span_types=span_types,
+            prompt_row_digest=entry.prompt_row_digest,
+            actor_response_token_ids=entry.actor_response_token_ids,
         )
 
     def __contains__(self, trajectory_id: object) -> bool:

diff --git a/src/miniverl/config/__init__.py b/src/miniverl/config/__init__.py
@@ -14,6 +14,7 @@
     GateConfig,
     GateSignal,
     LoRAConfig,
+    LossAggregation,
     LossConfig,
     LossMode,
     MemoryConfig,
@@ -60,6 +61,7 @@
     "GateSignal",
     "LoRAConfig",
     "LossConfig",
+    "LossAggregation",
     "LossMode",
     "MemoryConfig",
     "MemoryStrategy",

diff --git a/src/miniverl/config/models.py b/src/miniverl/config/models.py
@@ -45,6 +45,7 @@
     "Quantization",
     "TeacherContextMode",
     "LossMode",
+    "LossAggregation",
     "Divergence",
     "SelectorName",
     "MemoryStrategy",
@@ -152,6 +153,14 @@ class LossMode(str, Enum):
 
     EXACT_FULL_VOCAB = "exact_full_vocab"
     BUCKETED_TOPK_TAIL = "bucketed_topk_tail"
+    VERL_FORWARD_KL_TOPK = "forward_kl_topk"
+
+
+class LossAggregation(str, Enum):
+    """How selected token losses form one optimizer-step scalar."""
+
+    NATIVE_PER_TRAJECTORY = "native_per_trajectory"
+    TOKEN_MEAN = "token-mean"
 
 
 class Divergence(str, Enum):
@@ -422,12 +431,15 @@ class LossConfig(_Base):
     """Divergence objective and its vocabulary treatment."""
 
     mode: LossMode = LossMode.BUCKETED_TOPK_TAIL
+    aggregation: LossAggregation = LossAggregation.NATIVE_PER_TRAJECTORY
     divergence: Divergence = Divergence.REVERSE_KL
     temperature: float = Field(default=1.0, gt=0.0, le=20.0)
     scale_by_temperature_squared: bool = True
     top_k: int = Field(default=64, ge=1, le=262144)
     jsd_beta: float = Field(default=0.5, ge=0.0, le=1.0)
     tail_epsilon: float = Field(default=1e-9, gt=0.0, lt=1e-2)
+    log_prob_min_clamp: float | None = Field(default=None, le=0.0)
+    loss_max_clamp: float | None = Field(default=None, gt=0.0)
     #: Number of selected prediction positions projected through the LM head at
     #: once.  Purely a memory/throughput knob -- it does not change the loss.
     chunk_size: int = Field(default=256, ge=1, le=65536)
@@ -846,6 +858,25 @@ def _validate_combination(self) -> RunConfig:
             # side effect of merely constructing a RunConfig.
             self.loss = self.loss.model_copy(update={"top_k": 1})
 
+        if self.loss.mode is LossMode.VERL_FORWARD_KL_TOPK:
+            if self.loss.divergence is not Divergence.FORWARD_KL:
+                raise ValueError("loss.mode=forward_kl_topk requires loss.divergence=forward_kl")
+            if self.loss.aggregation is not LossAggregation.TOKEN_MEAN:
+                raise ValueError(
+                    "loss.mode=forward_kl_topk requires loss.aggregation=token-mean "
+                    "for the supported verl v0.8 profile"
+                )
+            if self.loss.temperature != 1.0 or self.loss.scale_by_temperature_squared:
+                raise ValueError(
+                    "loss.mode=forward_kl_topk uses upstream logits directly; set "
+                    "temperature=1.0 and scale_by_temperature_squared=false"
+                )
+            if self.loss.sampled_token_nll_weight != 0.0:
+                raise ValueError(
+                    "loss.mode=forward_kl_topk does not mix sampled-token NLL in the "
+                    "supported verl v0.8 profile"
+                )
+
         if mode is TrainingMode.SFT and self.loss.sampled_token_nll_weight not in (0.0, 1.0):
             raise ValueError(
                 "run.mode=sft trains with oracle cross-entropy only; "

diff --git a/src/miniverl/losses/chunked.py b/src/miniverl/losses/chunked.py
@@ -45,6 +45,7 @@
     "ChunkTargetProvider",
     "ExactTargetProvider",
     "BucketedTargetProvider",
+    "VerlTopKTargetProvider",
     "LossOutput",
     "chunked_selected_position_loss",
 ]
@@ -137,6 +138,42 @@ def teacher_entropy(self, start: int, end: int) -> torch.Tensor:
             )
 
 
+@dataclass
+class VerlTopKTargetProvider:
+    """Official verl v0.8 top-k-only teacher supervision."""
+
+    topk_indices: torch.Tensor
+    topk_log_probs: torch.Tensor
+    log_prob_min_clamp: float | None = None
+    loss_max_clamp: float | None = None
+    kind: str = "verl_forward_kl_topk"
+    diagnostics: list[dict[str, torch.Tensor]] = field(default_factory=list)
+
+    def divergence(self, start: int, end: int, student_logits: torch.Tensor) -> torch.Tensor:
+        from miniverl.losses.verl_topk import verl_forward_kl_topk
+
+        output = verl_forward_kl_topk(
+            student_logits,
+            self.topk_log_probs[start:end],
+            self.topk_indices[start:end],
+            log_prob_min_clamp=self.log_prob_min_clamp,
+            loss_max_clamp=self.loss_max_clamp,
+        )
+        self.diagnostics.append(
+            {
+                "student_mass": output.student_mass.detach().to("cpu"),
+                "teacher_mass": output.teacher_mass.detach().to("cpu"),
+                "overlap_count": output.overlap_count.detach().to("cpu"),
+                "overlap_token_advantage": output.overlap_token_advantage.detach().to("cpu"),
+            }
+        )
+        return output.loss
+
+    def teacher_entropy(self, start: int, end: int) -> torch.Tensor:
+        """Entropy is undefined without the omitted tail distribution."""
+        return torch.full((end - start,), float("nan"), dtype=torch.float32)
+
+
 @dataclass
 class LossOutput:
     """Result of one chunked objective evaluation."""