tinymind-native-8b-remote-handoff/bundle/evaluation/leaderboard_safe_improvement.py

"""Leaderboard-safe improvement gate.

The goal is improvement that can survive hidden/official leaderboard review,
not just a strong offline public score.  This gate separates three things:

1. training progress;
2. raw, uncontrolled model capability;
3. official/hidden leaderboard readiness.

Protocol repair, public benchmark overfitting, and local smoke scores are kept
as engineering evidence but never counted as leaderboard-safe model gains.
"""

from __future__ import annotations

from datetime import datetime, timezone
import json
from pathlib import Path
from typing import Any


def _load(path: str | Path | None) -> dict[str, Any]:
    if not path:
        return {}
    p = Path(path)
    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {}


def _num(payload: dict, *keys: str, default: float = 0.0) -> float:
    cur: Any = payload
    for key in keys:
        if not isinstance(cur, dict) or key not in cur:
            return default
        cur = cur[key]
    try:
        return float(cur)
    except (TypeError, ValueError):
        return default


def _bool(payload: dict, *keys: str) -> bool:
    cur: Any = payload
    for key in keys:
        if not isinstance(cur, dict) or key not in cur:
            return False
        cur = cur[key]
    return cur is True


def _official_ready(payload: dict) -> bool:
    if not payload:
        return False
    gate = payload.get("claim_gate", {})
    if gate.get("official_external_claim_allowed") is True:
        return True
    if gate.get("raw_external_gate_complete") is True:
        return True
    targets = payload.get("targets", {})
    if isinstance(targets, dict) and targets:
        return all(str(row.get("status")) in {"ready", "ready_to_contact"} for row in targets.values() if isinstance(row, dict))
    return False


def build_leaderboard_safe_improvement_report(
    out_dir: str | Path,
    *,
    train_report: str | Path | None = None,
    raw_probe_report: str | Path | None = None,
    controlled_probe_report: str | Path | None = None,
    official_report: str | Path | None = None,
    contamination_report: str | Path | None = None,
) -> dict[str, Any]:
    train = _load(train_report)
    raw = _load(raw_probe_report)
    controlled = _load(controlled_probe_report)
    official = _load(official_report)
    contamination = _load(contamination_report)

    pre_loss = _num(
        train,
        "training",
        "pre_eval_loss",
        default=_num(train, "metrics", "pre_eval_loss", default=_num(train, "pre_eval_loss", default=0.0)),
    )
    post_loss = _num(
        train,
        "training",
        "post_eval_loss",
        default=_num(train, "metrics", "post_eval_loss", default=_num(train, "post_eval_loss", default=0.0)),
    )
    loss_delta = pre_loss - post_loss if pre_loss and post_loss else 0.0
    split = train.get("split_report", train.get("split", {}))

    controlled_enabled = bool(controlled.get("controlled_repair_enabled") or _bool(controlled, "summary", "controlled_repair_enabled"))
    raw_controlled_enabled = bool(raw.get("controlled_repair_enabled") or _bool(raw, "summary", "controlled_repair_enabled"))
    raw_native_score = _num(raw, "summary", "native_score", default=_num(raw, "score", default=0.0))
    raw_baseline_score = _num(raw, "summary", "baseline_score", default=0.0)

    contamination_cleared = (
        contamination.get("claim_gate", {}).get("contamination_cleared") is True
        or contamination.get("claim_gate", {}).get("anti_contamination_passed") is True
        or contamination.get("contamination_risk", 100.0) == 0
    )
    checks = {
        "assumption_leaderboard_safe_not_offline_public_score": True,
        "training_eval_improved": bool(
            train.get("training", {}).get(
                "eval_loss_improved",
                train.get("metrics", {}).get("eval_loss_improved", train.get("eval_loss_improved", False)),
            )
        )
        or loss_delta > 0,
        "eval_credible": _bool(train, "claim_gate", "eval_credible"),
        "train_eval_hash_split_clean": int(split.get("train_eval_hash_overlap", train.get("split_report", {}).get("train_eval_hash_overlap", 1))) == 0,
        "train_eval_content_split_clean": int(split.get("train_eval_content_overlap", train.get("split_report", {}).get("train_eval_content_overlap", 1))) == 0,
        "raw_probe_uncontrolled": bool(raw) and not raw_controlled_enabled,
        "raw_probe_beats_baseline": raw_baseline_score > 0 and raw_native_score > raw_baseline_score,
        "controlled_repair_not_counted": not controlled_enabled or controlled.get("claim_gate", {}).get("raw_model_capability_claim") is False,
        "contamination_cleared_or_report_missing_blocks_claim": contamination_cleared,
        "official_or_hidden_eval_ready": _official_ready(official),
        "raw_outputs_or_rows_saved": bool(
            raw.get("rows")
            or raw.get("results")
            or raw.get("tasks")
            or raw.get("samples")
            or raw.get("native", {}).get("samples")
        ),
    }

    hard_blockers = [
        name
        for name in (
            "raw_probe_beats_baseline",
            "contamination_cleared_or_report_missing_blocks_claim",
            "official_or_hidden_eval_ready",
        )
        if not checks[name]
    ]
    report = {
        "schema_version": "tinymind-leaderboard-safe-improvement-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "assumptions": {
            "primary_goal": "leaderboard-safe improvement",
            "not_primary_goal": "offline public score maximization",
            "controlled_repair_is_runtime_capability": True,
            "controlled_repair_counts_as_raw_model_score": False,
        },
        "inputs": {
            "train_report": str(train_report) if train_report else None,
            "raw_probe_report": str(raw_probe_report) if raw_probe_report else None,
            "controlled_probe_report": str(controlled_probe_report) if controlled_probe_report else None,
            "official_report": str(official_report) if official_report else None,
            "contamination_report": str(contamination_report) if contamination_report else None,
        },
        "metrics": {
            "pre_eval_loss": pre_loss,
            "post_eval_loss": post_loss,
            "loss_delta_positive_means_lower_loss": loss_delta,
            "raw_native_score": raw_native_score,
            "raw_baseline_score": raw_baseline_score,
        },
        "checks": checks,
        "hard_blockers": hard_blockers,
        "claim_gate": {
            "leaderboard_safe_policy_active": True,
            "offline_public_score_only_allowed": False,
            "controlled_repair_score_can_promote_model": False,
            "training_progress_observed": checks["training_eval_improved"] and checks["eval_credible"],
            "leaderboard_safe_improvement_ready": all(checks.values()),
            "promotion_allowed": all(checks.values()),
            "world_best_claim_allowed": False,
            "reason": (
                "Leaderboard-safe promotion requires raw uncontrolled wins, contamination-cleared evidence, "
                "saved raw outputs, and hidden/official external readiness. Offline public scores alone are insufficient."
            ),
        },
    }
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    json_path = out / "leaderboard_safe_improvement_report.json"
    md_path = out / "leaderboard_safe_improvement_report.md"
    report["json_path"] = str(json_path)
    report["markdown_path"] = str(md_path)
    json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    md_path.write_text(_markdown(report), encoding="utf-8")
    return report


def _markdown(report: dict[str, Any]) -> str:
    lines = [
        "# TinyMind Leaderboard-Safe Improvement Gate",
        "",
        f"- Primary goal: {report['assumptions']['primary_goal']}",
        f"- Offline public score only allowed: {report['claim_gate']['offline_public_score_only_allowed']}",
        f"- Controlled repair can promote raw model: {report['claim_gate']['controlled_repair_score_can_promote_model']}",
        f"- Training progress observed: {report['claim_gate']['training_progress_observed']}",
        f"- Leaderboard-safe improvement ready: {report['claim_gate']['leaderboard_safe_improvement_ready']}",
        "",
        "## Checks",
        "",
    ]
    for key, value in report["checks"].items():
        lines.append(f"- {key}: {value}")
    if report["hard_blockers"]:
        lines.extend(["", "## Hard Blockers", ""])
        lines.extend(f"- {item}" for item in report["hard_blockers"])
    return "\n".join(lines) + "\n"