app.py

"""Arabic LLM Leaderboard — Docker Space backend.

Serves the static single-page leaderboard (index.html) AND provides a gated
model-submission endpoint:

  GET  /                 -> index.html
  GET  /api/me           -> {"user": <hf-username|null>}
  GET  /login            -> redirect to Hugging Face OAuth
  GET  /login/callback   -> OAuth callback, sets the session
  GET  /logout           -> clears the session
  POST /api/submit       -> validate a model on the Hub + enqueue it (PENDING)

Submissions require the visitor to sign in with their own Hugging Face account
(OAuth). Their username is recorded on the request for provenance. The actual
write to the `requests` dataset is done with the Space's OWN token (a Space
secret, never exposed to the browser) — the submitter's token can't write to a
dataset they don't own.
"""
import datetime
import html
import json
import os
import re
import secrets
import threading
import time

import requests
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse
from typing import List, Optional

from pydantic import BaseModel
from starlette.middleware.sessions import SessionMiddleware

from huggingface_hub import HfApi
from huggingface_hub.utils import (
    GatedRepoError,
    HfHubHTTPError,
    RepositoryNotFoundError,
)

# ----------------------------------------------------------------------------- config
HF_CO = "https://huggingface.co"
HERE = os.path.dirname(os.path.abspath(__file__))

HF_TOKEN = os.environ.get("HF_TOKEN")  # Space secret with write access to REQUESTS_REPO
REQUESTS_REPO = os.environ.get("REQUESTS_REPO", "Mushari440/requests")
RESULTS_REPO = os.environ.get("RESULTS_REPO", "Mushari440/results")
# Private scores live in a separate PRIVATE dataset, never in the public one --
# hiding a row in the UI would still leave it readable on the Hub.
RESULTS_PRIVATE_REPO = os.environ.get("RESULTS_PRIVATE_REPO", "Mushari440/results-private")
VISIBILITIES = {"public", "private"}

# Injected by HF when `hf_oauth: true` is set in the README frontmatter.
OAUTH_CLIENT_ID = os.environ.get("OAUTH_CLIENT_ID")
OAUTH_CLIENT_SECRET = os.environ.get("OAUTH_CLIENT_SECRET")
OAUTH_SCOPES = os.environ.get("OAUTH_SCOPES", "openid profile")
OPENID_PROVIDER_URL = os.environ.get("OPENID_PROVIDER_URL", HF_CO).rstrip("/")
SPACE_HOST = os.environ.get("SPACE_HOST", "")  # e.g. mushari440-benchmark.hf.space

# Closed-source models are served through a paid API billed to OUR key, so only
# the leaderboard owner may queue them. Open-weight models run on our own GPUs and
# cost nothing, so anyone signed in may submit those.
LEADERBOARD_OWNER = os.environ.get("LEADERBOARD_OWNER", "Mushari440")
SOURCES = {"open", "closed"}
# Guard rails for the owner-only partial-run controls.
MAX_EXAMPLES_CAP = int(os.environ.get("MAX_EXAMPLES_CAP", "1000"))
SUBTASK_RE = re.compile(r"^[a-z0-9_]{2,64}$")

MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$")
REVISION_RE = re.compile(r"^[A-Za-z0-9._/-]{1,120}$")
PRECISIONS = {"bfloat16", "float16"}
WEIGHT_TYPES = {"Original", "Adapter", "Delta"}
TYPE_MAP = {
    "fine-tuned": "🔶 fine-tuned",
    "instruction-tuned": "⭕ instruction-tuned",
    "pretrained": "🟢 pretrained",
    "RL-tuned": "🟦 RL-tuned",
}
BLOCK_RESUBMIT = {"PENDING", "RUNNING", "FINISHED"}  # FAILED is allowed to be re-submitted

api = HfApi(endpoint=HF_CO, token=HF_TOKEN)  # writes to REQUESTS_REPO + reads the queue
# Validate submitted models as an ANONYMOUS client sees them, so the Space's write token is
# never used as a "confused deputy" to read repos the public can't (private/gated-for-owner).
api_public = HfApi(endpoint=HF_CO, token=None)

_write_lock = threading.Lock()  # serialize dedupe-check + write within this process
# Short-TTL in-process guards to close the eventually-consistent-read gap (the queue read via
# the CDN lags a few seconds behind a just-written commit).
_RECENT_TTL = 120  # seconds
_recent_keys = {}  # (model, precision) -> monotonic ts  (just-submitted, not yet visible)


def _prune_recent(now):
    for k in [k for k, t in _recent_keys.items() if now - t > _RECENT_TTL]:
        del _recent_keys[k]

# ----------------------------------------------------------------------------- app
app = FastAPI(title="Arabic LLM Leaderboard")
# SameSite=None + Secure so the session cookie also works when the Space is embedded
# in an iframe on huggingface.co. Signed with the OAuth client secret (stable, private).
# Sign the session with the OAuth client secret (stable + private). If it is somehow absent,
# fall back to a RANDOM per-process key — forged cookies become impossible (the only cost is
# that sessions don't survive a restart). Never a hardcoded constant (would be forgeable).
app.add_middleware(
    SessionMiddleware,
    secret_key=OAUTH_CLIENT_SECRET or secrets.token_hex(32),
    same_site="none",
    https_only=True,
    max_age=8 * 60 * 60,
)

_oauth = None
if OAUTH_CLIENT_ID and OAUTH_CLIENT_SECRET:
    from authlib.integrations.starlette_client import OAuth

    _oauth = OAuth()
    _oauth.register(
        name="hf",
        client_id=OAUTH_CLIENT_ID,
        client_secret=OAUTH_CLIENT_SECRET,
        server_metadata_url=f"{OPENID_PROVIDER_URL}/.well-known/openid-configuration",
        client_kwargs={"scope": OAUTH_SCOPES},
    )


def _redirect_uri() -> str:
    return f"https://{SPACE_HOST}/login/callback"


def _origin_ok(request: Request) -> bool:
    """CSRF guard: the write must originate from our own page. Both the direct host and the
    huggingface.co iframe embed serve the page from https://{SPACE_HOST}, so that single origin
    is sufficient (and huggingface.co itself never posts here)."""
    origin = request.headers.get("origin") or ""
    return bool(SPACE_HOST) and origin == f"https://{SPACE_HOST}"


def _list_requests():
    """All request dicts currently in the queue (live read via the raw file API)."""
    headers = {"Authorization": f"Bearer {HF_TOKEN}"} if HF_TOKEN else {}
    out = []
    for path in api.list_repo_files(REQUESTS_REPO, repo_type="dataset"):
        if not path.endswith(".json"):
            continue
        try:
            r = requests.get(
                f"{HF_CO}/datasets/{REQUESTS_REPO}/raw/main/{path}",
                headers=headers,
                timeout=20,
            )
            r.raise_for_status()
            data = r.json()
        except (requests.RequestException, ValueError):
            continue
        if isinstance(data, dict) and "model" in data:
            # Carry the repo path: the manage endpoints need it to delete or
            # rewrite the exact file, and it cannot be reconstructed reliably
            # (the filename encodes precision/weight_type as submitted).
            data["_path"] = path
            out.append(data)
    return out


# ----------------------------------------------------------------------------- routes
@app.get("/", include_in_schema=False)
def index():
    return FileResponse(os.path.join(HERE, "index.html"))

@app.get("/healthz", include_in_schema=False)
def healthz():
    return {"ok": True, "oauth": bool(_oauth)}

@app.get("/api/me")
def me(request: Request):
    user = request.session.get("user")
    # `is_owner` only decides whether the UI OFFERS the closed-source option.
    # /api/submit re-checks server-side, so faking this client-side gains nothing.
    return {"user": user, "is_owner": bool(user) and user == LEADERBOARD_OWNER}

@app.get("/login", include_in_schema=False)
async def login(request: Request):
    if _oauth is None:
        raise HTTPException(503, "Sign-in is not configured on this Space yet.")
    return await _oauth.hf.authorize_redirect(request, _redirect_uri())

@app.get("/login/callback", include_in_schema=False)
async def login_callback(request: Request):
    if _oauth is None:
        raise HTTPException(503, "Sign-in is not configured on this Space yet.")
    try:
        token = await _oauth.hf.authorize_access_token(request)
    except Exception as e:  # OAuthError (incl. attacker-supplied ?error=), state/nonce, network
        # NEVER reflect the exception into the page: authorize_access_token raises from the
        # attacker-controlled `error_description` query param BEFORE any state check, so echoing
        # it would be a reflected-XSS sink. Log server-side only; show a static message.
        print(f"[oauth] sign-in failed: {type(e).__name__}: {e}")
        return HTMLResponse(
            "<!doctype html><meta charset=utf-8>"
            "<p style='font-family:system-ui'>Sign-in failed or was cancelled. "
            "<a href='/'>Return to the leaderboard</a>.</p>",
            status_code=400,
        )
    info = token.get("userinfo") or {}
    if not info:
        try:
            info = await _oauth.hf.userinfo(token=token)
        except Exception:
            info = {}
    username = info.get("preferred_username") or info.get("name") or info.get("sub")
    if not username:
        return HTMLResponse("<p>Could not read your Hugging Face username.</p>", status_code=400)
    request.session["user"] = str(username)
    safe_user = html.escape(str(username))  # display-name fallback can contain markup
    # Small page so the flow also works when opened in a new tab from an iframe.
    return HTMLResponse(
        "<!doctype html><meta charset=utf-8>"
        "<body style='font-family:system-ui;background:#071411;color:#eaf3ef;"
        "display:flex;align-items:center;justify-content:center;height:100vh;margin:0'>"
        f"<div style='text-align:center'><p>Signed in as <b>{safe_user}</b>.</p>"
        "<p><a style='color:#34d399' href='/'>Return to the leaderboard &rarr;</a></p>"
        "<script>try{if(window.opener){window.opener.focus();window.close();}"
        "else{location.href='/?signedin=1';}}catch(e){location.href='/?signedin=1';}</script>"
        "</div></body>"
    )

@app.get("/logout", include_in_schema=False)
def logout(request: Request):
    request.session.pop("user", None)
    return RedirectResponse("/", status_code=302)


def _require_owner(request: Request, check_origin: bool = True) -> str:
    """Every management route funnels through here. The identity comes from the
    signed session, so a crafted request cannot impersonate the owner.

    check_origin is the CSRF guard and applies to STATE-CHANGING requests only.
    Browsers omit the Origin header on same-origin GETs, so enforcing it on a read
    endpoint rejects our own page. Reads are still session-gated, and a cross-origin
    read cannot see the response anyway (no CORS headers are served)."""
    user = request.session.get("user")
    if not user:
        raise HTTPException(401, "Please sign in with Hugging Face.")
    if user != LEADERBOARD_OWNER:
        raise HTTPException(403, "Only the leaderboard maintainer can manage models.")
    if check_origin and not _origin_ok(request):
        raise HTTPException(403, "Request blocked (bad origin).")
    if not HF_TOKEN:
        raise HTTPException(503, "The Space has no write token configured.")
    return user


# ── Benchmark versions ───────────────────────────────────────────────────────
# Results are namespaced per version: <version>/<org>/results_<model>.json.
# LEGACY_VERSION also owns the un-prefixed files at the repo root, which is where
# every score lived before versioning. Keep this list in step with VERSIONS in
# index.html.
VERSIONS = [v.strip() for v in os.environ.get("VERSIONS", "v7,v8").split(",") if v.strip()]
LEGACY_VERSION = os.environ.get("LEGACY_VERSION", "v7")


def _clean_version(version: str | None) -> str:
    v = (version or LEGACY_VERSION).strip()
    if v not in VERSIONS:
        raise HTTPException(400, f"Unknown benchmark version {v!r}.")
    return v


def _results_path(model: str, version: str | None = None) -> str:
    org, _, name = model.partition("/")
    return f"{_clean_version(version)}/{org}/results_{name or org}.json"


def _results_paths(model: str, version: str | None = None) -> list[str]:
    """Every path a score for this model may sit at, newest layout first. The bare
    path is only in play for the legacy version, and only until migration finishes."""
    v = _clean_version(version)
    org, _, name = model.partition("/")
    paths = [f"{v}/{org}/results_{name or org}.json"]
    if v == LEGACY_VERSION:
        paths.append(f"{org}/results_{name or org}.json")
    return paths


class ModelRef(BaseModel):
    model: str
    version: str | None = None
    # "version" removes only this version's score and leaves the queue entry, so the
    # model can be re-evaluated. "all" removes the queue entry and every version's
    # score. Deleting a v8 score used to destroy the shared request too, which took
    # the v7 row's metadata with it and left no way to re-run without resubmitting.
    scope: str = "version"


@app.get("/api/manage")
def manage(request: Request, version: str | None = None):
    """Every request + where its scores live, per benchmark version. Owner-only."""
    _require_owner(request, check_origin=False)   # read-only GET
    api = HfApi(token=HF_TOKEN)
    have = {}
    for repo, key in ((RESULTS_REPO, "public"), (RESULTS_PRIVATE_REPO, "private")):
        try:
            for f in api.list_repo_files(repo, repo_type="dataset"):
                if f.endswith(".json"):
                    have.setdefault(f, set()).add(key)
        except Exception:
            pass
    rows = []
    for r in _list_requests():
        m = r.get("model", "")
        # Presence per version, so the panel can show at a glance which boards a
        # model is actually on. `result_in` stays for the selected version because
        # the older UI read it.
        by_version = {}
        for v in VERSIONS:
            where = set()
            for path in _results_paths(m, v):
                where |= have.get(path, set())
            if where:
                by_version[v] = sorted(where)
        rows.append({
            "model": m,
            "status": r.get("status"),
            "submitted_time": r.get("submitted_time"),
            "submitted_by": r.get("submitted_by"),
            "source": r.get("source", "open"),
            "visibility": r.get("visibility", "public"),
            "result_in": by_version.get(_clean_version(version), []),
            "results_by_version": by_version,
        })
    rows.sort(key=lambda x: x.get("submitted_time") or "", reverse=True)
    return {"models": rows, "owner": LEADERBOARD_OWNER,
            "versions": VERSIONS, "version": _clean_version(version)}


@app.post("/api/model/delete")
def model_delete(request: Request, body: ModelRef):
    """Remove a model's request AND its score from both results repos."""
    _require_owner(request)
    model = (body.model or "").strip()
    if not MODEL_RE.match(model):
        raise HTTPException(400, "Bad model id.")
    api = HfApi(token=HF_TOKEN)
    scope = (body.scope or "version").strip().lower()
    if scope not in ("version", "all"):
        raise HTTPException(400, "scope must be 'version' or 'all'.")
    removed = []
    if scope == "all":
        for r in _list_requests():
            if r.get("model") == model and r.get("_path"):
                try:
                    api.delete_file(path_in_repo=r["_path"], repo_id=REQUESTS_REPO,
                                    repo_type="dataset",
                                    commit_message=f"Remove request {model}")
                    removed.append(f"request:{r['_path']}")
                except Exception as e:
                    print(f"[delete] request {model}: {type(e).__name__}: {e}")
    version = _clean_version(body.version)
    targets = ([p for v in VERSIONS for p in _results_paths(model, v)]
               if scope == "all" else _results_paths(model, version))
    for repo in (RESULTS_REPO, RESULTS_PRIVATE_REPO):
        for path in dict.fromkeys(targets):
            try:
                api.delete_file(path_in_repo=path, repo_id=repo,
                                repo_type="dataset",
                                commit_message=f"Remove {version} results for {model}")
                removed.append(f"results:{repo}:{path}")
            except Exception:
                pass      # simply not present there
    if not removed:
        raise HTTPException(
            404,
            f"{model} has no score on {version}." if scope == "version"
            else f"Nothing found for {model}.")
    what = f"every version of {model}" if scope == "all" else f"{model} from {version}"
    return {"ok": True, "removed": removed,
            "message": f"Deleted {what} ({len(removed)} file(s))."}


@app.post("/api/model/rerun")
def model_rerun(request: Request, body: ModelRef):
    """Queue a model for re-evaluation. RERUN also clears the worker's done-log,
    which a plain PENDING does not -- that is why a re-submit alone never ran."""
    _require_owner(request)
    model = (body.model or "").strip()
    api = HfApi(token=HF_TOKEN)
    hit = None
    for r in _list_requests():
        if r.get("model") == model:
            hit = r
            break
    if not hit:
        raise HTTPException(404, f"{model} is not in the queue.")
    if not hit.get("_path"):
        raise HTTPException(500, "Could not resolve the request file path.")
    payload = {k: v for k, v in hit.items() if k != "_path"}
    payload["status"] = "RERUN"
    api.upload_file(
        path_or_fileobj=json.dumps(payload, ensure_ascii=False, indent=1).encode("utf-8"),
        path_in_repo=hit["_path"], repo_id=REQUESTS_REPO, repo_type="dataset",
        commit_message=f"{model} -> RERUN")
    return {"ok": True,
            "message": f"{model} queued for re-evaluation (RERUN). It lands on "
                       f"whichever version the worker is running (its RUN_ID)."}


# Statuses a requeue may revive. A model that ran, or tried and failed, can run
# again. Everything else was set deliberately and a sweep must not quietly undo it:
# HELD is parked on purpose, and CANCELLED / SUPERSEDED / REJECTED were excluded on
# purpose -- nvidia/Qwen3.6-35B-A3B-NVFP4 is CANCELLED because the A100 cannot run
# NVFP4 at all, so reviving it would fail on every sweep from here on.
REQUEUABLE = {"FINISHED", "FAILED"}
ALREADY_QUEUED = {"PENDING", "RUNNING", "RERUN"}


class RequeueRef(BaseModel):
    version: str | None = None
    # Default to only what is missing, so re-running a half-finished sweep does not
    # discard the models that already landed.
    only_missing: bool = True


@app.post("/api/requeue-all")
def requeue_all(request: Request, body: RequeueRef):
    """Set every queued model to RERUN so a whole version can be evaluated.

    The worker only ever picks up PENDING / RUNNING / RERUN, so after a sweep every
    model sits at FINISHED and a new version would evaluate nothing at all. This is
    the switch that starts the next version's run."""
    _require_owner(request)
    version = _clean_version(body.version)
    api = HfApi(token=HF_TOKEN)

    have = set()
    for repo in (RESULTS_REPO, RESULTS_PRIVATE_REPO):
        try:
            have |= {f for f in api.list_repo_files(repo, repo_type="dataset")
                     if f.endswith(".json")}
        except Exception:
            pass

    queued, skipped = [], {}
    for r in _list_requests():
        model = r.get("model")
        if not model or not r.get("_path"):
            continue
        status = str(r.get("status") or "").upper()
        if status in ALREADY_QUEUED:
            skipped[model] = "already queued"
            continue
        if status not in REQUEUABLE:
            skipped[model] = f"{status.lower()} on purpose"
            continue
        if body.only_missing and any(p in have for p in _results_paths(model, version)):
            skipped[model] = f"already scored on {version}"
            continue
        payload = {k: v for k, v in r.items() if k != "_path"}
        payload["status"] = "RERUN"
        try:
            api.upload_file(
                path_or_fileobj=json.dumps(payload, ensure_ascii=False, indent=1).encode("utf-8"),
                path_in_repo=r["_path"], repo_id=REQUESTS_REPO, repo_type="dataset",
                commit_message=f"{model} -> RERUN (requeue for {version})")
            queued.append(model)
        except Exception as e:
            skipped[model] = f"upload failed ({type(e).__name__})"
            print(f"[requeue-all] {model}: {type(e).__name__}: {e}")
    reasons: dict[str, int] = {}
    for why in skipped.values():
        reasons[why] = reasons.get(why, 0) + 1
    tail = ("  Left alone: "
            + ", ".join(f"{n} {why}" for why, n in sorted(reasons.items()))) if reasons else ""
    msg = (f"Queued {len(queued)} model(s) for {version}.{tail}" if queued
           else f"Nothing to queue for {version}.{tail}")
    return {"ok": True, "queued": queued, "skipped": skipped,
            "skipped_reasons": reasons, "message": msg}


@app.get("/api/private-results")
def private_results(request: Request, version: str | None = None):
    """The private scores, served only to the owner. The browser cannot read that
    dataset directly, so this endpoint is the only way they reach the board."""
    _require_owner(request, check_origin=False)   # read-only GET
    api = HfApi(token=HF_TOKEN)
    out = []
    v = _clean_version(version)
    try:
        files = [f for f in api.list_repo_files(RESULTS_PRIVATE_REPO, repo_type="dataset")
                 # `_`-prefixed files are side-car metrics, not model rows
                 if f.endswith(".json") and not f.split("/")[-1].startswith("_")]
    except Exception:
        files = []

    def _in_version(path: str) -> bool:
        if path.startswith(f"{v}/"):
            return True
        # Un-prefixed root files predate versioning and belong to the legacy board.
        return (v == LEGACY_VERSION and path.count("/") == 1
                and not any(path.startswith(f"{o}/") for o in VERSIONS))

    seen, keep = set(), []
    for f in sorted(files, key=lambda f: not f.startswith(f"{v}/")):
        if not _in_version(f):
            continue
        stem = f.split("/")[-1]
        if stem in seen:          # prefixed copy wins over a not-yet-removed root one
            continue
        seen.add(stem)
        keep.append(f)
    files = keep
    for f in files:
        try:
            url = f"{HF_CO}/datasets/{RESULTS_PRIVATE_REPO}/raw/main/{f}"
            r = requests.get(url, headers={"Authorization": f"Bearer {HF_TOKEN}"}, timeout=20)
            r.raise_for_status()
            d = r.json()
            d["_private"] = True
            out.append(d)
        except Exception as e:
            print(f"[private-results] {f}: {type(e).__name__}: {e}")
    return {"results": out}


@app.get("/api/private-metrics")
def private_metrics(request: Request, version: str | None = None):
    """Owner-only extra columns: policy A/B scores and the judge timestamp.

    Kept in the PRIVATE dataset, not the public results repo, so the numbers are
    not merely hidden in the UI -- an anonymous reader cannot fetch them at all.
    Returns {} rather than raising when the file is absent, so a version that has
    no side-car simply renders without the extra columns.
    """
    _require_owner(request, check_origin=False)   # read-only GET
    v = _clean_version(version)
    url = f"{HF_CO}/datasets/{RESULTS_PRIVATE_REPO}/raw/main/{v}/_metrics_extra.json"
    try:
        r = requests.get(url, headers={"Authorization": f"Bearer {HF_TOKEN}"}, timeout=20)
        if r.status_code == 404:
            return {"models": {}}
        r.raise_for_status()
        return r.json()
    except Exception as e:
        print(f"[private-metrics] {v}: {type(e).__name__}: {e}")
        return {"models": {}}


class SubmitBody(BaseModel):
    model: str
    revision: str = "main"
    precision: str = "bfloat16"
    model_type: str = "fine-tuned"
    weight_type: str = "Original"
    base_model: str = ""
    # "open"   -> open-weight repo on the Hub, run locally on our GPUs (anyone)
    # "closed" -> served via the OpenRouter API, billed to us (owner only)
    source: str = "open"
    # Partial-run controls, OWNER ONLY (see the gate in submit()). A capped or
    # subtask-filtered score is not comparable with the full-corpus board, so these
    # must never be settable by an outside submitter.
    max_examples: Optional[int] = None
    subtasks: List[str] = []
    # "public"  -> score published to the public board (default, anyone)
    # "private" -> score goes to the private dataset, visible only to the owner
    visibility: str = "public"


@app.post("/api/submit")
def submit(request: Request, body: SubmitBody):
    user = request.session.get("user")
    if not user:
        raise HTTPException(401, "Please sign in with Hugging Face before submitting.")
    if not _origin_ok(request):
        raise HTTPException(403, "Request blocked (bad origin).")
    if not HF_TOKEN:
        raise HTTPException(503, "The Space has no write token configured; submissions are disabled.")

    model = (body.model or "").strip()
    if not MODEL_RE.match(model):
        raise HTTPException(400, "Model id must look like 'org/name' (letters, digits, . _ - only).")
    if body.precision not in PRECISIONS:
        raise HTTPException(400, "Precision must be 'bfloat16' or 'float16'.")
    precision = body.precision
    if body.weight_type not in WEIGHT_TYPES:
        raise HTTPException(400, "Weight type must be Original, Adapter, or Delta.")
    weight_type = body.weight_type
    if body.model_type not in TYPE_MAP:
        raise HTTPException(400, "Unknown model type.")
    revision = (body.revision or "main").strip() or "main"
    if not REVISION_RE.match(revision):
        raise HTTPException(400, "Invalid revision.")

    # ── THE GATE ────────────────────────────────────────────────────────────
    # Authoritative because `user` comes from the signed OAuth session, not the
    # request body. A hand-crafted POST cannot get past this.
    source = (body.source or "open").strip().lower()
    if source not in SOURCES:
        raise HTTPException(400, "Source must be 'open' or 'closed'.")
    # Partial-run controls: owner-only, and rejected loudly rather than silently
    # dropped, so a non-owner is never misled into thinking they got a quick run.
    max_examples = body.max_examples
    subtasks = [str(s).strip() for s in (body.subtasks or []) if str(s).strip()]
    visibility = (body.visibility or "public").strip().lower()
    if visibility not in VISIBILITIES:
        raise HTTPException(400, "Visibility must be 'public' or 'private'.")
    if visibility == "private" and user != LEADERBOARD_OWNER:
        raise HTTPException(
            403, "Private evaluations are restricted to the leaderboard maintainer.")

    if (max_examples is not None or subtasks) and user != LEADERBOARD_OWNER:
        raise HTTPException(
            403,
            "Limiting the item count or the subtask list is restricted to the "
            "leaderboard maintainer — a partial run does not produce a score "
            "comparable with the models already on the board.",
        )
    if max_examples is not None:
        if not (1 <= max_examples <= MAX_EXAMPLES_CAP):
            raise HTTPException(400, f"max_examples must be between 1 and {MAX_EXAMPLES_CAP}.")
    if subtasks:
        bad = [s for s in subtasks if not SUBTASK_RE.match(s)]
        if bad:
            raise HTTPException(400, f"Invalid subtask id(s): {bad[:5]}")
        subtasks = sorted(set(subtasks))

    if source == "closed" and user != LEADERBOARD_OWNER:
        raise HTTPException(
            403,
            "Closed-source (API) models can only be submitted by the leaderboard "
            "maintainer, because they are billed per token. Please submit an "
            "open-weight model from the Hub instead.",
        )

    # Closed-source models are API slugs (openai/gpt-5.2), not Hub repos, so every
    # check below would 404 on them. The owner gate above already ran; the worker
    # validates the slug against OpenRouter at run time.
    if source == "closed":
        params, likes, lic = None, 0, "proprietary"
    else:
        # 1) The model must exist and be a public, non-gated, loadable Transformers repo. Validate
        #    with the ANONYMOUS client so we only ever see what the public sees.
        try:
            info = api_public.model_info(model, revision=revision, files_metadata=False)
        except GatedRepoError:
            raise HTTPException(400, "That model is gated — the evaluator can't download it. Submit a public model.")
        except RepositoryNotFoundError:
            raise HTTPException(404, f"'{model}' was not found on the Hub (it must be a public model repo).")
        except HfHubHTTPError:
            raise HTTPException(400, f"Could not read '{model}' at revision '{revision}' as a public model.")

        if getattr(info, "private", False):
            raise HTTPException(400, "That model is private. Submit a public model.")
        if getattr(info, "gated", False):
            raise HTTPException(400, "That model is gated. Submit a non-gated public model.")
        siblings = [getattr(s, "rfilename", "") for s in (getattr(info, "siblings", None) or [])]
        if "config.json" not in siblings:
            raise HTTPException(400, "That repo has no config.json — it doesn't look like a Transformers model the evaluator can run.")

        # parameter count (billions), best-effort
        params = None
        st = getattr(info, "safetensors", None)
        total = getattr(st, "total", None) if st else None
        if isinstance(total, (int, float)) and total > 0:
            params = round(total / 1e9, 3)
        if params is None:
            m = re.search(r"(\d+(?:\.\d+)?)\s*[bB]\b", model)
            if m:
                params = float(m.group(1))

        likes = int(getattr(info, "likes", 0) or 0)
        card = getattr(info, "card_data", None)
        lic = None
        if card is not None:
            try:
                lic = card.get("license")
            except Exception:
                lic = getattr(card, "license", None)
        lic = lic or "?"

    org, _, name = model.partition("/")
    path_in_repo = f"{org}/{name}_eval_request_False_{precision}_{weight_type}.json"

    key = (model, precision)
    with _write_lock:
        now_m = time.monotonic()
        _prune_recent(now_m)

        # 2) De-duplicate: don't re-queue something already pending/running/finished. Check both
        #    the (eventually-consistent) remote queue AND just-written keys held in-process.
        existing = _list_requests()
        for r in existing:
            if r.get("model") == model and r.get("precision") == precision \
                    and (r.get("status", "") or "").upper() in BLOCK_RESUBMIT:
                raise HTTPException(
                    409,
                    f"{model} ({precision}) is already {str(r.get('status')).lower()} in the queue.",
                )
        if key in _recent_keys:
            raise HTTPException(409, f"{model} ({precision}) was just submitted — it is already queued.")
        # No cap on how many distinct models a user may queue — only the same model+precision
        # is blocked from being queued twice (the dedupe check above).

        now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        payload = {
            "model": model,
            "base_model": (body.base_model or "").strip(),
            "revision": revision,
            "precision": precision,
            "weight_type": weight_type,
            "status": "PENDING",
            "submitted_time": now,
            "model_type": TYPE_MAP[body.model_type],
            "likes": likes,
            "params": params,  # float billions, or null when genuinely unknown
            "license": lic,
            "private": False,
            "submitted_by": user,
            # Read by the worker: "closed" routes to the OpenRouter API path (no
            # download, no GPU). Absent or "open" keeps the existing GPU route, so
            # every request queued before this change behaves exactly as before.
            "source": source,
            # Owner-only partial-run controls; absent on a normal full submission.
            # The worker re-checks the submitter before honouring them.
            **({"max_examples": max_examples} if max_examples is not None else {}),
            **({"subtasks": subtasks} if subtasks else {}),
            # Read by the worker: routes the score to the private dataset.
            **({"visibility": visibility} if visibility != "public" else {}),
        }
        try:
            api.upload_file(
                path_or_fileobj=json.dumps(payload, ensure_ascii=False, indent=1).encode("utf-8"),
                path_in_repo=path_in_repo,
                repo_id=REQUESTS_REPO,
                repo_type="dataset",
                commit_message=f"Submit {model} ({precision}) — by {user}",
            )
        except Exception as e:
            print(f"[submit] upload failed for {model}: {type(e).__name__}: {e}")
            raise HTTPException(502, "Failed to enqueue the submission. Please try again shortly.")

        # Record in-process so an immediate duplicate can't slip past the remote read lag.
        _recent_keys[key] = now_m

    return JSONResponse(
        {
            "ok": True,
            "message": f"Submitted {model} — it is now PENDING. The worker will evaluate it and it will appear on the board.",
            "params": params,
            "path": path_in_repo,
        }
    )