Blame for app.py

"""CADGenBench Leaderboard Space - Gradio UI + report-proxy mount.

Read path lives in :mod:`leaderboard`. Submit-tab validation lives in
:mod:`submit`. Both are wired into the Gradio Blocks below. The
Gradio app is mounted under a FastAPI parent so the custom
``/reports/{submission_id}.html`` route can re-serve dataset HTML
with ``Content-Type: text/html`` (HF Hub's ``/resolve/`` serves it
as ``text/plain`` by policy, which makes the browser show source
rather than render).
"""
from __future__ import annotations

import html
import logging
import os
from functools import lru_cache
from pathlib import Path

import gradio as gr
import pandas as pd
import uvicorn
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, Response
from gradio_leaderboard import Leaderboard
from huggingface_hub import hf_hub_download

from leaderboard import (
    ADMIN_COLUMNS,
    ADMIN_SELECT_COL,
    HF_DATA_REPO,
    HF_SUBMISSIONS_REPO,
    LEADERBOARD_DATATYPES,
    LEADERBOARD_HIDE_COLUMNS,
    VALIDATED_LEADERBOARD_DATATYPES,
    _fmt_timestamp,
    build_combined_csv,
    load_admin_table,
    load_leaderboard_split,
)
from admin import (
    VALID_METHODS,
    delete_rows,
    demote_rows,
    is_admin,
    promote_rows,
)
from submit import handle_submit

logger = logging.getLogger(__name__)

# Surface module-level logger.info / logger.warning / logger.exception
# calls from leaderboard.py + submit.py in the Space's runtime logs.
# Otherwise they go nowhere and any refresh / worker pathology is
# silent. Format keeps timestamps + module + level + message.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)


# Canonical policy doc lives in the code repo so contributors reading
# the GitHub repo see it without needing to visit the Space. Linked
# from both the Leaderboard tab's Validation Guidelines accordion and
# the About tab.
VALIDATION_DOC_URL = (
    "https://github.com/huggingface/cadgenbench/blob/main/docs/benchmark/validation.md"
)

ABOUT_MD = f"""## About

**CADGenBench** evaluates AI-driven CAD generation: how well a model can
turn a description of a mechanical part into a valid, geometrically
correct 3D model.

- **Reference baseline**: an iterative AI agent that writes build123d Python.
- **Submission flow**: upload a zip of per-fixture STEP files; the Space
  runs the eval and appends a row to the submissions dataset.
- **Datasets**: fixture inputs in
  [`{HF_DATA_REPO}`](https://huggingface.co/datasets/{HF_DATA_REPO});
  submissions and computed results in
  [`{HF_SUBMISSIONS_REPO}`](https://huggingface.co/datasets/{HF_SUBMISSIONS_REPO}).
- **Code**: [`huggingface/cadgenbench`](https://github.com/huggingface/cadgenbench).
- **Validation policy**: [`docs/benchmark/validation.md`]({VALIDATION_DOC_URL}).
- **Data**: CAD geometry from [Mecado](https://www.mecado.com).
"""

# Verbatim BibTeX entry locked in space-setup/bundles/1-2-space-ux.md
# (Locked decisions section). Shown in the Citation accordion as a
# copy-paste handle for anyone citing this benchmark; the About tab
# already links the source code via huggingface/cadgenbench so the
# Space URL is the right deep-link target for the citation.
CITATION_BIBTEX = r"""@misc{cadgenbench2026,
  author       = {Rabinovich, Michael and {Hugging Face}},
  title        = {{CADGenBench}: a benchmark for {AI}-driven {CAD} generation},
  year         = {2026},
  publisher    = {Hugging Face},
  howpublished = {\url{https://huggingface.co/spaces/HuggingAI4Engineering/cadgenbench-leaderboard}},
}"""

VALIDATION_GUIDELINES_MD = f"""Submissions appear on the **Unvalidated** table the moment evaluation completes. Maintainers promote rows to **Validated** after methodology review, accepting one of four evidence types (`code`, `traces`, `api`, `manual`).

Full policy: [`docs/benchmark/validation.md`]({VALIDATION_DOC_URL})."""

DETAIL_PLACEHOLDER = "_Click a row above for details._"


def _has(value) -> bool:
    """True for values that should show up in the detail panel."""
    if value is None:
        return False
    if isinstance(value, float) and pd.isna(value):
        return False
    return str(value).strip() != ""


def _build_report_iframe(html_bytes: bytes) -> str:
    """Wrap a fetched report's HTML bytes into a self-contained iframe.

    ``srcdoc`` puts the entire report HTML directly inside the iframe
    attribute. The iframe gets its own document context, so the
    report's CSS can't collide with Gradio's, and an explicit
    ``height: 90vh`` keeps the report from being clipped by Gradio's
    column-flex layout (which was the visible cut-off the user saw
    in earlier rendering attempts).

    The Space is private; the FastAPI ``/reports/<id>.html`` route
    works server-side under Bearer auth but breaks for a logged-in
    browser user (HF's edge gates same-origin pathname navigations
    on a JWT that the browser doesn't carry forward). srcdoc
    sidesteps that entirely by inlining the bytes; no second HTTP
    request leaves the browser.
    """
    escaped = html.escape(
        html_bytes.decode("utf-8", errors="replace"), quote=True,
    )
    return (
        f'<iframe srcdoc="{escaped}" '
        'style="width:100%; height:90vh; border:0; display:block;" '
        'title="Submission report"></iframe>'
    )


def _refresh_leaderboard_with_toast():
    """Manual Refresh button handler: toast + fresh DataFrames.

    The Timer auto-refresh wires straight to ``load_leaderboard_split``
    so it stays silent (a toast every 10s would be noise). Only the
    explicit click goes through this wrapper.
    """
    gr.Info("Leaderboard refreshed.")
    return load_leaderboard_split()


def _enable_submit_when_logged_in(
    profile: gr.OAuthProfile | None,
) -> gr.Button:
    """Flip the Submit button's interactivity based on login state.

    Runs once per page load via ``blocks.load``. Gradio injects
    ``gr.OAuthProfile`` automatically (``None`` if the visitor isn't
    logged in via the LoginButton). The visible-disable mirrors the
    server-side gate in :func:`submit.handle_submit`; the handler
    still raises ``gr.Error`` defensively if it ever gets called
    without a profile.
    """
    return gr.Button(interactive=profile is not None)


def _selected_ids(table_df: pd.DataFrame | None) -> list[str]:
    """Submission ids of the rows whose ``select`` checkbox is ticked."""
    if (
        table_df is None
        or len(table_df) == 0
        or ADMIN_SELECT_COL not in table_df.columns
        or "submission_id" not in table_df.columns
    ):
        return []
    mask = table_df[ADMIN_SELECT_COL].apply(bool)
    return [str(s) for s in table_df.loc[mask, "submission_id"].tolist() if s]


def _admin_selection_status(table_df: pd.DataFrame | None) -> str:
    """Live count line under the admin table, updated as boxes are ticked."""
    n = len(_selected_ids(table_df))
    return f"**{n}** row(s) selected." if n else "_No rows selected._"


def _gate_admin_controls(
    profile: gr.OAuthProfile | None,
) -> tuple[gr.Dataframe, gr.Radio, gr.Button, gr.Button, gr.Checkbox, gr.Button, str]:
    """Enable the admin controls only for a logged-in user in the admin set.

    Runs on every page load and re-runs on LoginButton auth events.
    Non-admins and logged-out visitors get the tab with the table
    read-only and every control disabled, mirroring the server-side
    re-check in each handler. The delete button always loads disarmed:
    it only enables once the confirm checkbox is ticked.
    """
    admin = is_admin(profile)
    if profile is None:
        status = "Log in with an admin account to enable the controls below."
    elif admin:
        status = f"Signed in as `{profile.username}`. Admin controls enabled."
    else:
        status = (
            f"Signed in as `{profile.username}`, which is not in the admin "
            "set. Controls are disabled."
        )
    return (
        gr.Dataframe(interactive=admin),
        gr.Radio(interactive=admin),
        gr.Button(interactive=admin),
        gr.Button(interactive=admin),
        gr.Checkbox(interactive=admin, value=False),
        gr.Button(interactive=False),
        status,
    )


def _arm_delete(
    confirm: bool, profile: gr.OAuthProfile | None,
) -> gr.Button:
    """Enable the delete button only when an admin has ticked the confirm box."""
    return gr.Button(interactive=bool(confirm) and is_admin(profile))


def _admin_promote(
    table_df: pd.DataFrame | None,
    method: str | None,
    profile: gr.OAuthProfile | None,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """Promote every ticked row, then refresh the admin table + both tiers.

    Re-checks :func:`admin.is_admin` server-side so a tampered client
    that re-enables the button still can't write.
    """
    if not is_admin(profile):
        raise gr.Error("You are not in the admin set.")
    ids = _selected_ids(table_df)
    if not ids:
        raise gr.Error("Tick at least one row first.")
    if not method:
        raise gr.Error("Pick a validation_method first.")
    try:
        promote_rows(ids, method)
    except (LookupError, ValueError) as e:
        raise gr.Error(str(e))
    gr.Info(f"Promoted {len(ids)} row(s) to validated ({method}).")
    validated, unvalidated = load_leaderboard_split()
    return load_admin_table(), validated, unvalidated


def _admin_demote(
    table_df: pd.DataFrame | None,
    profile: gr.OAuthProfile | None,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    """Demote every ticked row, then refresh the admin table + both tiers."""
    if not is_admin(profile):
        raise gr.Error("You are not in the admin set.")
    ids = _selected_ids(table_df)
    if not ids:
        raise gr.Error("Tick at least one row first.")
    try:
        demote_rows(ids)
    except (LookupError, ValueError) as e:
        raise gr.Error(str(e))
    gr.Info(f"Demoted {len(ids)} row(s) to unvalidated.")
    validated, unvalidated = load_leaderboard_split()
    return load_admin_table(), validated, unvalidated


def _admin_delete(
    table_df: pd.DataFrame | None,
    confirm: bool,
    profile: gr.OAuthProfile | None,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, gr.Checkbox, gr.Button]:
    """Delete every ticked row (artifacts + row), then refresh and disarm.

    Resets the confirm checkbox and re-disables the delete button on
    the way out so the next deletion needs a fresh, deliberate confirm.
    """
    if not is_admin(profile):
        raise gr.Error("You are not in the admin set.")
    if not confirm:
        raise gr.Error("Tick the confirmation box to enable delete.")
    ids = _selected_ids(table_df)
    if not ids:
        raise gr.Error("Tick at least one row first.")
    try:
        delete_rows(ids)
    except ValueError as e:
        raise gr.Error(str(e))
    gr.Info(f"Deleted {len(ids)} submission(s).")
    validated, unvalidated = load_leaderboard_split()
    return (
        load_admin_table(),
        validated,
        unvalidated,
        gr.Checkbox(value=False),
        gr.Button(interactive=False),
    )


def _format_detail_and_report(
    df: pd.DataFrame | None, evt: gr.SelectData,
) -> tuple[str, str]:
    """Return ``(detail_markdown, report_iframe_html)`` for the clicked row.

    The detail panel holds metadata (submitter, status, timestamp,
    notes, links to ZIP and external agent URL). The HTML viewer
    holds the rendered report when one exists (status == completed
    AND the row carries the modern-pipeline ``submission_sha256``
    sentinel) - the leaderboard reader computes ``report_url`` only
    for rows that satisfy that gate.

    Returns ``(DETAIL_PLACEHOLDER, "")`` on a null / out-of-range
    event so the panel falls back to its initial state.
    """
    if df is None or len(df) == 0 or evt is None or evt.index is None:
        return DETAIL_PLACEHOLDER, ""
    idx = evt.index[0] if isinstance(evt.index, (list, tuple)) else evt.index
    if idx < 0 or idx >= len(df):
        return DETAIL_PLACEHOLDER, ""
    row = df.iloc[idx]

    title = row.get("submission_name") or "(unnamed submission)"
    lines = [f"### {title}", ""]
    if _has(row.get("submitter_name")):
        lines.append(f"- **Submitter**: {row['submitter_name']}")
    if _has(row.get("status")):
        lines.append(f"- **Status**: {row['status']}")
    if _has(row.get("submitted_at")):
        lines.append(f"- **Submitted**: {_fmt_timestamp(row['submitted_at'])}")
    if _has(row.get("notes")):
        lines.append(f"- **Notes**: {row['notes']}")
    lines.append(
        f"- **Model details (optional)**: "
        f"{row.get('model details (optional)') or '_None_'}"
    )
    if _has(row.get("submission_blob_url")):
        lines.append(
            f"- **Submission ZIP**: [download]({row['submission_blob_url']})"
        )
    if row.get("status") == "failed" and _has(row.get("failure_reason")):
        lines.append(f"- **Failure reason**: {row['failure_reason']}")
    detail_md = "\n".join(lines)

    report_iframe = ""
    if _has(row.get("report_url")) and _has(row.get("submission_id")):
        content = _fetch_report_html(str(row["submission_id"]))
        if content:
            report_iframe = _build_report_iframe(content)
    return detail_md, report_iframe


@lru_cache(maxsize=128)
def _fetch_report_html(submission_id: str) -> bytes | None:
    """Pull ``reports/<id>.html`` off the submissions dataset.

    Cached in-process so repeat clicks on the same row don't hit
    the Hub. Returns ``None`` on any failure so the caller can
    serve a clean 404 rather than leaking a stack trace.
    """
    try:
        local_path = hf_hub_download(
            repo_id=HF_SUBMISSIONS_REPO,
            filename=f"reports/{submission_id}.html",
            repo_type="dataset",
        )
        return Path(local_path).read_bytes()
    except Exception as e:  # noqa: BLE001 - any Hub failure -> 404
        logger.warning(
            "Failed to fetch report for %s (%s: %s)",
            submission_id, type(e).__name__, e,
        )
        return None


def serve_report(submission_id: str) -> Response:
    """Proxy a per-submission HTML report through the Space.

    HF Hub serves dataset HTML under ``/resolve/`` with
    ``Content-Type: text/plain`` (security: dataset files can't host
    live HTML), so a direct dataset link shows source instead of
    rendering. This route lives on the Space (which can legitimately
    serve text/html) and re-streams the file's bytes with the right
    content-type.
    """
    content = _fetch_report_html(submission_id)
    if content is None:
        return HTMLResponse(
            content="<h1>Report not found</h1>",
            status_code=404,
        )
    return Response(content=content, media_type="text/html; charset=utf-8")


with gr.Blocks(title="CADGenBench Leaderboard", theme=gr.themes.Soft()) as blocks:
    gr.Markdown(
        "# CADGenBench Leaderboard\n"
        "_Benchmarking AI-driven CAD generation._"
    )

    with gr.Tab("Leaderboard"):
        # Collapsed accordions above the tables. Validation guidelines
        # gives the short two-tier story + link to the full policy
        # doc; Citation carries the verbatim BibTeX entry. Both start
        # closed so the leaderboard itself stays above the fold.
        with gr.Accordion("Validation guidelines", open=False):
            gr.Markdown(VALIDATION_GUIDELINES_MD)
        with gr.Accordion("Citation", open=False):
            # language=None -> plain monospaced render (gr.Code doesn't
            # ship a BibTeX highlighter); show_line_numbers off because
            # the entry is meant to be copy-pasted, not annotated.
            gr.Code(
                value=CITATION_BIBTEX,
                language=None,
                show_line_numbers=False,
            )

        # Two stacked tables, split by `validation_status`. Validated
        # on top so the curated results are above the fold; unvalidated
        # below carries every other row (auto-published, awaiting
        # methodology review). See decisions/validation-policy.md.
        initial_validated, initial_unvalidated = load_leaderboard_split()
        validated_view = Leaderboard(
            value=initial_validated,
            datatype=VALIDATED_LEADERBOARD_DATATYPES,
            search_columns=["submission_name", "submitter_name"],
            hide_columns=LEADERBOARD_HIDE_COLUMNS,
            label="Validated Leaderboard",
            interactive=False,
        )
        unvalidated_view = Leaderboard(
            value=initial_unvalidated,
            datatype=LEADERBOARD_DATATYPES,
            search_columns=["submission_name", "submitter_name"],
            hide_columns=LEADERBOARD_HIDE_COLUMNS,
            label="Unvalidated Leaderboard",
            interactive=False,
        )
        with gr.Row():
            refresh_btn = gr.Button("Refresh", size="sm")
            # One file, both tables, `validation_status` discriminator
            # column. Fresh CSV is generated on every click so the
            # download reflects the latest data, not a stale snapshot
            # captured at boot.
            download_btn = gr.DownloadButton(
                label="Download CSV", size="sm",
            )
        refresh_btn.click(
            fn=_refresh_leaderboard_with_toast,
            outputs=[validated_view, unvalidated_view],
        )
        download_btn.click(fn=build_combined_csv, outputs=download_btn)

        # Row-click panel: one shared metadata markdown component +
        # one report viewer below it. The viewer holds an iframe
        # containing the full per-submission report (srcdoc-inlined
        # so no second HTTP request needs to leave the browser - the
        # Space is private and HF's edge would 404 same-origin
        # pathname navigations that aren't carrying the iframe's
        # short-lived `__sign` JWT). Both tables share both outputs.
        detail_panel = gr.Markdown(
            value=DETAIL_PLACEHOLDER,
            label="Selected submission",
        )
        report_viewer = gr.HTML(value="", label="Report")
        for view in (validated_view, unvalidated_view):
            view.select(
                fn=_format_detail_and_report,
                inputs=view,
                outputs=[detail_panel, report_viewer],
            )

    with gr.Tab("Submit"):
        gr.Markdown(
            f"""
**Submission format.** A single zip with:

- one folder per fixture in `{HF_DATA_REPO}`, each containing `output.step`;
- a top-level `meta.json`:

```json
{{
  "submitter_name": "your name or team",
  "submission_name": "MyAgent v2.3 (or whatever describes your system)",
  "agent_url": "https://github.com/...   (optional)",
  "notes": "free text, optional, max 500 chars, single line, plain text",
  "agree_to_publish": true
}}
```

**Submission name.** Free text describing the system being benchmarked,
however you choose to describe it. The benchmark is system-agnostic: your
submission may use no LLM, one, or many. If you want to disclose your
stack, put it here or in `notes`.

**Notes field.** Plain text only (no markdown / HTML). Capped at 500 chars
and stripped to a single line. Shown in the per-submission detail view,
not in the main leaderboard table.

**Consent.** `"agree_to_publish": true` in `meta.json` is your consent
to publish the resulting row on the public leaderboard.
"""
        )
        # OAuth gate. The user must log in via the HF button before
        # the Submit button becomes interactive; the row gets the
        # canonical `hf_username` from `gr.OAuthProfile.username`
        # (not a free-text claim in meta.json). README front-matter
        # already carries `hf_oauth: true` so HF's OAuth integration
        # is wired up at the Space level.
        login_btn = gr.LoginButton()
        zip_in = gr.File(label="Submission ZIP", file_types=[".zip"])
        # Starts disabled; the `blocks.load` handler below flips it
        # to interactive when an OAuthProfile is present.
        submit_btn = gr.Button("Submit", variant="primary", interactive=False)
        # No static markdown output: handle_submit surfaces every
        # status update via gr.Info / gr.Error toasts. The handler
        # also reads `gr.OAuthProfile` implicitly via its parameter
        # type annotation (Gradio's dependency-injection convention).
        submit_btn.click(fn=handle_submit, inputs=[zip_in])

    with gr.Tab("About"):
        gr.Markdown(ABOUT_MD)

    with gr.Tab("Admin"):
        # Maintainer-only controls. The tab is visible to everyone (a
        # hint the path exists); the table + buttons are gated to OAuth
        # users in the CADGENBENCH_ADMINS set via the `blocks.load`
        # handler below + a server-side re-check in every handler. See
        # decisions/validation-policy.md.
        gr.Markdown(
            "## Admin\n"
            "Tick rows in the **select** column, then promote them into the "
            "**Validated** tier (recording an evidence type), demote them back "
            "to **Unvalidated**, or delete them. Actions apply to every ticked "
            "row at once. Limited to maintainers in the admin set; everyone "
            "else sees the tab with the controls disabled."
        )
        admin_login_btn = gr.LoginButton()
        admin_status = gr.Markdown(
            "Log in with an admin account to enable the controls below."
        )
        # Only the leading `select` column is editable; the rest is
        # read-only context. Click-to-tick drives every action below.
        admin_table = gr.Dataframe(
            value=load_admin_table(),
            datatype=[
                "bool", "str", "str", "str", "str", "str", "str", "number",
                "str",
            ],
            static_columns=list(range(1, len(ADMIN_COLUMNS))),
            interactive=False,
            label="Submissions (tick select to choose rows)",
            wrap=True,
        )
        admin_selection_md = gr.Markdown("_No rows selected._")
        admin_method_radio = gr.Radio(
            choices=list(VALID_METHODS),
            value="manual",
            label="validation_method (applied to all rows on promote)",
            interactive=False,
        )
        with gr.Row():
            promote_btn = gr.Button(
                "Mark validated", variant="primary", interactive=False,
            )
            demote_btn = gr.Button("Mark unvalidated", interactive=False)
        with gr.Accordion("Danger zone: delete", open=False):
            gr.Markdown(
                "Permanently deletes the ticked rows **and** their uploaded "
                "zip + report files from the submissions dataset. This cannot "
                "be undone (only a manual revert of the dataset commit)."
            )
            delete_confirm = gr.Checkbox(
                label=(
                    "I understand this permanently deletes the selected "
                    "submissions and their files."
                ),
                value=False,
                interactive=False,
            )
            delete_btn = gr.Button(
                "Delete selected", variant="stop", interactive=False,
            )
        admin_refresh_btn = gr.Button("Refresh", size="sm")

        admin_table.change(
            fn=_admin_selection_status,
            inputs=admin_table,
            outputs=admin_selection_md,
        )
        promote_btn.click(
            fn=_admin_promote,
            inputs=[admin_table, admin_method_radio],
            outputs=[admin_table, validated_view, unvalidated_view],
        )
        demote_btn.click(
            fn=_admin_demote,
            inputs=[admin_table],
            outputs=[admin_table, validated_view, unvalidated_view],
        )
        delete_confirm.change(
            fn=_arm_delete, inputs=[delete_confirm], outputs=delete_btn,
        )
        delete_btn.click(
            fn=_admin_delete,
            inputs=[admin_table, delete_confirm],
            outputs=[
                admin_table, validated_view, unvalidated_view,
                delete_confirm, delete_btn,
            ],
        )
        admin_refresh_btn.click(fn=load_admin_table, outputs=admin_table)

    # gradio_leaderboard.Leaderboard handles its own update path
    # cleanly; bind a Timer to push fresh dataframes every 10 seconds.
    # Single tick runs `load_leaderboard_split` once and pushes the
    # tuple's two halves into the validated / unvalidated widgets.
    auto_refresh_timer = gr.Timer(10)
    auto_refresh_timer.tick(
        fn=load_leaderboard_split,
        outputs=[validated_view, unvalidated_view],
    )

    # On page load, read the visitor's OAuth profile (None if not
    # logged in) and flip the Submit button's interactivity. Runs once
    # per page load; LoginButton clicks also re-trigger this through
    # Gradio's auth-event plumbing.
    blocks.load(fn=_enable_submit_when_logged_in, outputs=submit_btn)

    # Same per-load OAuth read, gating the Admin tab's controls on
    # membership in the CADGENBENCH_ADMINS set. Logged-out / non-admin
    # visitors get the tab with everything disabled.
    blocks.load(
        fn=_gate_admin_controls,
        outputs=[
            admin_table,
            admin_method_radio,
            promote_btn,
            demote_btn,
            delete_confirm,
            delete_btn,
            admin_status,
        ],
    )


# Mount Gradio under a FastAPI parent so the custom proxy route
# above lives at the same origin as the UI. Direct routes on `app`
# get checked before the Gradio sub-app, so `/reports/<sid>.html`
# never gets shadowed.
app = FastAPI()
app.add_api_route(
    "/reports/{submission_id}.html",
    serve_report,
    methods=["GET"],
)
app = gr.mount_gradio_app(app, blocks, path="/")


if __name__ == "__main__":
    host = os.getenv("GRADIO_SERVER_NAME", "0.0.0.0")
    port = int(os.getenv("GRADIO_SERVER_PORT", "7860"))
    uvicorn.run(app, host=host, port=port)