"""miniVERL command line interface.

A thin layer: every command parses arguments, calls one library function, and

renders the result. No training logic lives here.

Only the heavy commands import torch, and they do it inside the command body,

so ``miniverl --help``, ``doctor``, ``validate``, ``inspect``, ``report`` and

``cache`` all work from a bare ``pip install miniverl``.

"""

from __future__ import annotations

import hashlib

import json

import sys

import time

from pathlib import Path

from typing import Any, Literal, Optional

import typer

from pydantic import ValidationError

from rich.console import Console

from rich.markup import escape

from rich.table import Table

from miniverl import __version__

from miniverl.errors import ConfigError, MiniVerlError

from miniverl.utils.runs import make_run_id

app = typer.Typer(

name="miniverl",

help=(

"Run a documented subset of verl-style OPD on one consumer GPU.\n\n"

"Compile typed verl-shaped config, reuse Parquet prompts, execute actor rollout, "

"teacher scoring and actor update locally, then export inspectable PEFT artifacts."

),

add_completion=False,

no_args_is_help=True,

rich_markup_mode=None,

)

cache_app = typer.Typer(help="Inspect and validate a teacher-target cache.", no_args_is_help=True)

app.add_typer(cache_app, name="cache")

bridge_app = typer.Typer(

help="Inspect a pinned, exported verl scale-out bundle.", no_args_is_help=True

)

app.add_typer(bridge_app, name="bridge")

alignment_suite_app = typer.Typer(

help=(

"Prepare, validate and report the pinned external alignment suite. "

"Needs the alignment-benchmarks extra."

),

no_args_is_help=True,

)

app.add_typer(alignment_suite_app, name="alignment-suite")

evidence_app = typer.Typer(

help="Show and validate evidence packaged with the installed wheel.", no_args_is_help=True

)

app.add_typer(evidence_app, name="evidence")

data_app = typer.Typer(help="Create and inspect portable prompt data.", no_args_is_help=True)

app.add_typer(data_app, name="data")

console = Console()

err_console = Console(stderr=True)

_STATUS_STYLE = {"ok": "green", "warn": "yellow", "missing": "yellow", "fail": "red"}

def _emit_json(payload: Any) -> None:

try:

serialized = json.dumps(payload, default=str, allow_nan=False)

except (OverflowError, RecursionError, TypeError, ValueError) as exc:

from miniverl.errors import SerializationError

_fail(SerializationError(f"command result is not finite JSON: {exc}"))

return

console.print_json(serialized)

def _esc(value: object) -> str:

"""Escape dynamic text before it reaches Rich.

Rich treats square brackets as markup, so unescaped values silently lose

content: the hint ``pip install "miniverl[train]"`` would print as

``pip install "miniverl"`` -- the wrong command. Every dynamic string this

module prints goes through here.

"""

return escape(str(value))

def _require_training_stack(purpose: str) -> None:

"""Fail with an install command if the training extra is not present.

Called before any heavy import so a bare ``pip install miniverl`` produces

the exact command to run rather than a bare ``ModuleNotFoundError``.

"""

from miniverl.errors import MissingDependencyError

from miniverl.utils.lazy import have_module

for module in ("torch", "transformers", "peft"):

if not have_module(module):

raise MissingDependencyError(module, "train", purpose)

def _fail(exc: Exception, *, code: int = 1) -> None:

"""Print an actionable error and exit non-zero."""

if isinstance(exc, ModuleNotFoundError):

from miniverl.errors import MissingDependencyError

exc = MissingDependencyError(exc.name or "a dependency", "train", "This command")

if isinstance(exc, MiniVerlError):

err_console.print(f"[red]error[/red] {_esc(exc.message)}")

if exc.hint:

err_console.print(f"[yellow]hint[/yellow] {_esc(exc.hint)}")

else:

err_console.print(f"[red]error[/red] {_esc(exc)}")

raise typer.Exit(code)

def _version_callback(value: bool) -> None:

if value:

console.print(f"miniverl {__version__}")

raise typer.Exit(0)

@app.callback()

def main(

version: bool = typer.Option(

False,

"--version",

"-V",

help="Print the miniVERL version and exit.",

callback=_version_callback,

is_eager=True,

),

log_level: str = typer.Option(

"INFO",

"--log-level",

help="Logging verbosity (DEBUG, INFO, WARNING, ERROR).",

envvar="MINIVERL_LOG_LEVEL",

),

) -> None:

"""miniVERL: the bounded local runtime for verl-style OPD on one GPU."""

from miniverl.utils.logging import configure_logging

configure_logging(log_level)

@data_app.command("sample")

def data_sample_command(

out: Path = typer.Option(..., "--out", help="Output Parquet path."),

format_name: str = typer.Option(

"verl-parquet", "--format", help="Portable output format (verl-parquet only)."

),

rows: int = typer.Option(4, "--rows", min=1, max=1024, help="Number of sample prompts."),

) -> None:

"""Create a small reward-free verl-style Parquet prompt dataset."""

if format_name != "verl-parquet":

_fail(ConfigError("--format must be verl-parquet"))

return

try:

import pyarrow as pa

import pyarrow.parquet as pq

except ModuleNotFoundError as exc:

from miniverl.errors import MissingDependencyError

_fail(MissingDependencyError(exc.name or "pyarrow", "bridge", "Parquet sample data"))

return

prompts = [

"Explain why exact provenance matters in one concise sentence.",

"Give one safe way to recover from a CUDA out-of-memory error.",

"What does token-mean loss aggregation mean?",

"State one limitation of a single-GPU training runtime.",

]

records = []

for index in range(rows):

prompt = prompts[index % len(prompts)]

if index >= len(prompts):

prompt = f"Sample {index + 1}: {prompt}"

records.append(

{

"prompt": [

{"role": "system", "content": "Answer clearly and briefly."},

{"role": "user", "content": prompt},

],

"data_source": "miniverl_quickstart",

"ability": "short_answer",

"extra_info": {"sample_index": index},

}

)

out.parent.mkdir(parents=True, exist_ok=True)

pq.write_table(pa.Table.from_pylist(records), out)

digest = hashlib.sha256(out.read_bytes()).hexdigest()

console.print(f"[green]wrote[/green] {_esc(out)} ({rows} rows, sha256 {digest})")

# ---------------------------------------------------------------- doctor

@app.command()

def doctor(

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

output_dir: Path = typer.Option(

Path("runs"), "--output", help="Directory to test for writability."

),

) -> None:

"""Report what this machine can run, and what to install for the rest."""

from miniverl.doctor import run_doctor

try:

report = run_doctor(output_dir)

except Exception as exc:

_fail(exc)

return

if as_json:

_emit_json(report.to_dict())

return

table = Table(title=f"miniVERL {report.miniverl_version} environment", show_lines=False)

table.add_column("check", style="bold")

table.add_column("status")

table.add_column("detail")

for check in report.checks:

table.add_row(

_esc(check.name),

f"[{_STATUS_STYLE.get(check.status, 'white')}]{check.status}[/]",

_esc(check.detail),

)

console.print(table)

verdict = report.to_dict()["verdict"]

console.print()

for label, key, command in (

("core commands (doctor/validate/inspect/report/cache)", "core_commands", None),

(

"CPU + toy training (demo, recipes/toy_cpu.yaml)",

"cpu_training",

'pip install "miniverl[train]"',

),

(

"single-GPU CUDA training (native recipes)",

"gpu_training",

"install a CUDA build of torch",

),

("4-bit QLoRA", "qlora_4bit", 'pip install "miniverl[train,cuda]"'),

):

ready = verdict[key]

# Pad to the width of the longer word so the labels line up in a column.

mark = "[green]yes[/green]" if ready else "[yellow]no [/yellow]"

suffix = "" if ready or not command else f" -> {command}"

console.print(f" {mark} {_esc(label)}{_esc(suffix)}")

hints = [c for c in report.checks if c.hint and c.status in {"fail", "missing", "warn"}]

if hints:

console.print()

console.print("[bold]suggestions[/bold]")

for check in hints:

console.print(f" - {_esc(check.name)}: {_esc(check.hint)}")

# -------------------------------------------------------------- validate

@app.command()

def validate(

recipe: Path = typer.Argument(..., help="Path to a recipe YAML file."),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

) -> None:

"""Validate a recipe without downloading models or allocating memory."""

from miniverl.config import RunConfig

from miniverl.environments.registry import make_environment

try:

config = RunConfig.from_yaml(recipe)

except ValidationError as exc:

if as_json:

_emit_json({"valid": False, "path": str(recipe), "errors": exc.errors()})

raise typer.Exit(1) from None

err_console.print(f"[red]invalid recipe[/red] {_esc(recipe)}")

for error in exc.errors():

location = ".".join(str(p) for p in error["loc"])

err_console.print(f" [red]{_esc(location or '<root>')}[/red]: {_esc(error['msg'])}")

err_console.print(

"\n[yellow]hint[/yellow] compare against recipes/toy_cpu.yaml, which is "

"validated in CI"

)

raise typer.Exit(1) from None

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

warnings: list[str] = []

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,

(config.train.rollouts_per_cycle + config.train.gradient_accumulation_steps - 1)

// config.train.gradient_accumulation_steps,

)

if config.run.mode.value == "opd" and config.train.opd_freshness.value == "replay":

warnings.append(

f"opd_freshness=replay permits {steps_per_cycle} optimizer step(s) per "

"rollout batch; this is online distillation with replay, not genuine OPD"

)

if config.models.backend.value == "hf" and not config.models.student.revision:

warnings.append("models.student.revision is unpinned; the manifest will record 'unpinned'")

if config.models.backend.value == "hf" and not config.models.teacher.revision:

warnings.append("models.teacher.revision is unpinned")

payload = {

"valid": True,

"path": str(recipe),

"run_name": config.run.name,

"mode": config.run.mode.value,

"is_on_policy": config.is_on_policy,

"opd_freshness": (

config.train.opd_freshness.value if config.run.mode.value == "opd" else None

),

"backend": config.models.backend.value,

"student": config.models.student.model_id,

"teacher": config.models.teacher.model_id,

"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"

else (

"online_distillation_with_replay"

if config.run.mode.value == "opd" and not config.is_on_policy

else config.run.mode.value

)

),

"loss_mode": config.loss.mode.value if config.run.mode.value != "sft" else None,

"divergence": config.loss.divergence.value if config.run.mode.value != "sft" else None,

"top_k": config.loss.top_k if config.run.mode.value != "sft" else None,

"selector": config.selection.selector.value,

"memory_strategy": config.memory.strategy.value,

"cycles": config.train.cycles,

"sft_warmup_cycles": config.train.sft_warmup_cycles,

"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 if environment_config is not None else config.eval.tasks

),

"seed": config.run.seed,

"warnings": warnings,

}

if as_json:

_emit_json(payload)

return

console.print(f"[green]valid[/green] {_esc(recipe)}")

table = Table(show_header=False, box=None, pad_edge=False)

for key, value in payload.items():

if key in {"valid", "path", "warnings"}:

continue

table.add_row(f"[dim]{key}[/dim]", _esc(value))

console.print(table)

for warning in warnings:

console.print(f"[yellow]warning[/yellow] {_esc(warning)}")

# ------------------------------------------------------------------ demo

@app.command()

def demo(

output: Path = typer.Option(Path("runs/demo"), "--output", help="Run directory to create."),

fast: bool = typer.Option(False, "--fast", help="Shrink every budget (CI smoke test)."),

overwrite: bool = typer.Option(

False,

"--overwrite",

help="Explicitly replace the whole existing demo run directory.",

),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

report_html: bool = typer.Option(True, "--report/--no-report", help="Also render report.html."),

) -> None:

"""Run the embedded no-network toy pipeline end to end."""

try:

_require_training_stack("miniverl demo")

from miniverl.demo import demo_config

from miniverl.trainer import OPDTrainer

except (MiniVerlError, ModuleNotFoundError) as exc:

_fail(exc)

return

target = Path(output)

try:

config = demo_config(fast=fast, output_dir=target.parent)

if overwrite and target.exists() and not as_json:

console.print(

f"[yellow]overwrite[/yellow] replacing whole run directory {_esc(target)}"

)

with OPDTrainer.from_config(

config,

output_dir=target.parent,

run_id=target.name,

overwrite=overwrite,

) as trainer:

result = trainer.train()

paths = trainer.paths

report_path: Path | None = None

if report_html:

from miniverl.reporting import ReportData, write_markdown, write_report

report_path = write_report(paths.root, paths.report_html)

write_markdown(ReportData.from_run(paths.root), paths.summary_md)

artifacts = _artifact_listing(paths.root)

except MiniVerlError as exc:

_fail(exc)

return

payload = {

**result.to_dict(),

"artifacts": artifacts,

"report": str(report_path) if report_path else None,

}

if as_json:

_emit_json(payload)

return

console.print()

console.print(f"[bold green]demo complete[/bold green] {_esc(paths.root)}")

table = Table(show_header=False, box=None)

baseline = (result.baseline_eval or {}).get("success_rate")

final = (result.eval or {}).get("success_rate")

table.add_row("mode", f"{result.mode} (genuine on-policy distillation)")

table.add_row("optimizer steps", str(result.global_step))

table.add_row("parameter version", str(result.parameter_version))

table.add_row("rollout iterations", str(result.cycles_completed))

table.add_row("wall clock", f"{result.duration_seconds:.1f} s")

provenance = _provenance_summary(paths.trajectories)

if provenance:

table.add_row("token provenance", provenance)

compression = _cache_summary(paths.teacher_cache)

if compression:

table.add_row("teacher cache", compression)

table.add_row(

"task success",

f"{_fmt_pct(baseline)} -> {_fmt_pct(final)} (greedy, held-out eval split)",

)

console.print(table)

console.print()

for line in (

"This demo proves the [bold]machinery[/bold], not capability.",

"At this size the toy student learns the tool-call format and not the",

"arithmetic, so 0% here is the expected outcome, not a failure.",

"For a CPU run that does learn (measured 0.0% -> 91.7% in 192 s):",

" [bold]miniverl train recipes/toy_cpu.yaml[/bold]",

):

console.print(line)

console.print()

console.print("[bold]artifacts[/bold]")

for name, size in artifacts.items():

console.print(f" {_esc(name)} [dim]{_esc(size)}[/dim]")

console.print()

console.print("[bold]next[/bold]")

console.print(f" miniverl inspect {_esc(paths.trajectories)}")

console.print(f" miniverl cache stats {_esc(paths.teacher_cache)}")

console.print(f" miniverl report {_esc(paths.root)} --out {_esc(paths.report_html)}")

def _provenance_summary(path: Path) -> str:

"""One line summarizing which tokens could enter the loss."""

try:

from miniverl.inspection import summarize_file

summary = summarize_file(path, limit=0)

except (MiniVerlError, OSError):

# A cosmetic summary line must never turn a completed run into a failure.

return ""

if not summary.tokens:

return ""

return (

f"{summary.model_tokens} of {summary.tokens} tokens trainable "

f"({summary.model_token_fraction * 100:.0f}%); "

f"{summary.context_tokens} are context and can never be a target"

)

def _cache_summary(path: Path) -> str:

"""One line summarizing the teacher-target cache."""

try:

from miniverl.cache.stats import compute_stats

stats = compute_stats(path, verify_checksums=False)

except (MiniVerlError, OSError):

return ""

if not stats.get("selected_positions"):

return ""

return (

f"{stats['selected_positions']} scored positions, "

f"{stats['actual_bytes'] / 1024:.1f} KiB on disk, "

f"{stats['compression_ratio']:.1f}x smaller than a dense fp16 dump"

)

def _fmt_pct(value: Any) -> str:

try:

return f"{float(value) * 100:.1f}%"

except (TypeError, ValueError):

return "n/a"

def _artifact_listing(root: Path) -> dict[str, str]:

out: dict[str, str] = {}

for path in sorted(root.rglob("*")):

if path.is_file():

rel = path.relative_to(root).as_posix()

out[rel] = f"{path.stat().st_size} B"

return out

# ----------------------------------------------------- prepare-offline-kd

@app.command("prepare-offline-kd")

def prepare_offline_kd_command(

recipe: Path = typer.Option(..., "--recipe", help="Frozen-student offline-KD recipe."),

checkpoint: Path = typer.Option(

..., "--checkpoint", help="Exact shared cold-start checkpoint directory."

),

out: Path = typer.Option(..., "--out", help="New immutable dataset bundle directory."),

offline: bool = typer.Option(

False, "--offline", help="Refuse network access; use only cached model files."

),

dry_run: bool = typer.Option(

False,

"--dry-run",

help="Validate and print the collection plan without loading models or writing files.",

),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

) -> None:

"""Collect one frozen-student trajectory/teacher-target dataset."""

from miniverl.config import OfflineKDTrajectorySource, RunConfig, TrainingMode

try:

config = RunConfig.from_yaml(recipe)

if config.run.mode is not TrainingMode.OFFLINE_KD:

raise ConfigError("prepare-offline-kd requires run.mode=offline_kd")

if config.offline_kd.trajectory_source is not OfflineKDTrajectorySource.FROZEN_STUDENT:

raise ConfigError(

"prepare-offline-kd requires offline_kd.trajectory_source=frozen_student"

)

if not checkpoint.is_dir():

raise ConfigError(f"checkpoint directory not found: {checkpoint}")

if out.exists():

raise ConfigError(

f"output already exists: {out}",

hint="choose a new directory; frozen datasets are never overwritten",

)

except (ValidationError, MiniVerlError) as exc:

if isinstance(exc, MiniVerlError):

_fail(exc)

err_console.print(f"[red]invalid recipe[/red] {_esc(recipe)}\n{_esc(exc)}")

raise typer.Exit(1) from None

plan = {

"dry_run": dry_run,

"recipe": str(recipe),

"checkpoint": str(checkpoint),

"out": str(out),

"offline": offline,

"trajectory_source": config.offline_kd.trajectory_source.value,

"collection_seed": config.offline_kd.collection_seed,

"collection_tasks": (config.offline_kd.collection_tasks or config.train.rollouts_per_cycle),

"student": config.models.student.model_id,

"student_revision": config.models.student.revision,

"teacher": config.models.teacher.model_id,

"teacher_revision": config.models.teacher.revision,

}

if dry_run:

if as_json:

_emit_json(plan)

else:

console.print(f"[green]dry run ok[/green] {_esc(recipe)}")

for key, value in plan.items():

console.print(f" [dim]{key}[/dim] {_esc(value)}")

console.print("\nNo models were loaded, files written, or downloads attempted.")

return

try:

_require_training_stack("miniverl prepare-offline-kd")

from miniverl.trainer import OPDTrainer

from miniverl.training.checkpoint import load_checkpoint, validate_checkpoint

validated = validate_checkpoint(checkpoint)

with OPDTrainer.from_config(

config,

output_dir=out.parent,

run_id=out.name,

local_files_only=offline,

) as trainer:

load_checkpoint(

checkpoint,

backend=trainer.student,

optimizer=None,

device=trainer.student.device,

include_optimizer=False,

include_rng=False,

expected_identity=(trainer._checkpoint_identity() if validated.identity else None),

)

trainer.set_offline_collection_checkpoint_digest(validated.content_digest)

summary = trainer.prepare_offline_dataset()

destination = trainer.paths.root

except (MiniVerlError, ModuleNotFoundError) as exc:

_fail(exc)

return

payload = {**plan, "written": str(destination), **summary}

if as_json:

_emit_json(payload)

return

console.print(f"[green]offline dataset prepared[/green] {_esc(destination)}")

console.print(f" digest {_esc(summary['dataset_digest'])}")

console.print(f" trajectories {_esc(summary['trajectories'])}")

# --------------------------------------------------------- qualify-teacher

@app.command("qualify-teacher")

def qualify_teacher_command(

recipe: Path = typer.Option(..., "--recipe", help="Recipe defining the frozen teacher."),

candidate: str = typer.Option(..., "--candidate", help="Preregistered candidate id."),

out: Path = typer.Option(..., "--out", help="New immutable gate-result directory."),

tasks: Optional[int] = typer.Option(None, "--tasks", help="Eval-task count override."),

offline: bool = typer.Option(

False, "--offline", help="Refuse network access; use only cached model files."

),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

) -> None:

"""Evaluate one RecoveryBench teacher candidate on eval only."""

from miniverl.config import RunConfig

try:

config = RunConfig.from_yaml(recipe)

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

result = evaluate_teacher_candidate(

config,

candidate_id=candidate,

out=out,

tasks=tasks,

split="eval",

local_files_only=offline,

)

except (ValidationError, MiniVerlError, ModuleNotFoundError) as exc:

if isinstance(exc, ValidationError):

err_console.print(f"[red]invalid recipe[/red] {_esc(recipe)}\n{_esc(exc)}")

raise typer.Exit(1) from None

_fail(exc)

return

if as_json:

_emit_json(result)

return

status = "passed" if result["gate"]["passed"] else "failed"

style = "green" if result["gate"]["passed"] else "yellow"

console.print(f"[{style}]teacher gate {status}[/{style}] {_esc(candidate)}")

console.print(f" result {_esc(out / 'result.json')}")

# ------------------------------------------------------------- evidence

@evidence_app.command("show")

def evidence_show(

study_id: str = typer.Argument(..., help="Packaged study identifier."),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

) -> None:

"""Show a packaged, typed study result without a repository checkout."""

from miniverl.evidence import show_builtin_study

try:

payload = show_builtin_study(study_id)

except (MiniVerlError, OSError, ValidationError) as exc:

_fail(exc)

return

if as_json:

_emit_json(payload)

return

console.print(f"[bold]{_esc(study_id)}[/bold]")

console.print_json(json.dumps(payload["result"], allow_nan=False))

@evidence_app.command("validate")

def evidence_validate(

study_id: str = typer.Argument(..., help="Packaged study identifier."),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

) -> None:

"""Validate packaged result, schema, preregistration and task evidence."""

from miniverl.evidence import validate_builtin_study

try:

payload = validate_builtin_study(study_id)

except (MiniVerlError, OSError, ValidationError) as exc:

_fail(exc)

return

if as_json:

_emit_json(payload)

elif payload["valid"]:

console.print(

f"[green]valid[/green] {_esc(study_id)} · {_esc(payload['task_rows'])} task rows"

)

else:

for problem in payload["problems"]:

err_console.print(f"[red]invalid[/red] {_esc(problem)}")

raise typer.Exit(1)

# ----------------------------------------------------------------- pilot

@app.command()

def pilot(

recipe: Optional[Path] = typer.Argument(

None, help="Alignment recipe containing bounded pilot evidence."

),

study_result: Optional[Path] = typer.Option(

None,

"--study-result",

help="Schema-validated external-study result; does not load a model.",

),

builtin_study: Optional[str] = typer.Option(

None,

"--builtin-study",

help="Packaged external-study result; works from an installed wheel.",

),

out: Optional[Path] = typer.Option(None, "--out", help="Optional JSON output path."),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

) -> None:

"""Recommend a method from explicit pilot evidence without loading a model."""

from miniverl.utils.runs import write_json_atomic

payload: dict[str, Any]

try:

selected = sum(value is not None for value in (recipe, study_result, builtin_study))

if selected > 1:

raise ConfigError(

"miniverl pilot accepts exactly one of a recipe, --study-result, or --builtin-study"

)

builtin = None

if builtin_study is not None:

from miniverl.evidence import get_builtin_study

builtin = get_builtin_study(builtin_study)

study_result = builtin.result_path

if study_result is not None:

from miniverl.alignment_external.result import load_alignment_external_result

result = load_alignment_external_result(study_result)

if result.study_status != "terminated_at_checkpoint_selection":

raise ConfigError(

"this pilot evidence path currently requires "

"study_status=terminated_at_checkpoint_selection"

)

payload = {

"study_status": result.study_status,

"recommendation": "do_not_continue",

"recommendation_scope": "do_not_continue_this_study",

"method_recommendation": "insufficient_evidence",

"reasons": [

"no candidate satisfied the retained-utility gate",

"no starting checkpoint was selected",

"teacher qualification was not run",

"no continuation method was authorized",

"the reserved final test was not accessed",

],

"evidence": {

"path": str(study_result),

"builtin_study": builtin.study_id if builtin is not None else None,

"sha256": hashlib.sha256(study_result.read_bytes()).hexdigest(),

"preregistration": result.preregistration.model_dump(mode="json"),

"task_evidence": result.checkpoint_selection.task_evidence.model_dump(

mode="json"

),

},

"universal_claim": False,

}

else:

if recipe is None:

raise ConfigError(

"miniverl pilot requires a recipe, --study-result or --builtin-study"

)

from miniverl.alignment import PilotEvidence, recommend_alignment_method

from miniverl.config import RunConfig

config = RunConfig.from_yaml(recipe)

if config.alignment is None:

raise ConfigError("miniverl pilot requires a recipe with an alignment section")

evidence = config.alignment.pilot or PilotEvidence()

recommendation = recommend_alignment_method(evidence)

payload = recommendation.model_dump(mode="json")

if out is not None:

write_json_atomic(out, payload)

except (ValidationError, MiniVerlError) as exc:

if isinstance(exc, MiniVerlError):

_fail(exc)

source = recipe if recipe is not None else study_result or builtin_study

err_console.print(f"[red]invalid pilot evidence[/red] {_esc(source)}\n{_esc(exc)}")

raise typer.Exit(1) from None

if as_json:

_emit_json(payload)

return

recommendation_text = payload.get("method_recommendation", payload["recommendation"])

console.print(f"[bold]recommendation[/bold] {_esc(recommendation_text)}")

for reason in payload["reasons"]:

console.print(f" - {_esc(reason)}")

if out is not None:

console.print(f" evidence {_esc(out)}")

@app.command()

def align(

recipe: Path = typer.Argument(..., help="Path to a post-SFT alignment recipe YAML file."),

output: Optional[Path] = typer.Option(

None, "--output", help="Parent directory for the run (default: run.output_dir)."

),

run_id: Optional[str] = typer.Option(None, "--run-id", help="Explicit run id."),

overwrite: bool = typer.Option(

False,

"--overwrite",

help="Explicitly replace the whole target run directory.",

),

dry_run: bool = typer.Option(

False,

"--dry-run",

help="Validate and print all alignment stages without loading models.",

),

offline: bool = typer.Option(

False, "--offline", help="Refuse network access; use only cached model files."

),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

) -> None:

"""Run base -> SFT checkpoint -> teacher -> alignment -> eval -> card."""

from miniverl.alignment import build_alignment_stage_plan

from miniverl.config import RunConfig

try:

config = RunConfig.from_yaml(recipe)

if config.alignment is None:

raise ConfigError("miniverl align requires a recipe with an alignment section")

except (ValidationError, MiniVerlError) as exc:

if isinstance(exc, MiniVerlError):

_fail(exc)

err_console.print(f"[red]invalid recipe[/red] {_esc(recipe)}\n{_esc(exc)}")

raise typer.Exit(1) from None

workflow = build_alignment_stage_plan(

config.alignment,

sft_warmup_cycles=config.train.sft_warmup_cycles,

)

if dry_run:

payload = {

"dry_run": True,

"recipe": str(recipe),

"method": config.alignment.method.value,

"workflow": workflow,

"backend": config.models.backend.value,

"downloads_required": config.models.backend.value == "hf",

"output_dir": str(output or config.run.output_dir),

}

if as_json:

_emit_json(payload)

else:

console.print(f"[green]alignment dry run ok[/green] {_esc(recipe)}")

for stage in workflow["stages"]:

console.print(f" - {_esc(stage['name'])}")

console.print("\nNo models were loaded and nothing was downloaded.")

return

try:

_require_training_stack("miniverl align")

from miniverl.alignment import run_alignment

payload = run_alignment(

config,

output_dir=output,

run_id=run_id,

local_files_only=offline,

overwrite=overwrite,

)

except (MiniVerlError, ModuleNotFoundError) as exc:

_fail(exc)

return

if as_json:

_emit_json(payload)

return

console.print(f"[bold green]alignment complete[/bold green] {_esc(payload['run_dir'])}")

console.print(f" card {_esc(Path(str(payload['run_dir'])) / 'alignment-card.md')}")

# ---------------------------------------------------------- verl-shaped plan/run

@app.command(

"plan",

context_settings={"allow_extra_args": True, "ignore_unknown_options": True},

)

def plan_command(

ctx: typer.Context,

config: str = typer.Option(..., "--config", help="Resolved YAML path or builtin profile."),

profile: str = typer.Option(

"verl-opd-v0.8-single-gpu-v1", "--profile", help="Pinned compatibility profile."

),

overrides: list[str] = typer.Option([], "--set", help="Repeatable dotted key=value override."),

override_files: list[Path] = typer.Option(

[],

"--overrides-file",

help="Repeatable plain key=value file or JSON argv array.",

),

accept_local_reinterpretations: bool = typer.Option(

False,

"--accept-local-reinterpretations",

help="Accept the reported acknowledgement-required one-GPU reinterpretations.",

),

out: Optional[Path] = typer.Option(

None,

"--out",

help="Atomically write a data-bound immutable execution plan.",

),

as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),

offline: bool = typer.Option(False, "--offline", help="Do not access the network."),

probe: bool = typer.Option(False, "--probe", help="Run a bounded no-update CUDA calibration."),

probe_cache: Path = typer.Option(

Path.home() / ".cache" / "miniverl" / "probes",

"--probe-cache",

help="Exact-identity hardware probe cache directory.",

),

force_probe: bool = typer.Option(

False, "--force-probe", help="Ignore a compatible cached probe and measure again."

),

) -> None:

"""Plan pinned single-GPU verl-style OPD without loading model weights."""

try:

from miniverl.bridge.opd_runtime import build_system_plan

from miniverl.bridge.opd_v08 import (

VERL_OPD_V08_PROFILE,

load_verl_opd_v08_source,

)

if profile != VERL_OPD_V08_PROFILE:

raise ConfigError(

f"unsupported OPD profile {profile!r}", hint=f"use --profile {VERL_OPD_V08_PROFILE}"

)

compiled = load_verl_opd_v08_source(

config,

override_files=override_files,

overrides=overrides,

trailing_overrides=ctx.args,

accept_local_reinterpretations=accept_local_reinterpretations,

)

plan = build_system_plan(compiled)

payload = plan.model_dump(mode="json")

artifact = None

if out is not None or probe:

from miniverl.bridge.opd_plan import (

attach_hardware_probe,

build_immutable_opd_plan,

write_immutable_opd_plan,

)

artifact = build_immutable_opd_plan(compiled, source=config, system_plan=plan)

if probe:

_require_training_stack("miniverl plan --probe")

from miniverl.bridge.opd_probe import run_hardware_probe

from miniverl.config import RunConfig

native = RunConfig.model_validate(artifact.resolved_native_config)

measured = run_hardware_probe(

native,

plan_digest=artifact.plan_digest,

cache_dir=probe_cache,

offline=offline,

force=force_probe,

)

artifact = attach_hardware_probe(artifact, measured)

if out is not None:

write_immutable_opd_plan(out, artifact)

payload = artifact.model_dump(mode="json")

except MiniVerlError as exc:

_fail(exc)