evaluation/remote_artifact.py

"""Schema for evaluation results produced by Hugging Face Jobs and consumed by the Space."""

from __future__ import annotations

from typing import Any

ARTIFACT_SCHEMA_VERSION = 1


def default_remote_artifact_path(job_id: str) -> str:
    """Hub bucket path for a remote eval JSON artifact."""
    jid = (job_id or "").strip().replace("/", "_")
    return f"results/remote_artifacts/{jid}.json"


def normalize_artifact_bucket_path(artifact_ref: str) -> str:
    """
    Resolve a moderator-entered artifact reference to a bucket path.

    Accepts ``jobid.json``, ``jobid``, or ``results/remote_artifacts/jobid.json``.
    """
    ref = (artifact_ref or "").strip().replace("\\", "/").lstrip("/")
    if not ref:
        raise ValueError("Enter a JSON artifact file name or bucket path.")
    if ".." in ref.split("/"):
        raise ValueError("Invalid artifact path.")
    if "/" not in ref:
        ref = f"results/remote_artifacts/{ref}"
    if not ref.lower().endswith(".json"):
        ref = f"{ref}.json"
    return ref


def build_artifact(
    *,
    model_id: str,
    family_id: str,
    result: dict[str, Any],
    job_id: str = "",
    logs_tail: str = "",
) -> dict[str, Any]:
    """Wrap ``run_evaluation`` output for upload + validation on the Space."""
    return {
        "schema_version": ARTIFACT_SCHEMA_VERSION,
        "job_id": job_id,
        "model_id": model_id,
        "family_id": family_id,
        "result": dict(result),
        "logs_tail": (logs_tail or "")[-8000:],
    }


def extract_result_or_raise(data: dict[str, Any]) -> dict[str, Any]:
    """
    If the worker uploaded a failure envelope (``error`` field), raise ``RuntimeError``.
    Otherwise return validated ``result``.
    """
    if not isinstance(data, dict):
        raise ValueError("artifact must be a JSON object")
    err = data.get("error")
    if err:
        raise RuntimeError(str(err))
    return validate_remote_artifact(data)


def validate_remote_artifact(data: dict[str, Any]) -> dict[str, Any]:
    """
    Validate top-level remote artifact and return the inner ``result`` dict.

    Raises ``ValueError`` if the payload is not acceptable for leaderboard merge.
    """
    if not isinstance(data, dict):
        raise ValueError("artifact must be a JSON object")
    ver = data.get("schema_version")
    if ver != ARTIFACT_SCHEMA_VERSION:
        raise ValueError(f"unsupported schema_version: {ver!r} (expected {ARTIFACT_SCHEMA_VERSION})")
    result = data.get("result")
    if not isinstance(result, dict):
        raise ValueError("artifact.result must be an object")
    mid = str(result.get("model_id", "")).strip()
    if not mid:
        raise ValueError("artifact.result.model_id is required")
    if str(result.get("eval_family", "")).strip() == "":
        raise ValueError("artifact.result.eval_family is required")
    return result