admin.py

"""Admin-tab handlers for the CADGenBench leaderboard Space.

Bundle 7: promote a row into the validated tier (recording the evidence
type) and demote it back. Gating is the ``CADGENBENCH_ADMINS`` Space
variable (comma-separated HF usernames); :func:`is_admin` is the single
predicate the UI uses to enable or disable the controls.

Row writes go through :func:`submit._hub_rmw_results`, so the
``_HUB_LOCK`` + read-modify-write semantics match the submit path
exactly. There is no second writer of ``results.jsonl`` with its own
locking story.
"""
from __future__ import annotations

import logging
import os
from typing import Any

import gradio as gr

from submit import _hub_rmw_results

logger = logging.getLogger(__name__)

ADMINS_ENV = "CADGENBENCH_ADMINS"

# The evidence types accepted on promotion. Mirrors the
# `validation_method` enum in cadgenbench-submissions/schema.md and the
# validation policy doc.
VALID_METHODS: tuple[str, ...] = ("code", "traces", "api", "manual")


def admin_usernames() -> set[str]:
    """Parse ``CADGENBENCH_ADMINS`` into a set of HF usernames.

    Comma-separated, whitespace-trimmed, empties dropped. Read fresh on
    each call so flipping the Space variable takes effect without a code
    deploy. Empty or unset yields an empty set, which means no one is an
    admin and the controls stay inert.
    """
    raw = os.environ.get(ADMINS_ENV, "")
    return {part.strip() for part in raw.split(",") if part.strip()}


def is_admin(profile: gr.OAuthProfile | None) -> bool:
    """Return whether *profile* is a logged-in user in the admin set.

    Logged-out users (``profile is None``) are never admins. With an
    empty admin set no profile qualifies, so the admin controls remain
    disabled for everyone until ``CADGENBENCH_ADMINS`` is populated.
    """
    if profile is None:
        return False
    return profile.username in admin_usernames()


def promote_row(submission_id: str, method: str) -> None:
    """Move a row into the validated tier with the given evidence type.

    Sets ``validation_status`` to ``"validated"`` and
    ``validation_method`` to *method*. Idempotent: re-promoting an
    already-validated row to the same method lands the same values.

    Raises:
        ValueError: *method* is not one of :data:`VALID_METHODS`.
        LookupError: no row in ``results.jsonl`` carries *submission_id*.
    """
    if method not in VALID_METHODS:
        raise ValueError(
            f"Unknown validation_method {method!r}; expected one of "
            f"{', '.join(VALID_METHODS)}."
        )

    def mutate(rows: list[dict[str, Any]]) -> None:
        for row in rows:
            if row.get("submission_id") == submission_id:
                row["validation_status"] = "validated"
                row["validation_method"] = method
                return
        raise LookupError(
            f"No row with submission_id={submission_id!r} in results.jsonl."
        )

    _hub_rmw_results(
        mutate,
        commit_message=f"promote {submission_id} to validated ({method})",
    )


def demote_row(submission_id: str) -> None:
    """Return a row to the unvalidated tier, clearing ``validation_method``.

    Sets ``validation_status`` to ``"unvalidated"`` and nulls
    ``validation_method``. Idempotent on an already-unvalidated row.

    Raises:
        LookupError: no row in ``results.jsonl`` carries *submission_id*.
    """
    def mutate(rows: list[dict[str, Any]]) -> None:
        for row in rows:
            if row.get("submission_id") == submission_id:
                row["validation_status"] = "unvalidated"
                row["validation_method"] = None
                return
        raise LookupError(
            f"No row with submission_id={submission_id!r} in results.jsonl."
        )

    _hub_rmw_results(
        mutate,
        commit_message=f"demote {submission_id} to unvalidated",
    )