La tabla cada noche: doce etapas, un comando, una fecha

class NightlyRecord(BaseModel):
    """One stage of the run: what was measured, from what, and how it went."""

    model_config = ConfigDict(extra="forbid")

    name: str
    stage: str
    measure: str
    checkpoint: str | None = None
    model_sha: str | None = None
    weights_sha: str | None = None
    seconds: float = 0.0
    status: str = "planned"
    """``planned`` (dry run), ``measured``, ``skipped`` or ``failed``."""
    detail: str | None = None


class NightlyReport(BaseModel):
    """What one nightly did; serialised as ``artifacts/eval/nightly.json``."""

    model_config = ConfigDict(extra="forbid")

    date: str
    config: str
    encoder_config: str | None
    records: list[NightlyRecord]
    results: str
    benchmarks: str | None
    seconds: float

def _run_decoder(checkpoint: Path, config: Path | None, stage: str, use_cache: bool, device):
    from rukh.eval.suite import load_suite, run_suite

    cfg = load_suite("full", config).model_copy(update={"stage": stage})
    return run_suite(checkpoint, cfg, suite="full", use_cache=use_cache, device=device)


def _run_encoder(checkpoint: Path, config: Path | None, stage: str, use_cache: bool, device):
    from rukh.eval.encoder import load_encoder_suite, run_encoder_suite

    cfg = load_encoder_suite(config).model_copy(update={"stage": stage})
    return run_encoder_suite(checkpoint, cfg, use_cache=use_cache, device=device)


def _run_qwen(adapter_dir: Path, config: Path | None, stage: str, use_cache: bool, device):
    from rukh.eval.qwen_suite import run_qwen_suite
    from rukh.eval.suite import load_suite

    cfg = load_suite("full", config)
    return run_qwen_suite(adapter_dir, cfg, stage=stage, use_cache=use_cache, device=device)


def _run_karvonen(checkpoint: Path, config: Path | None, stage: str, use_cache: bool, device):
    from rukh.eval.karvonen_suite import run_karvonen_suite
    from rukh.eval.suite import load_suite

    cfg = load_suite("full", config)
    return run_karvonen_suite(
        cfg, stage=stage, use_cache=use_cache, device=device, checkpoint=checkpoint
    )

def _checkpoint_of(artefact: Artefact, pull_missing: bool) -> Path:
    """Where the artefact is on disk, pulling it from the Hub when it is not and allowed."""
    if artefact.measure == "karvonen":
        from rukh.eval.karvonen import ensure_karvonen

        return ensure_karvonen(download=pull_missing)
    target = resolve_run(paths.resolve(artefact.target))
    if not target.exists() and pull_missing:
        target = pull(artefact)
    if not target.exists():
        raise FileNotFoundError(
            f"{artefact.name}: {target} is not on disk (rukh pull {artefact.name})"
        )
    if artefact.base is not None:
        base = resolve_run(paths.resolve(artefact.base))
        if not base.exists() and pull_missing:
            from rukh.hub import lookup

            base = pull(lookup("medium-v4"))
        if not base.exists():
            raise FileNotFoundError(f"{artefact.name}: base {base} is not on disk")
        return merged_checkpoint(base, target, paths.resolve(MERGED_DIR) / f"{artefact.name}.pt")
    return target

    results_path = results or paths.resolve("artifacts/web/results.json")
    benchmarks_path: Path | None = None
    if not dry_run and results_path.is_file():
        benchmarks_path = benchmarks or paths.resolve(BENCHMARKS_FILE)
        if benchmarks_path.is_file():
            write_benchmarks(results_path, benchmarks_path)
        else:
            benchmarks_path = None
    report = NightlyReport(
        date=datetime.now(UTC).date().isoformat(),
        config=(config or Path("configs/eval/full.yaml")).as_posix(),
        encoder_config=encoder_config.as_posix() if encoder_config else None,
        records=records,
        results=results_path.as_posix(),
        benchmarks=benchmarks_path.as_posix() if benchmarks_path else None,
        seconds=round(time.perf_counter() - started, 1),
    )
    if not dry_run:
        target = out or paths.resolve(NIGHTLY_FILE)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(report.model_dump_json(indent=2) + "\n", encoding="utf-8", newline="\n")
    return report

def render_plan(report: NightlyReport) -> str:
    """One line per stage, for the terminal."""
    lines = [f"{'stage':<32}{'measure':<9}{'status':<10}{'seconds':>8}  checkpoint"]
    for record in report.records:
        lines.append(
            f"{record.stage:<32}{record.measure:<9}{record.status:<10}{record.seconds:>8.1f}  "
            f"{record.checkpoint or '-'}" + (f"  ({record.detail})" if record.detail else "")
        )
    return "\n".join(lines)

    for row in playing:
        lines.append(
            "| "
            + " | ".join(
                [
                    f"`{row.get('stage')}`" + (" (externo)" if row.get("baseline") else ""),
                    _percent(row.get("legality")),
                    _percent(row.get("top1")),
                    _percent(row.get("top3")),
                    _band(row, "puzzles", "1000-1500"),
                    _band(row, "puzzles", "1500-2000"),
                    _band(row, "puzzles", "2000+"),
                    _elo_cell(row),
                    "n/a" if row.get("delta_cp") is None else f"{row['delta_cp']:.1f}",
                    "n/a"
                    if row.get("first_move_entropy") is None
                    else f"{row['first_move_entropy']:.4f}",
                    str(row.get("date") or ""),
                ]
            )
            + " |"
        )

    lines.append("")
    external = sum(1 for row in playing if row.get("baseline"))
    lines.append(
        f"{len(playing)} etapas medidas con la misma suite"
        + (
            f", {external} de ellas externas (modelos de otros autores, marcados como "
            "`(externo)`, medidos con este mismo harness como control)"
            if external
            else ""
        )
        + ". Las filas del encoder viven aparte porque no comparten una sola columna con "
        "estas; las que una medición retiró no están (`rukh eval drop`)."
    )

@eval_app.command("nightly")
def eval_nightly_cmd(
    config: Annotated[
        Path | None,
        typer.Option(
            "--config",
            exists=True,
            dir_okay=False,
            help="Suite YAML every decoder is measured with (default: configs/eval/greedy.yaml).",
        ),
    ] = None,
    encoder_config: Annotated[
        Path | None,
        typer.Option("--encoder-config", exists=True, dir_okay=False, help="Encoder suite YAML."),
    ] = None,
    only: Annotated[
        str | None, typer.Option("--only", help="Comma-separated names or stages to measure.")
    ] = None,
    dry_run: Annotated[
        bool, typer.Option("--dry-run", help="List what would be measured and exit.")
    ] = False,
    no_pull: Annotated[
        bool, typer.Option("--no-pull", help="Never download; a missing checkpoint fails.")
    ] = False,
    no_cache: Annotated[bool, typer.Option("--no-cache", help="Replay every game.")] = False,
    device: Annotated[str | None, typer.Option("--device", help="Where to run.")] = None,
) -> None:
    """Measure every published stage under one config and rewrite the whole table.

    The catalogue of `rukh pull` says which stage each model is and how it is measured; what the
    caches hold is not replayed, so a nightly after a nightly costs minutes. `artifacts/eval/
    nightly.json` records what came from where, with which sha, and how long it took.
    """
    from rukh.eval.nightly import render_plan, run_nightly
    from rukh.paths import resolve

    logging.basicConfig(level=logging.INFO, format="%(asctime)s  %(message)s")
    wanted = {part.strip() for part in (only or "").split(",") if part.strip()} or None
    report = run_nightly(
        config=config or resolve("configs/eval/greedy.yaml"),
        encoder_config=encoder_config,
        only=wanted,
        dry_run=dry_run,
        pull_missing=not no_pull,
        use_cache=not no_cache,
        device=device,
    )
    typer.echo(render_plan(report))
    if not dry_run:
        typer.echo(f"results:    {report.results}")
        typer.echo(f"benchmarks: {report.benchmarks or 'not rewritten'}")
        typer.echo(f"seconds:    {report.seconds:.0f}")
    if any(record.status == "failed" for record in report.records):
        raise typer.Exit(code=1)

"""Tests for rukh.eval.nightly: the catalogue drives the run, and the run says what it did."""

from __future__ import annotations

import json
from pathlib import Path

import pytest
import torch
from typer.testing import CliRunner

from rukh.cli import app
from rukh.eval import nightly
from rukh.hub import catalogue
from rukh.models import DecoderConfig, MoveDecoder
from rukh.train import load_model, save_checkpoint

pytestmark = pytest.mark.unit

TOY = DecoderConfig(vocab_size=64, n_layer=1, n_head=2, d_model=16, block=32)


def test_every_published_model_the_table_cites_is_in_the_plan() -> None:
    """The stages of `docs/benchmarks.md` that a reader can pull are all measured by nightly."""
    stages = {a.stage for a in nightly.plan()}
    assert stages >= {
        "tiny-greedy",
        "small-v3-greedy",
        "medium-v4-greedy",
        "medium-elo",
        "medium-masters",
        "lora-e4",
        "lora-d4",
        "qwen3-pgn-qlora",
        "medium-v4-dpo-onpolicy-greedy",
        "medium-v4-grpo-greedy",
        "encoder-v4",
    }
    # Nothing without a way to measure it: datasets, the reward model, the pretraining encoder.
    assert all(a.measure != "none" for a in nightly.plan())
    assert {a.name for a in catalogue() if a.measure == "none"} >= {"rm", "encoder-mmm-v4"}
    assert [a.name for a in nightly.plan({"tiny", "encoder-v4"})] == ["tiny", "encoder-v4"]
    assert [a.stage for a in nightly.plan({"medium-elo"})] == ["medium-elo"]

def _toy(path: Path, seed: int = 0) -> Path:
    torch.manual_seed(seed)
    return save_checkpoint(
        path, step=1, model=MoveDecoder(TOY), optimizer=None, cfg={}, model_cfg=TOY.model_dump()
    )


def test_a_dry_run_lists_the_plan_and_writes_nothing(rukh_home: Path) -> None:
    report = nightly.run_nightly(dry_run=True, only={"tiny", "encoder-v4"})
    assert [r.status for r in report.records] == ["planned", "planned"]
    assert not (rukh_home / "artifacts" / "eval" / "nightly.json").exists()
    assert nightly.render_plan(report).count("planned") == 2

def test_the_run_measures_in_catalogue_order_and_records_what_it_did(
    rukh_home: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    measured: list[tuple[str, str]] = []

    def fake_decoder(checkpoint, config, stage, use_cache, device):  # noqa: ANN001
        measured.append((stage, Path(checkpoint).name))
        return None

    def fake_encoder(checkpoint, config, stage, use_cache, device):  # noqa: ANN001
        measured.append((stage, Path(checkpoint).name))
        return None

    monkeypatch.setattr(nightly, "_run_decoder", fake_decoder)
    monkeypatch.setattr(nightly, "_run_encoder", fake_encoder)
    monkeypatch.setattr(
        nightly, "_run_qwen", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("no gpu"))
    )
    # `tiny` under its stamped run folder: the stable name resolves to it.
    _toy(rukh_home / "checkpoints" / "tiny-20260919-060101" / "best.pt")
    _toy(rukh_home / "checkpoints" / "encoder-heads-v4" / "step-4000.pt")
    (rukh_home / "checkpoints" / "qwen3-pgn-qlora").mkdir(parents=True)
    results = rukh_home / "artifacts" / "web" / "results.json"
    results.parent.mkdir(parents=True)
    results.write_text(json.dumps({"rows": [{"stage": "kept", "params": 1}]}), encoding="utf-8")
    benchmarks = rukh_home / "docs" / "benchmarks.md"
    benchmarks.parent.mkdir(parents=True)
    benchmarks.write_text("# B\n\ntexto\n\n## Resultados\n\nold\n", encoding="utf-8")

    report = nightly.run_nightly(
        only={"tiny", "encoder-v4", "qwen3-pgn-qlora", "medium-v4"},
        pull_missing=False,
        results=results,
        benchmarks=benchmarks,
    )
    assert measured == [("tiny-greedy", "best.pt"), ("encoder-v4", "step-4000.pt")]
    by_name = {r.name: r for r in report.records}
    assert by_name["tiny"].status == "measured" and by_name["tiny"].model_sha
    assert by_name["tiny"].checkpoint.endswith("tiny-20260919-060101/best.pt")
    assert by_name["qwen3-pgn-qlora"].status == "failed"
    assert "no gpu" in (by_name["qwen3-pgn-qlora"].detail or "")
    # A model that is not on disk and may not be pulled fails on its own, not the run.
    assert by_name["medium-v4"].status == "failed"
    assert "rukh pull medium-v4" in (by_name["medium-v4"].detail or "")
    # The rows nightly did not measure are untouched; the benchmarks doc was rewritten.
    assert json.loads(results.read_text(encoding="utf-8"))["rows"] == [
        {"stage": "kept", "params": 1}
    ]
    assert "texto" in benchmarks.read_text(encoding="utf-8")
    assert "old" not in benchmarks.read_text(encoding="utf-8")
    written = json.loads((rukh_home / "artifacts" / "eval" / "nightly.json").read_text("utf-8"))
    assert [r["name"] for r in written["records"]] == [
        "tiny",
        "medium-v4",
        "encoder-v4",
        "qwen3-pgn-qlora",
    ]

def test_an_adapter_is_merged_onto_its_base_before_it_is_measured(
    rukh_home: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from rukh.models.lora import LoraConfig, apply_lora, save_adapter

    base = _toy(rukh_home / "checkpoints" / "medium-v4" / "best.pt")
    torch.manual_seed(1)
    model = MoveDecoder(TOY)
    model.load_state_dict(load_model(base)[0].state_dict())
    apply_lora(model, LoraConfig(r=2, alpha=4, targets=["q", "v"]))
    with torch.no_grad():
        for name, parameter in model.named_parameters():
            if ".b." in name:
                parameter.add_(0.5)
    adapter_dir = rukh_home / "checkpoints" / "lora-e4"
    adapter_dir.mkdir(parents=True)
    save_adapter(
        model, adapter_dir / "adapter.safetensors", LoraConfig(r=2, alpha=4, targets=["q", "v"])
    )

    seen: dict[str, Path] = {}
    monkeypatch.setattr(
        nightly, "_run_decoder", lambda ckpt, *a, **k: seen.__setitem__("ckpt", Path(ckpt))
    )
    report = nightly.run_nightly(only={"lora-e4"}, pull_missing=False)
    assert report.records[0].status == "measured"
    merged, payload = load_model(seen["ckpt"])
    assert payload["cfg"]["adapter"].endswith("lora-e4")
    plain, _ = load_model(base)
    # The merged weights differ from the base exactly where the adapter pushed.
    assert not torch.equal(merged.blocks[0].attn.qkv.weight, plain.blocks[0].attn.qkv.weight)
    assert not any(".lora" in k or ".a." in k for k in merged.state_dict())

def test_the_cli_dry_run_prints_the_plan(rukh_home: Path) -> None:
    result = CliRunner().invoke(app, ["eval", "nightly", "--dry-run", "--only", "tiny,rm"])
    assert result.exit_code == 0, result.output
    assert "tiny-greedy" in result.output and "planned" in result.output
    assert "rm" not in result.output.replace("tiny-greedy", "")