DaoyuanLi2816 · Aug 29, 2026 · Aug 29, 2026 · Aug 29, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -15,6 +15,16 @@ All notable changes to miniVERL are recorded here. The format follows
 - Froze the 24-cell pre-v0.11 `hf_reference` actor-rollout baseline on the WSL2
   RTX 4080, with exact wheel, weight, policy and raw-measurement identity.
 
+### Typed rollout core
+
+- Added a torch-free Rollout Runtime v2 contract for requests, grouped sample
+  identity, policy snapshots, synchronization, lifecycle and backend metrics.
+- Preserved `hf_reference` for existing recipes and added opt-in `hf_cached`
+  generation with one padded prefill, incremental KV-cache decode, per-row RNG,
+  sampled-token log-probabilities, stop provenance and OOM-safe repartitioning.
+- Bound prompt-generation manifests and every v2 request to the live actor
+  parameter version, base/tokenizer/adapter identity, backend and execution plan.
+
 ### Documentation narrative
 
 - Rebuilt the English, Chinese and PyPI landing narratives around the local

diff --git a/docs/rollout-runtime-v2.md b/docs/rollout-runtime-v2.md
@@ -0,0 +1,40 @@
+# Rollout Runtime v2
+
+Rollout Runtime v2 gives prompt-based OPD a typed boundary between the live
+actor and generation. Existing recipes continue to use `hf_reference`. A new
+recipe can opt into the local cached path:
+
+```yaml
+rollout:
+  backend: hf_cached
+  samples_per_prompt: 1
+  prompt_batch_size: 4
+  max_padded_tokens: 4096
+  synchronization: strict
+  compile_backend: true  # measured WSL2/CUDA fast path; first run compiles
+  record_logprobs: true
+```
+
+`hf_cached` performs one padded prefill for each physical batch, then advances
+the model with one token per active row and the returned KV cache. Each logical
+sample owns its CPU generator, so OOM bisection and physical repartitioning do
+not enter the seed derivation. Greedy and stochastic requests retain EOS, text
+stop, maximum-token and sampled-token-log-probability provenance.
+
+`compile_backend: true` uses a generation-only Inductor decoder with CUDA
+graphs disabled. It has a substantial first-run compilation cost, so it is
+explicit rather than a legacy default. Set it to `false` for the eager cached
+path or on hosts without a working CUDA compiler toolchain.
+
+## Policy binding
+
+Before generation, the runtime synchronizes a `PolicySnapshot` that binds the
+parameter version, model revision, tokenizer structure, adapter manifest and
+live trainable-tensor digests, precision, quantization, backend version,
+profile identity and execution plan. A request for any other identity fails
+before model execution. Lifecycle transitions are explicit: `new` →
+`synchronized` → `quiesced` or `closed`.
+
+The v0.11 baseline result remains the pre-change `hf_reference` measurement.
+It is not evidence for `hf_cached` performance. Grouped `n>1` publication and
+external-engine support are separate release-chain stages.
diff --git a/mkdocs.yml b/mkdocs.yml
@@ -72,6 +72,7 @@ nav:
       - Legacy environment/PPO bridge: legacy-verl-bridge.md
   - Community benchmarks: community-benchmarks.md
   - Rollout Runtime v2:
+      - Runtime contract: rollout-runtime-v2.md
       - Competitive landscape: competitive-landscape.md
       - Benchmark method: benchmarks/rollout-runtime-v2-method.md
       - Frozen HF baseline: benchmarks/rollout-runtime-v2.md

diff --git a/src/miniverl/config/models.py b/src/miniverl/config/models.py
@@ -27,6 +27,7 @@
     TeacherMode,
 )
 from miniverl.errors import ConfigError
+from miniverl.runtime.generation import RolloutBackendKind
 from miniverl.utils.runs import write_text
 
 __all__ = [
@@ -62,6 +63,7 @@
     "LossConfig",
     "SelectionConfig",
     "RolloutConfig",
+    "RolloutEngineConfig",
     "SourceKind",
     "PromptTruncation",
     "EnvironmentSourceConfig",
@@ -523,6 +525,16 @@ def _gate_matches_selector(self) -> SelectionConfig:
         return self
 
 
+class RolloutEngineConfig(_Base):
+    """Lifecycle and resource bounds for an optional managed generation engine."""
+
+    managed: bool = True
+    host: str = Field(default="127.0.0.1", min_length=1)
+    startup_timeout_seconds: float = Field(default=120.0, gt=0.0, le=3600.0)
+    request_timeout_seconds: float = Field(default=120.0, gt=0.0, le=3600.0)
+    memory_fraction: float = Field(default=0.7, gt=0.0, le=1.0)
+
+
 class RolloutConfig(_Base):
     """Bounds and sampling parameters for student rollouts."""
 
@@ -540,8 +552,15 @@ class RolloutConfig(_Base):
     # first parse error.
     max_parse_errors: int = Field(default=2, ge=0, le=32)
     max_repeated_calls: int = Field(default=2, ge=1, le=32)
+    #: The historical path remains the default for old recipes. New v2 recipes
+    #: select ``hf_cached`` explicitly.
+    backend: RolloutBackendKind = RolloutBackendKind.HF_REFERENCE
+    samples_per_prompt: int = Field(default=1, ge=1, le=128)
     prompt_batch_size: int = Field(default=1, ge=1, le=1024)
     max_padded_tokens: int = Field(default=4096, ge=16, le=1048576)
+    synchronization: Literal["strict"] = "strict"
+    compile_backend: bool = False
+    engine: RolloutEngineConfig = Field(default_factory=RolloutEngineConfig)
 
 
 class SourceKind(str, Enum):
@@ -826,8 +845,21 @@ def _validate_combination(self) -> RunConfig:
         mode = self.run.mode
         if self.source.kind is SourceKind.ENVIRONMENT and self.environment is None:
             raise ValueError("source.kind=environment requires an environment configuration")
+        if (
+            self.source.kind is SourceKind.ENVIRONMENT
+            and self.rollout.backend is not RolloutBackendKind.HF_REFERENCE
+        ):
+            raise ValueError(
+                "rollout.backend=hf_cached currently requires source.kind=verl_parquet; "
+                "tool-environment episodes retain the hf_reference path"
+            )
         if self.source.kind is SourceKind.VERL_PARQUET and self.environment is not None:
             raise ValueError("source.kind=verl_parquet must not define environment")
+        if self.rollout.samples_per_prompt != 1:
+            raise ValueError(
+                "rollout.samples_per_prompt > 1 requires the transactional grouped-rollout "
+                "runtime, which is not enabled in this build"
+            )
         if self.source.kind is SourceKind.VERL_PARQUET:
             if mode is not TrainingMode.OPD:
                 raise ValueError(

diff --git a/src/miniverl/models/hf.py b/src/miniverl/models/hf.py
@@ -29,7 +29,11 @@
 from miniverl.errors import BackendError, MissingDependencyError
 from miniverl.models.adapters import ArchitectureAdapter
 from miniverl.models.base import BackendCapabilities, CausalLMBackend, GenerationOutput
-from miniverl.models.sampling import run_generation, run_greedy_padded_generation
+from miniverl.models.sampling import (
+    run_cached_padded_generation,
+    run_generation,
+    run_greedy_padded_generation,
+)
 from miniverl.utils.lazy import have_module, require_peft, require_transformers
 
 __all__ = ["HFBackend", "resolve_dtype", "supports_bfloat16"]
@@ -146,6 +150,7 @@ def __init__(
         self._dtype = dtype
         self.adapter_provenance = adapter_provenance
         self.adapter = ArchitectureAdapter.resolve(model)
+        self._cached_generation_backbone: Any | None = None
         num_params = sum(p.numel() for p in model.parameters())
         num_trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
         self.capabilities = BackendCapabilities(
@@ -392,19 +397,52 @@ def _backbone_forward(
         attention_mask: torch.Tensor | None = None,
         past_key_values: Any = None,
         use_cache: bool = False,
+        position_ids: torch.Tensor | None = None,
+        cache_position: torch.Tensor | None = None,
+        cached_generation: bool = False,
     ) -> tuple[torch.Tensor, Any]:
-        backbone = self.adapter.backbone
+        backbone = (
+            self._cached_generation_backbone
+            if cached_generation and self._cached_generation_backbone is not None
+            else self.adapter.backbone
+        )
         kwargs: dict[str, Any] = {"input_ids": input_ids, "use_cache": use_cache}
         if attention_mask is not None:
             kwargs["attention_mask"] = attention_mask
         if past_key_values is not None:
             kwargs["past_key_values"] = past_key_values
+        if position_ids is not None:
+10BC8
            kwargs["position_ids"] = position_ids
+        if cache_position is not None:
+            kwargs["cache_position"] = cache_position
         outputs = backbone(**kwargs)
         hidden = getattr(outputs, "last_hidden_state", None)
         if hidden is None:
             hidden = outputs[0]
         return hidden, getattr(outputs, "past_key_values", None)
 
+    def enable_cached_generation_compilation(self) -> None:
+        """Compile only the cached decode backbone, with CUDA graphs disabled."""
+
+        if self._cached_generation_backbone is not None:
+            return
+        if not self._device.startswith("cuda"):
+            raise BackendError(
+                "compiled hf_cached generation requires CUDA",
+                hint="set rollout.compile_backend=false on CPU",
+            )
+        try:
+            self._cached_generation_backbone = torch.compile(
+                self.adapter.backbone,
+                dynamic=True,
+                options={"triton.cudagraphs": False},
+            )
+        except Exception as exc:
+            raise BackendError(
+                "could not initialize compiled hf_cached generation",
+                hint="set rollout.compile_backend=false to use the eager KV-cache path",
+            ) from exc
+
     # -- generation ---------------------------------------------------------
 
     def generate(
@@ -515,6 +553,128 @@ def step(rows: list[list[int]]) -> torch.Tensor:
             if was_training:
                 self.model.train()
 
+    def generate_batch_cached(
+        self,
+        prefix_token_ids: Sequence[Sequence[int]],
+        *,
+        max_new_tokens: int,
+        stop_sequences: Sequence[str] = (),
+        temperature: float = 0.0,
+        top_p: float = 1.0,
+        top_k: int = 0,
+        seeds: Sequence[int | None] | None = None,
+        record_logprobs: bool = False,
+    ) -> list[GenerationOutput]:
+        """Decode a physical batch with one prefill and an incremental KV cache."""
+
+        if not prefix_token_ids:
+            return []
+        if any(not row for row in prefix_token_ids):
+            raise ValueError("cached generation cannot contain an empty prompt")
+        chosen_seeds = list(seeds or [0] * len(prefix_token_ids))
+        if len(chosen_seeds) != len(prefix_token_ids):
+            raise ValueError("generate_batch_cached needs exactly one seed per prompt")
+        generators = []
+        for seed in chosen_seeds:
+            generator = torch.Generator(device="cpu")
+            generator.manual_seed(int(seed or 0))
+            generators.append(generator)
+
+        was_training = self.model.training
+        self.model.eval()
+        cache_flag = getattr(self.model.config, "use_cache", True)
+        self.model.config.use_cache = True
+        attention_mask: torch.Tensor | None = None
+
+        def prefill(rows: list[list[int]]) -> tuple[torch.Tensor, Any]:
+            nonlocal attention_mask
+            lengths = [len(row) for row in rows]
+            width = max(lengths)
+            input_ids = torch.full(
+                (len(rows), width),
+                self.tokenizer.pad_token_id,
+                dtype=torch.long,
+                device=self._device,
+            )
+            attention_mask = torch.zeros((len(rows), width), dtype=torch.bool, device=self._device)
+            for index, row in enumerate(rows):
+                input_ids[index, : len(row)] = torch.tensor(
+                    row, dtype=torch.long, device=self._device
+                )
+                attention_mask[index, : len(row)] = True
+            position_ids = attention_mask.long().cumsum(dim=-1) - 1
+            position_ids.masked_fill_(~attention_mask, 0)
+            with torch.no_grad():
+                hidden, present = self._backbone_forward(
+                    input_ids,
+                    attention_mask=attention_mask,
+                    use_cache=True,
+                    position_ids=position_ids,
+                    cached_generation=True,
+                )
+                positions = torch.tensor(
+                    [length - 1 for length in lengths], dtype=torch.long, device=self._device
+                )
+                selected = hidden[torch.arange(len(rows), device=self._device), positions]
+                logits = self.adapter.lm_head(selected)
+            return logits, present
+
+        def decode_step(
+            tokens: list[int], active: list[bool], state: Any
+        ) -> tuple[torch.Tensor, Any]:
+            nonlocal attention_mask
+            assert attention_mask is not None
+            active_tensor = torch.tensor(active, dtype=torch.bool, device=self._device)
+            input_ids = torch.tensor(tokens, dtype=torch.long, device=self._device).unsqueeze(1)
+            input_ids = torch.where(
+                active_tensor.unsqueeze(1),
+                input_ids,
+                torch.full_like(input_ids, self.tokenizer.pad_token_id),
+            )
+            position_ids = attention_mask.long().sum(dim=-1, keepdim=True)
+            attention_mask = torch.cat((attention_mask, active_tensor.unsqueeze(1)), dim=1)
+            with torch.no_grad():
+                hidden, present = self._backbone_forward(
+                    input_ids,
+                    attention_mask=attention_mask,
+                    past_key_values=state,
+                    use_cache=True,
+                    position_ids=position_ids,
+                    cached_generation=True,
+                )
+                logits = self.adapter.lm_head(hidden[:, -1, :])
+            return logits, present
+
+        try:
+            try:
+                return run_cached_padded_generation(
+                    prefill=prefill,
+                    decode_step=decode_step,
+                    prefix_token_ids=prefix_token_ids,
+                    decode=self.tokenizer.decode,
+                    eos_token_id=self.tokenizer.eos_token_id,
+                    max_new_tokens=max_new_tokens,
+                    stop_sequences=stop_sequences,
+                    temperature=temperature,
+                    top_p=top_p,
+                    top_k=top_k,
+                    generators=generators,
+                    record_logprobs=record_logprobs,
+                )
+            except torch.OutOfMemoryError:
+                raise
+            except Exception as exc:
+                if self._cached_generation_backbone is None:
+                    raise
+                raise BackendError(
+                    "compiled hf_cached generation failed",
+                    hint="set rollout.compile_backend=false to use the eager KV-cache path",
+                ) from exc
+        finally:
+            self.model.config.use_cache = cache_flag
+            if was_training:
+                self.model.train()
+
     # -- scoring ------------------------------------------------------------
 
     def hidden_states_at(