from __future__ import annotations import io import hashlib import json import logging import time from collections import deque from datetime import datetime, timezone from pathlib import Path from threading import RLock, BoundedSemaphore from typing import Any # Import Spaces before Gradio so ZeroGPU can initialize its integration first. import spaces import gradio as gr from fastapi import HTTPException, Request from fastapi.responses import HTMLResponse, Response, JSONResponse from starlette.concurrency import run_in_threadpool from starlette.datastructures import UploadFile from src.leaderboard.build_rankings import build_rankings, frame_for_download from src.leaderboard.schema import ACTIVE_BENCHMARK_VERSION, METRIC_OPTIONS, LABELS logging.basicConfig(level=logging.INFO) LOGGER = logging.getLogger("ark-leaderboard") ROOT = Path(__file__).parent STATIC_DIR = ROOT / "static" SCORES_DIR = ROOT / "scores" DATA_LOCK = RLock() DATA: dict[str, Any] = {} VALID_METRICS = {(family, metric) for family, metric, _ in METRIC_OPTIONS} EVALUATION_SERVICE = None EVALUATION_LOCK = RLock() SUBMISSION_SLOT = BoundedSemaphore(1) SUBMISSION_ATTEMPTS: deque[float] = deque() # Only explicitly decorated functions may request GPU time. Ordinary browsing # must remain CPU-only even if the hosting environment enables auto-wrapping. spaces.disable_gradio_auto_wrap() def _empty_payload(message: str | None = None) -> dict[str, Any]: return { "views": {"overview": [], "knowledge_domain": [], "cognitive_demand": []}, "metric_options": [{"family": f, "metric": m, "label": l} for f, m, l in METRIC_OPTIONS], "active_benchmark_version": ACTIVE_BENCHMARK_VERSION, "updated_at": datetime.now(timezone.utc).isoformat(), "copy": { "title": "ARK-Bench Leaderboard", "description": "Knowledge- and reasoning-intensive multimodal retrieval benchmark for embedding models.", "about": ( "## About ARK-Bench\n\n" "ARK-Bench evaluates multimodal embedding models on 17 knowledge subtypes and six reasoning skills. " "Knowledge macro is the equal-weight average over the 17 subtypes; reasoning macro is the " "equal-weight average over KR, SR, FGVR, LR, SyR, and CA.\n\n" "Overview Avg is the leaderboard-defined `(Knowledge Macro + Reasoning Macro) / 2`. Cognitive Demand " "defaults to the hidden equal-weight mean of PO, KI, RI, and KRI.\n\n" "Current benchmark version: 1.0. Read the [paper](https://arxiv.org/pdf/2602.09839) for the protocol." ), "submission": ( "## Submit retrieval predictions\n\n" "1. Download the template and rank 20 candidates from each query's topic Gallery.\n" "2. Select Evaluate to calculate and inspect aggregate scores against hidden labels.\n" "3. Submit the report to notify maintainers through a private repository Issue.\n" "4. A maintainer reviews the report before publication on the leaderboard." ), }, "error": message, } def _build_payload(family: str, metric: str) -> dict[str, Any]: ranking = build_rankings(SCORES_DIR, family, metric, ACTIVE_BENCHMARK_VERSION) return { "views": {key: ranking[key] for key in ("overview", "knowledge_domain", "reasoning_skill", "cognitive_demand")}, "errors": ranking["errors"], "labels": LABELS, "score_date": max((row["date"] for row in ranking["overview"]), default=None), } def refresh_data() -> dict[str, Any]: """Load reviewed score files and build the default leaderboard views.""" payload = _empty_payload() try: SCORES_DIR.mkdir(parents=True, exist_ok=True) result = _build_payload("Recall", "R@1") payload.update(result) if result["errors"]: payload["error"] = "; ".join(result["errors"]) except Exception as exc: LOGGER.exception("Could not build leaderboard data") payload["error"] = f"Could not build leaderboard data: {exc}" with DATA_LOCK: DATA.clear() DATA.update(payload) return dict(DATA) server = gr.Server(title="ARK-Bench Leaderboard", description="ARK-Bench embedding model leaderboard.", version="1.0.0") def evaluation_service(): global EVALUATION_SERVICE with EVALUATION_LOCK: if EVALUATION_SERVICE is None: from src.evaluation.service import EvaluationService EVALUATION_SERVICE = EvaluationService(ROOT) return EVALUATION_SERVICE @server.get("/api/evaluation/protocol") def evaluation_protocol(): return JSONResponse(evaluation_service().protocol(), headers={"Cache-Control": "no-store"}) @server.get("/api/evaluation/template") def evaluation_template(): return Response(evaluation_service().template(), media_type="application/x-ndjson", headers={ "Content-Disposition": 'attachment; filename="ark-predictions-template.jsonl"', "Cache-Control": "no-store", "X-Content-Type-Options": "nosniff", }) async def _read_evaluation_form(request: Request, *, include_receipt: bool = False): from src.evaluation.service import MAX_UPLOAD_BYTES if not request.headers.get("content-type", "").lower().startswith("multipart/form-data;"): raise HTTPException(415, "Upload predictions using a multipart form.") body = bytearray() async for chunk in request.stream(): body.extend(chunk) if len(body) > MAX_UPLOAD_BYTES + 65536: raise HTTPException(413, "Upload exceeds the 2 MiB file limit.") request._body = bytes(body) async with request.form(max_files=1, max_fields=6 if include_receipt else 5, max_part_size=16384) as form: fields = {} upload = None allowed = {"model_name", "model_url", "params_b", "benchmark_version", "dataset_revision"} if include_receipt: allowed.add("evaluation_id") seen = set() for key, value in form.multi_items(): if key in seen: raise HTTPException(422, "Duplicate form fields are not allowed.") seen.add(key) if key == "file" and isinstance(value, UploadFile): upload = value elif key in allowed and isinstance(value, str): fields[key] = value else: raise HTTPException(422, "Unexpected submission field.") if upload is None: raise HTTPException(422, "Select a predictions JSONL file.") content = await upload.read(MAX_UPLOAD_BYTES + 1) if len(content) > MAX_UPLOAD_BYTES: raise HTTPException(413, "Upload exceeds the 2 MiB file limit.") return content, fields @server.post("/api/evaluation/evaluate") async def evaluate_predictions(request: Request): from src.evaluation.scoring import SubmissionError from src.evaluation.service import EvaluationUnavailable if not SUBMISSION_SLOT.acquire(blocking=False): raise HTTPException(429, "Another submission is being evaluated. Please retry shortly.", headers={"Retry-After": "10"}) try: # A global ceiling bounds anonymous scoring probes and private repository writes. # This is an in-process guard; use authenticated quotas for larger competitions. now = time.monotonic() while SUBMISSION_ATTEMPTS and SUBMISSION_ATTEMPTS[0] <= now - 3600: SUBMISSION_ATTEMPTS.popleft() if len(SUBMISSION_ATTEMPTS) >= 30: raise HTTPException(429, "The evaluation service has reached its hourly submission limit. Please try again later.", headers={"Retry-After": str(max(1, int(3600 - now + SUBMISSION_ATTEMPTS[0])))}) content, fields = await _read_evaluation_form(request) SUBMISSION_ATTEMPTS.append(now) result = await run_in_threadpool(evaluation_service().evaluate, content, fields) return JSONResponse(result, headers={"Cache-Control": "no-store"}) except SubmissionError as exc: raise HTTPException(422, str(exc)) from None except EvaluationUnavailable as exc: raise HTTPException(503, str(exc)) from None except HTTPException: raise except Exception: LOGGER.error("Evaluation request failed") raise HTTPException(503, "Evaluation is temporarily unavailable. Please retry later.") from None finally: SUBMISSION_SLOT.release() @server.post("/api/evaluation/submit") async def submit_evaluation_report(request: Request): from src.evaluation.scoring import SubmissionError from src.evaluation.service import EvaluationUnavailable if not SUBMISSION_SLOT.acquire(blocking=False): raise HTTPException(429, "Another evaluation request is being processed. Please retry shortly.", headers={"Retry-After": "10"}) try: content, fields = await _read_evaluation_form(request, include_receipt=True) result = await run_in_threadpool(evaluation_service().submit_report, content, fields) return JSONResponse(result, headers={"Cache-Control": "no-store"}) except SubmissionError as exc: raise HTTPException(422, str(exc)) from None except EvaluationUnavailable as exc: raise HTTPException(503, str(exc)) from None except Exception: LOGGER.error("Evaluation report submission failed") raise HTTPException(503, "Could not submit this report for review. Please retry later.") from None finally: SUBMISSION_SLOT.release() @server.api(name="zerogpu_check", queue=True, concurrency_limit=1) @spaces.GPU(duration=10) def zerogpu_check() -> dict[str, Any]: """Opt-in GPU diagnostic; never called by startup or leaderboard browsing. Registration must happen at module import for ZeroGPU startup discovery. Calling this endpoint on ZeroGPU requests GPU time and may consume quota. """ import torch if not torch.cuda.is_available(): return {"status": "unavailable", "message": "No CUDA GPU on this local host; leaderboard remains CPU-only."} checksum = torch.tensor([1.0, 2.0, 3.0], device="cuda").sum().item() return {"status": "ok", "device": torch.cuda.get_device_name(0), "checksum": checksum} @server.get("/", response_class=HTMLResponse, include_in_schema=False) async def homepage() -> HTMLResponse: html = (STATIC_DIR / "index.html").read_text(encoding="utf-8") for filename in ("styles.css", "app.js"): digest = hashlib.sha256((STATIC_DIR / filename).read_bytes()).hexdigest()[:20] html = html.replace(f"/static/{filename}", f"/ark-assets/{digest}/{filename}") return HTMLResponse(html, headers={"Cache-Control": "no-cache, max-age=0, must-revalidate"}) @server.get("/ark-assets/{version}/{filename}", include_in_schema=False) async def frontend_asset(version: str, filename: str) -> Response: # Separate namespace from Gradio; content-addressed URLs cannot reuse old CSS/JS. media_types = {"styles.css": "text/css", "app.js": "application/javascript"} if filename not in media_types: raise HTTPException(status_code=404) content = (STATIC_DIR / filename).read_bytes() digest = hashlib.sha256(content).hexdigest()[:20] if version != digest: raise HTTPException(status_code=404) return Response(content, media_type=media_types[filename], headers={ "Cache-Control": "public, max-age=31536000, immutable", "X-Content-Type-Options": "nosniff", }) @server.api(name="leaderboard_data", queue=False, concurrency_limit=None) def leaderboard_data(family: str = "Recall", metric: str = "R@1") -> dict[str, Any]: """Return all four leaderboard views for one selected metric.""" if (family, metric) not in VALID_METRICS: return _empty_payload(f"Unsupported metric: {family}/{metric}") try: result = _build_payload(family, metric) with DATA_LOCK: payload = dict(DATA) if DATA else _empty_payload() payload.update(result) payload["selected_metric"] = {"family": family, "metric": metric} payload["updated_at"] = datetime.now(timezone.utc).isoformat() return payload except Exception as exc: LOGGER.exception("Could not load leaderboard data") return _empty_payload(f"Could not load leaderboard data: {exc}") @server.api(name="download_ranking", queue=False, concurrency_limit=None) def download_ranking(view: str = "overview", family: str = "Recall", metric: str = "R@1") -> dict[str, str]: """Return the selected ranking as CSV and JSON strings.""" payload = leaderboard_data(family, metric) rows = payload.get("views", {}).get(view, []) frame = frame_for_download(rows) csv_buffer = io.StringIO() frame.to_csv(csv_buffer, index=False) download_rows = [{key: value for key, value in row.items() if key != "cognitive_avg"} for row in rows] return {"csv": csv_buffer.getvalue(), "json": json.dumps(download_rows, indent=2)} if __name__ == "__main__": refresh_data() server.launch(blocked_paths=[str(ROOT / name) for name in ("evaluation", "evaluation-cache", "submissions")], max_file_size="2mb")