DaoyuanLi2816 · DaoyuanLi2816 · Aug 12, 2026 · Aug 12, 2026 · Aug 12, 2026 · Aug 12, 2026
diff --git a/PROJECT_STATE.md b/PROJECT_STATE.md
@@ -23,6 +23,29 @@ distributed dimensions fail closed. `bridge compile-opd` is deliberately an
 offline config compiler: it loads no weights and does not claim the prompt
 runtime or verl engine equivalence implemented by later PRs.
 
+## v0.8.0 single-GPU verl OPD pivot — PR B
+
+The trainer now has a discriminated `environment` / `verl_parquet` source
+contract. Parquet rows stream by record batch, reject every invalid row, retain
+the supported verl metadata and record schema, content, row, tokenizer and
+rendered-prompt digests. Pure prompt OPD needs neither a `ToolEnvironment` nor
+a reward model; task rewards, oracle SFT and privileged teacher context fail
+closed instead of being ignored.
+
+`PromptDatasetRolloutRuntime` applies the actor chat template once, performs
+real masked padded greedy generation, restores logical row order and creates
+trajectories whose selected model spans contain response tokens only. A
+bounded padded-token budget is enforced before allocation. CUDA OOM retry may
+split physical batches but never changes logical batch size, seeds or training
+configuration. The existing multi-turn environment runner is retained behind
+`ToolEnvironmentRolloutRuntime` and its prior integration suite remains green.
+
+Focused validation on 2026-08-11: 17 new config/data/runtime/end-to-end tests,
+94 existing trainer/toy-pipeline tests, Ruff, and mypy over 126 source files.
+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.7.1 Product correction — RELEASE CANDIDATE
 
 Branch `v0.7.1-product-correction` starts from synchronized main

diff --git a/pyproject.toml b/pyproject.toml
@@ -56,6 +56,7 @@ train = [
   "peft>=0.12",
   "accelerate>=0.33",
   "numpy>=1.24",
+  "pyarrow>=15",
 ]
 cuda = ["bitsandbytes>=0.43; platform_system != 'Darwin'"]
 dpo = [

diff --git a/src/miniverl/cli.py b/src/miniverl/cli.py
@@ -241,18 +241,20 @@ def validate(
         return
 
     warnings: list[str] = []
-    try:
-        environment = make_environment(config.environment.name, **config.environment.params)
-        if config.models.teacher.mode.value == "privileged_context" and not hasattr(
-            environment, "privileged_context"
-        ):
-            warnings.append("environment provides no privileged context")
-    except MiniVerlError as exc:
-        if as_json:
-            _emit_json({"valid": False, "path": str(recipe), "errors": [exc.message]})
-            raise typer.Exit(1) from None
-        _fail(exc)
-        return
+    environment_config = config.environment
+    if environment_config is not None:
+        try:
+            environment = make_environment(environment_config.name, **environment_config.params)
+            if config.models.teacher.mode.value == "privileged_context" and not hasattr(
+                environment, "privileged_context"
+            ):
+                warnings.append("environment provides no privileged context")
+        except MiniVerlError as exc:
+            if as_json:
+                _emit_json({"valid": False, "path": str(recipe), "errors": [exc.message]})
+                raise typer.Exit(1) from None
+            _fail(exc)
+            return
 
     steps_per_cycle = max(
         1,
@@ -281,8 +283,9 @@ def validate(
         "backend": config.models.backend.value,
         "student": config.models.student.model_id,
         "teacher": config.models.teacher.model_id,
-        "environment": config.environment.name,
-        "difficulty": config.environment.difficulty,
+        "source_kind": config.source.kind.value,
+        "environment": environment_config.name if environment_config is not None else None,
+        "difficulty": environment_config.difficulty if environment_config is not None else None,
         "objective": (
             "sft_cross_entropy"
             if config.run.mode.value == "sft"
@@ -302,7 +305,9 @@ def validate(
         "optimizer_steps_per_cycle": steps_per_cycle,
         "planned_optimizer_steps": steps_per_cycle
         * (config.train.cycles + config.train.sft_warmup_cycles),
-        "eval_tasks": config.effective_eval_tasks,
+        "eval_tasks": (
+            config.effective_eval_tasks if environment_config is not None else config.eval.tasks
+        ),
         "seed": config.run.seed,
         "warnings": warnings,
     }
@@ -594,7 +599,7 @@ def qualify_teacher_command(
 
     try:
         config = RunConfig.from_yaml(recipe)
-        if config.environment.name != "sqlite_recovery":
+        if config.require_environment("qualify-teacher").name != "sqlite_recovery":
             raise ConfigError("qualify-teacher requires environment.name=sqlite_recovery")
         _require_training_stack("miniverl qualify-teacher")
         from miniverl.evaluation.teacher_gate import evaluate_teacher_candidate

diff --git a/src/miniverl/config/__init__.py b/src/miniverl/config/__init__.py
@@ -9,6 +9,7 @@
     CacheConfig,
     Divergence,
     EnvironmentConfig,
+    EnvironmentSourceConfig,
     EvalConfig,
     GateConfig,
     GateSignal,
@@ -25,6 +26,7 @@
     OPDFreshness,
     OptimizerName,
     Precision,
+    PromptTruncation,
     Quantization,
     ReferenceModelConfig,
     ReportConfig,
@@ -33,6 +35,7 @@
     RunMeta,
     SelectionConfig,
     SelectorName,
+    SourceKind,
     StudentModelConfig,
     TeacherAdapterConfig,
     TeacherContextMode,
@@ -41,6 +44,7 @@
     ToyModelConfig,
     TrainConfig,
     TrainingMode,
+    VerlParquetSourceConfig,
 )
 
 __all__ = [
@@ -50,6 +54,7 @@
     "CacheConfig",
     "Divergence",
     "EnvironmentConfig",
+    "EnvironmentSourceConfig",
     "EvalConfig",
     "GateConfig",
     "GateSignal",
@@ -66,6 +71,7 @@
     "OPDFreshness",
     "OptimizerName",
     "Precision",
+    "PromptTruncation",
     "Quantization",
     "ReferenceModelConfig",
     "ReportConfig",
@@ -75,11 +81,13 @@
     "SelectionConfig",
     "SelectorName",
     "StudentModelConfig",
+    "SourceKind",
     "TeacherContextMode",
     "TeacherMode",
     "TeacherAdapterConfig",
     "TeacherModelConfig",
     "ToyModelConfig",
     "TrainConfig",
     "TrainingMode",
+    "VerlParquetSourceConfig",
 ]
diff --git a/src/miniverl/config/models.py b/src/miniverl/config/models.py
@@ -14,7 +14,7 @@
 
 from enum import Enum
 from pathlib import Path
-from typing import Any, Literal
+from typing import Annotated, Any, Literal
 
 import yaml
 from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, field_validator, model_validator
@@ -60,6 +60,10 @@
     "LossConfig",
     "SelectionConfig",
     "RolloutConfig",
+    "SourceKind",
+    "PromptTruncation",
+    "EnvironmentSourceConfig",
+    "VerlParquetSourceConfig",
     "EnvironmentConfig",
     "TrainConfig",
     "MemoryConfig",
@@ -491,6 +495,45 @@ 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)
+    prompt_batch_size: int = Field(default=1, ge=1, le=1024)
+    max_padded_tokens: int = Field(default=4096, ge=16, le=1048576)
+
+
+class SourceKind(str, Enum):
+    """Where rollout inputs originate."""
+
+    ENVIRONMENT = "environment"
+    VERL_PARQUET = "verl_parquet"
+
+
+class PromptTruncation(str, Enum):
+    """Explicit overlength handling for prompt datasets."""
+
+    ERROR = "error"
+    LEFT = "left"
+    RIGHT = "right"
+
+
+class EnvironmentSourceConfig(_Base):
+    """Discriminator for the backward-compatible registered-environment path."""
+
+    kind: Literal[SourceKind.ENVIRONMENT] = SourceKind.ENVIRONMENT
+
+
+class VerlParquetSourceConfig(_Base):
+    """Bounded, deterministic verl-style Parquet prompt input."""
+
+    kind: Literal[SourceKind.VERL_PARQUET] = SourceKind.VERL_PARQUET
+    train_files: list[str] = Field(min_length=1)
+    val_files: list[str] = Field(default_factory=list)
+    prompt_key: str = Field(default="prompt", min_length=1)
+    allow_plain_string_prompts: bool = False
+    use_task_rewards: bool = False
+    max_prompt_length: int = Field(default=512, ge=1, le=131072)
+    truncation: PromptTruncation = PromptTruncation.ERROR
+    shuffle: bool = True
+    seed: int = Field(default=0, ge=0)
+    row_batch_size: int = Field(default=256, ge=1, le=65536)
 
 
 class EnvironmentConfig(_Base):
@@ -681,7 +724,11 @@ class RunConfig(_Base):
     schema_version: int = CONFIG_SCHEMA_VERSION
     run: RunMeta = Field(default_factory=RunMeta)
     models: ModelsConfig
-    environment: EnvironmentConfig
+    source: Annotated[
+        EnvironmentSourceConfig | VerlParquetSourceConfig,
+        Field(discriminator="kind"),
+    ] = Field(default_factory=EnvironmentSourceConfig)
+    environment: EnvironmentConfig | None = None
     rollout: RolloutConfig = Field(default_factory=RolloutConfig)
     selection: SelectionConfig = Field(default_factory=SelectionConfig)
     loss: LossConfig = Field(default_factory=LossConfig)
@@ -725,6 +772,30 @@ 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.VERL_PARQUET and self.environment is not None:
+            raise ValueError("source.kind=verl_parquet must not define environment")
+        if self.source.kind is SourceKind.VERL_PARQUET:
+            if mode is not TrainingMode.OPD:
+                raise ValueError(
+                    "source.kind=verl_parquet supports pure OPD in v0.8; it has no oracle "
+                    "labels for sft or offline_kd"
+                )
+            if self.train.sft_warmup_cycles:
+                raise ValueError(
+                    "source.kind=verl_parquet cannot run sft_warmup_cycles without oracle labels"
+                )
+            if self.models.teacher.mode is not TeacherContextMode.STANDARD:
+                raise ValueError(
+                    "source.kind=verl_parquet requires the actor and teacher to score the "
+                    "same rendered prompt; privileged_context is outside the v0.8 profile"
+                )
+            if self.source.use_task_rewards:
+                raise ValueError(
+                    "source.use_task_rewards=true is not implemented for the pure OPD prompt "
+                    "runtime; refusing to ignore row reward_model metadata"
+                )
         if mode is TrainingMode.OPD and self.cache.reuse_across_policy_versions:
             raise ValueError(
                 "cache.reuse_across_policy_versions=true contradicts run.mode=opd: "
@@ -829,7 +900,7 @@ def _validate_combination(self) -> RunConfig:
 
         alignment = self.alignment
 
-        if self.eval.enabled:
+        if self.eval.enabled and self.environment is not None:
             split_sizes = {
                 "train": self.environment.train_tasks,
                 "eval": self.environment.eval_tasks,
@@ -917,8 +988,22 @@ def effective_eval_tasks(self) -> int:
         """Number of evaluation tasks after applying the eval override."""
         if self.eval.tasks is not None:
             return self.eval.tasks
+        if self.environment is None:
+            raise ConfigError(
+                "eval.tasks is required when source.kind=verl_parquet",
+                hint="set eval.tasks to a bounded number no larger than the validation rows",
+            )
         return self.environment.eval_tasks
 
+    def require_environment(self, operation: str = "this operation") -> EnvironmentConfig:
+        """Return the registered environment or fail with a source-aware message."""
+        if self.environment is None:
+            raise ConfigError(
+                f"{operation} requires source.kind=environment",
+                hint="the verl_parquet source uses prompt rollout and has no ToolEnvironment",
+            )
+        return self.environment
+
     @classmethod
     def from_yaml(cls, path: str | Path) -> RunConfig:
         """Load and validate a recipe from a YAML file."""
@@ -988,6 +1073,20 @@ def resolved_for_runtime(self) -> RunConfig:
                 source = private.get("_source_path") if isinstance(private, dict) else None
                 base = source.parent if isinstance(source, Path) else Path.cwd()
                 runtime.offline_kd.dataset_path = str((base / path).resolve())
+        if isinstance(runtime.source, VerlParquetSourceConfig):
+            private = getattr(self, "__pydantic_private__", None)
+            source_path = private.get("_source_path") if isinstance(private, dict) else None
+            base = source_path.parent if isinstance(source_path, Path) else Path.cwd()
+
+            def resolve_files(paths: list[str]) -> list[str]:
+                return [
+                    str(path if path.is_absolute() else (base / path).resolve())
+                    for raw in paths
+                    for path in [Path(raw)]
+                ]
+
+            runtime.source.train_files = resolve_files(runtime.source.train_files)
+            runtime.source.val_files = resolve_files(runtime.source.val_files)
         return runtime
 
     def to_yaml(self) -> str:

diff --git a/src/miniverl/data/__init__.py b/src/miniverl/data/__init__.py
@@ -0,0 +1,17 @@
+"""Typed, source-agnostic training data inputs."""
+
+from miniverl.data.verl_parquet import (
+    PromptDatasetManifest,
+    PromptRecord,
+    RenderedPrompt,
+    VerlParquetDataset,
+    render_prompt,
+)
+
+__all__ = [
+    "PromptDatasetManifest",
+    "PromptRecord",
+    "RenderedPrompt",
+    "VerlParquetDataset",
+    "render_prompt",
+]