app.py

"""Public direct-submission FinReason Task 1 development portal."""

from __future__ import annotations

import base64
import hashlib
import io
import json
import os
import stat
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

import gradio as gr
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download, list_repo_files

from finreason_task1.admission import MAX_ARCHIVE_BYTES, read_submission_archive
from finreason_task1.contracts import ContractError, load_json_object_snapshot, load_jsonl_snapshot
from finreason_task1.hf_portal import (
    build_submission_event,
    canonical_json_bytes,
    development_leaderboard,
    encrypt_submission_archive,
    event_context_is_accepted,
    event_is_leaderboard_eligible,
    event_receipt_id,
    event_team_id,
    immutable_public_dataset_url,
    make_idempotency_key,
    make_logical_replay_key,
    make_submission_id,
    normalize_team_name,
    quota_state,
    submission_event_path,
    team_id_from_name,
    team_name_identity,
    validate_portal_contract,
    validate_submission_event,
)
from finreason_task1.leaderboard import admit_production_leaderboard_submission
from finreason_task1.scoring import SCORER_VERSION, score_submission


ROOT = Path(__file__).resolve().parent
if os.environ.get("SYSTEM") != "spaces":
    raise ContractError("E_PLATFORM_CONFIGURATION")


def _required_secret(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise ContractError("E_PLATFORM_CONFIGURATION")
    return value


def _load_contract_secret() -> tuple[bytes, Mapping[str, Any]]:
    try:
        contract_bytes = base64.b64decode(
            _required_secret("FINREASON_PORTAL_CONTRACT_B64"), validate=True
        )
        contract = json.loads(contract_bytes.decode("utf-8", errors="strict"))
    except Exception as exc:
        raise ContractError("E_PLATFORM_CONFIGURATION") from exc
    if (
        not isinstance(contract, Mapping)
        or contract_bytes != canonical_json_bytes(contract) + b"\n"
    ):
        raise ContractError("E_PLATFORM_CONFIGURATION")
    validate_portal_contract(contract)
    return contract_bytes, contract


CONTRACT_BYTES, CONTRACT = _load_contract_secret()
NON_PRODUCTION_MARKER = ROOT / "NON_PRODUCTION.txt"
NON_PRODUCTION_MARKER_BYTES = (
    b"This deterministic package is a non-production rehearsal and must not be distributed.\n"
)
if CONTRACT["mode"] == "production" and CONTRACT["publication_stage"] == "launch":
    if NON_PRODUCTION_MARKER.exists():
        raise ContractError("E_PLATFORM_CONFIGURATION")
elif (
    CONTRACT["mode"] == "rehearsal"
    and CONTRACT["publication_stage"] == "hosted_rehearsal"
):
    try:
        valid_non_production_marker = (
            NON_PRODUCTION_MARKER.is_file()
            and NON_PRODUCTION_MARKER.read_bytes() == NON_PRODUCTION_MARKER_BYTES
        )
    except OSError:
        valid_non_production_marker = False
    if not valid_non_production_marker:
        raise ContractError("E_PLATFORM_CONFIGURATION")
else:
    raise ContractError("E_PLATFORM_CONFIGURATION")
CONTRACT_SHA256 = hashlib.sha256(CONTRACT_BYTES).hexdigest()


def _intake_schedule(contract: Mapping[str, Any], close_override: str | None) -> dict[str, str]:
    schedule = dict(contract["schedule"])
    if close_override is not None:
        try:
            closed = datetime.strptime(close_override, "%Y-%m-%dT%H:%M:%SZ")
            opened = datetime.strptime(schedule["development_open"], "%Y-%m-%dT%H:%M:%SZ")
        except (TypeError, ValueError) as exc:
            raise ContractError("E_PLATFORM_CONFIGURATION") from exc
        if closed <= opened or closed.strftime("%Y-%m-%dT%H:%M:%SZ") != close_override:
            raise ContractError("E_PLATFORM_CONFIGURATION")
        schedule["development_close"] = close_override
    return schedule


# The organizer deadline may change without changing stored event identities.
INTAKE_SCHEDULE = _intake_schedule(CONTRACT, os.environ.get("FINREASON_SUBMISSION_CLOSE_UTC"))
ENCRYPTION_PUBLIC_KEY_SHA256 = hashlib.sha256(
    base64.b64decode(CONTRACT["vault_recipient_public_key_b64"], validate=True)
).hexdigest()
API = HfApi()
MATERIALIZED_DOWNLOAD_ROOT = Path(
    tempfile.mkdtemp(prefix="finreason-hf-materialized-")
)
os.chmod(MATERIALIZED_DOWNLOAD_ROOT, 0o700)


def _reference_revision() -> str:
    revision = _commit_sha(
        _required_secret("FINREASON_REFERENCE_REVISION"),
        "E_REFERENCE_REVISION",
    )
    if revision != CONTRACT["reference_revision"]:
        raise ContractError("E_REFERENCE_REVISION")
    return revision


def _tokens() -> tuple[str, str]:
    reference = _required_secret("FINREASON_REFERENCE_TOKEN")
    vault = _required_secret("FINREASON_VAULT_TOKEN")
    if reference == vault:
        raise ContractError("E_PLATFORM_TOKEN_SEPARATION")
    return reference, vault


def _download(
    repo_id: str,
    filename: str,
    token: str,
    *,
    revision: str | None = None,
    error_code: str,
) -> Path:
    repository_root = MATERIALIZED_DOWNLOAD_ROOT / hashlib.sha256(
        repo_id.encode("utf-8", errors="strict")
    ).hexdigest()
    try:
        repository_root.mkdir(mode=0o700, exist_ok=True)
        os.chmod(repository_root, 0o700)
        downloaded = Path(
            hf_hub_download(
                repo_id=repo_id,
                filename=filename,
                repo_type="dataset",
                token=token,
                revision=revision,
                local_dir=repository_root,
            )
        )
        if not downloaded.is_absolute():
            raise OSError("Hub download did not return an absolute path")
        downloaded.relative_to(repository_root)
        current = downloaded.parent
        while True:
            metadata = os.lstat(current)
            if not stat.S_ISDIR(metadata.st_mode):
                raise OSError("Hub download parent is not a regular directory")
            if current == repository_root:
                break
            current = current.parent
        metadata = os.lstat(downloaded)
        if not stat.S_ISREG(metadata.st_mode):
            raise OSError("Hub download is not a regular file")
        os.chmod(downloaded, 0o600)
        return downloaded
    except Exception as exc:
        raise ContractError(error_code) from exc


def _commit_sha(value: Any, code: str) -> str:
    if not isinstance(value, str) or len(value) != 40 or any(
        character not in "0123456789abcdef" for character in value
    ):
        raise ContractError(code)
    return value


def _repo_head(repo_id: str, token: str, code: str) -> str:
    return _commit_sha(
        API.repo_info(repo_id, repo_type="dataset", token=token).sha,
        code,
    )


def _verified_reference(
    filename: str,
    expected_sha256: str,
    token: str,
    *,
    revision: str,
) -> Path:
    path = _download(
        CONTRACT["reference_repo_id"],
        filename,
        token,
        revision=revision,
        error_code="E_REFERENCE_UNAVAILABLE",
    )
    if hashlib.sha256(path.read_bytes()).hexdigest() != expected_sha256:
        raise ContractError("E_REFERENCE_COMMITMENT")
    return path


def _index_jsonl(path: Path, label: str) -> dict[str, Mapping[str, Any]]:
    _content, loaded = load_jsonl_snapshot(path)
    if not loaded.valid:
        raise ContractError(f"E_REFERENCE_{label.upper()}")
    result: dict[str, Mapping[str, Any]] = {}
    for row in loaded.records:
        case_id = row.get("case_id")
        if not isinstance(case_id, str) or case_id in result:
            raise ContractError(f"E_REFERENCE_{label.upper()}")
        result[case_id] = row
    return result


def _load_events(
    vault_token: str,
) -> tuple[str, list[Mapping[str, Any]], frozenset[str]]:
    events: list[Mapping[str, Any]] = []
    head = _repo_head(CONTRACT["vault_repo_id"], vault_token, "E_VAULT_REVISION")
    all_paths = frozenset(
        path
        for path in list_repo_files(
            CONTRACT["vault_repo_id"],
            repo_type="dataset",
            token=vault_token,
            revision=head,
        )
    )
    event_paths = sorted(
        path
        for path in all_paths
        if path.startswith("events/") and path.endswith(".json")
    )
    seen_submission_ids: set[str] = set()
    seen_idempotency_keys: set[str] = set()
    seen_logical_replay_keys: set[str] = set()
    seen_archive_paths: set[str] = set()
    expected_context = {
        "dataset_version": CONTRACT["dataset_version"],
        "release_version": CONTRACT["release_version"],
        "scorer_version": SCORER_VERSION,
        "release_manifest_sha256": CONTRACT["release_manifest_sha256"],
        "questions_jsonl_sha256": CONTRACT["leaderboard_questions_sha256"],
        "expected_ids_json_sha256": CONTRACT["leaderboard_expected_ids_sha256"],
        "gold_jsonl_sha256": CONTRACT["leaderboard_gold_sha256"],
        "manifest_jsonl_sha256": CONTRACT["leaderboard_manifest_sha256"],
    }
    for filename in event_paths:
        event_path = _download(
            CONTRACT["vault_repo_id"],
            filename,
            vault_token,
            revision=head,
            error_code="E_STORAGE_UNAVAILABLE",
        )
        _content, event = load_json_object_snapshot(event_path)
        validate_submission_event(event)
        if filename != submission_event_path(event["submission_id"]):
            raise ContractError("E_VAULT_EVENT_PATH")
        if any(event.get(field) != expected for field, expected in expected_context.items()):
            raise ContractError("E_VAULT_EVENT_CONTEXT")
        if not event_context_is_accepted(
            event,
            contract=CONTRACT,
            portal_contract_sha256=CONTRACT_SHA256,
        ):
            raise ContractError("E_VAULT_EVENT_CONTEXT")
        historical_context = next(
            (context for context in CONTRACT["accepted_event_contexts"]
             if context["portal_contract_sha256"] == event["portal_contract_sha256"]),
            {},
        )
        expected_key_sha256 = historical_context.get(
            "encryption_public_key_sha256", ENCRYPTION_PUBLIC_KEY_SHA256
        )
        if event["encryption_public_key_sha256"] != expected_key_sha256:
            raise ContractError("E_VAULT_EVENT_CONTEXT")
        if event["encrypted_archive_path"] not in all_paths:
            raise ContractError("E_VAULT_ARCHIVE_MISSING")
        logical_replay_key = make_logical_replay_key(
            event_team_id(event),
            event["archive_sha256"],
            event["release_manifest_sha256"],
        )
        if (
            event["submission_id"] in seen_submission_ids
            or event["idempotency_key"] in seen_idempotency_keys
            or logical_replay_key in seen_logical_replay_keys
            or event["encrypted_archive_path"] in seen_archive_paths
        ):
            raise ContractError("E_VAULT_EVENT_DUPLICATE")
        seen_submission_ids.add(event["submission_id"])
        seen_idempotency_keys.add(event["idempotency_key"])
        seen_logical_replay_keys.add(logical_replay_key)
        seen_archive_paths.add(event["encrypted_archive_path"])
        events.append(event)
    return head, events, all_paths


def _read_archive_bytes(path: str) -> bytes:
    descriptor = os.open(
        path,
        os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0),
    )
    try:
        before = os.fstat(descriptor)
        if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= MAX_ARCHIVE_BYTES:
            raise ContractError("E_SUBMISSION_ARCHIVE")
        chunks: list[bytes] = []
        while True:
            chunk = os.read(descriptor, 1024 * 1024)
            if not chunk:
                break
            chunks.append(chunk)
        after = os.fstat(descriptor)
        if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) != (
            after.st_dev,
            after.st_ino,
            after.st_size,
            after.st_mtime_ns,
        ):
            raise ContractError("E_SUBMISSION_CHANGED")
        content = b"".join(chunks)
        if len(content) != after.st_size:
            raise ContractError("E_SUBMISSION_CHANGED")
        return content
    finally:
        os.close(descriptor)


def _within_window(received_at: datetime) -> bool:
    opened = datetime.strptime(
        INTAKE_SCHEDULE["development_open"], "%Y-%m-%dT%H:%M:%SZ"
    ).replace(tzinfo=timezone.utc)
    closed = datetime.strptime(
        INTAKE_SCHEDULE["development_close"], "%Y-%m-%dT%H:%M:%SZ"
    ).replace(tzinfo=timezone.utc)
    return opened <= received_at < closed


def _eligible_events(events: list[Mapping[str, Any]]) -> list[Mapping[str, Any]]:
    return [
        event
        for event in events
        if event_is_leaderboard_eligible(
            event,
            contract=CONTRACT,
            portal_contract_sha256=CONTRACT_SHA256,
        )
    ]


def _leaderboard_payload(events: list[Mapping[str, Any]]) -> dict[str, Any]:
    return development_leaderboard(_eligible_events(events))


def _current_rank(
    event: Mapping[str, Any], events: list[Mapping[str, Any]]
) -> int | None:
    rank_events = _eligible_events(events)
    if (
        CONTRACT["mode"] == "rehearsal"
        and CONTRACT["publication_stage"] == "hosted_rehearsal"
    ):
        # Rehearsal submissions stay out of the public projection, but the
        # private submit response still exercises deterministic ranking.
        rank_events = [
            stored
            for stored in events
            if stored["portal_contract_sha256"] == CONTRACT_SHA256
        ]
    identity = team_name_identity(event["team_display_name"])
    for row in development_leaderboard(rank_events)["rows"]:
        if team_name_identity(row["team_name"]) == identity:
            return row["rank"]
    return None


def _success(
    event: Mapping[str, Any],
    events: list[Mapping[str, Any]],
    *,
    replayed: bool,
) -> dict[str, Any]:
    return {
        "status": "accepted",
        "replayed": replayed,
        "team_name": event["team_display_name"],
        "final_answer_score": event["scores"]["seen_fac"],
        "reasoning_steps_score": event["scores"]["seen_checkpoint"],
        "receipt_id": event_receipt_id(event),
        "current_rank": _current_rank(event, events),
        "accepted_at": event["accepted_at"],
    }


def _persist(
    event: Mapping[str, Any],
    encrypted_archive: bytes,
    vault_token: str,
    *,
    parent_commit: str,
    known_paths: frozenset[str],
) -> tuple[Mapping[str, Any], list[Mapping[str, Any]]]:
    if hashlib.sha256(encrypted_archive).hexdigest() != event["encrypted_archive_sha256"]:
        raise ContractError("E_ENCRYPTED_ARCHIVE_CHANGED")
    event_bytes = canonical_json_bytes(event) + b"\n"
    event_path = submission_event_path(event["submission_id"])
    target_paths = {event["encrypted_archive_path"], event_path}
    if target_paths & known_paths:
        raise ContractError("E_STORAGE_PATH_EXISTS")
    operations = [
        CommitOperationAdd(
            path_in_repo=event["encrypted_archive_path"],
            path_or_fileobj=io.BytesIO(encrypted_archive),
        ),
        CommitOperationAdd(
            path_in_repo=event_path,
            path_or_fileobj=io.BytesIO(event_bytes),
        ),
    ]
    if not _within_window(datetime.now(timezone.utc)):
        raise ContractError("E_SUBMISSION_WINDOW_CLOSED")
    try:
        API.create_commit(
            repo_id=CONTRACT["vault_repo_id"],
            repo_type="dataset",
            token=vault_token,
            parent_commit=parent_commit,
            operations=operations,
            commit_message=f"Accept {event['submission_id']}",
        )
    except Exception:
        # Never retry an ambiguous write. The readback below decides whether an
        # exact logical duplicate won or the operation failed.
        pass
    try:
        _verified_head, verified_events, _verified_paths = _load_events(vault_token)
    except Exception as exc:
        raise ContractError("E_STORAGE_UNAVAILABLE") from exc
    requested_replay_key = make_logical_replay_key(
        event_team_id(event),
        event["archive_sha256"],
        event["release_manifest_sha256"],
    )
    for stored in verified_events:
        stored_replay_key = make_logical_replay_key(
            event_team_id(stored),
            stored["archive_sha256"],
            stored["release_manifest_sha256"],
        )
        if stored_replay_key != requested_replay_key:
            continue
        try:
            stored_archive = _download(
                CONTRACT["vault_repo_id"],
                stored["encrypted_archive_path"],
                vault_token,
                revision=_verified_head,
                error_code="E_STORAGE_UNAVAILABLE",
            ).read_bytes()
        except Exception as exc:
            raise ContractError("E_STORAGE_UNAVAILABLE") from exc
        if (
            hashlib.sha256(stored_archive).hexdigest()
            != stored["encrypted_archive_sha256"]
        ):
            raise ContractError("E_STORAGE_CONFLICT")
        return stored, verified_events
    raise ContractError("E_STORAGE_UNAVAILABLE")


def submit(
    team_name: str,
    upload_path: str | None,
) -> dict[str, Any]:
    try:
        received = datetime.now(timezone.utc).replace(microsecond=0)
        display_name = normalize_team_name(team_name)
        team = {
            "team_id": team_id_from_name(display_name),
            "display_name": display_name,
        }
        reference_token, vault_token = _tokens()
        reference_revision = _reference_revision()
        if not upload_path:
            raise ContractError("E_UPLOAD_REQUIRED")
        admission = admit_production_leaderboard_submission(
            ROOT / "data" / "leaderboard_questions.jsonl",
            ROOT / "data" / "leaderboard_expected_ids.json",
            upload_path,
            release_manifest_path=ROOT / "data" / "release_manifest.json",
            expected_release_manifest_sha256=CONTRACT["release_manifest_sha256"],
        )
        if admission["status"] != "PASS":
            error_codes = sorted(
                {
                    error["code"]
                    for error in admission.get("errors", [])
                    if isinstance(error, Mapping)
                    and isinstance(error.get("code"), str)
                }
            )
            print(
                json.dumps(
                    {
                        "event": "submission_rejected",
                        "stage": "admission",
                        "error_codes": error_codes,
                    },
                    sort_keys=True,
                    separators=(",", ":"),
                ),
                flush=True,
            )
            if CONTRACT["mode"] == "rehearsal" and error_codes:
                raise ContractError(
                    "E_SUBMISSION_REJECTED[" + ",".join(error_codes) + "]"
                )
            raise ContractError("E_SUBMISSION_REJECTED")
        archive = read_submission_archive(upload_path)
        if (
            not archive.valid
            or archive.archive_sha256 != admission["archive_sha256"]
            or archive.predictions_jsonl_sha256 != admission["predictions_jsonl_sha256"]
        ):
            raise ContractError("E_SUBMISSION_CHANGED")
        idempotency_key = make_idempotency_key(
            team["team_id"],
            archive.archive_sha256,
            CONTRACT["release_manifest_sha256"],
            CONTRACT_SHA256,
        )
        logical_replay_key = make_logical_replay_key(
            team["team_id"],
            archive.archive_sha256,
            CONTRACT["release_manifest_sha256"],
        )
        quota = CONTRACT["quota"]
        _initial_head, events, _initial_paths = _load_events(vault_token)
        state = quota_state(
            events,
            team_id=team["team_id"],
            accepted_at=received.strftime("%Y-%m-%dT%H:%M:%SZ"),
            idempotency_key=idempotency_key,
            logical_replay_key=logical_replay_key,
            per_utc_day=quota["per_utc_day"],
            total=quota["total"],
        )
        if state["duplicate"] is not None:
            return _success(state["duplicate"], events, replayed=True)
        if not _within_window(received):
            raise ContractError("E_SUBMISSION_WINDOW_CLOSED")
        if not state["allowed"]:
            raise ContractError("E_QUOTA_EXCEEDED")
        questions_path = ROOT / "data" / "leaderboard_questions.jsonl"
        questions = _index_jsonl(questions_path, "questions")
        gold_path = _verified_reference(
            "leaderboard_gold.jsonl",
            CONTRACT["leaderboard_gold_sha256"],
            reference_token,
            revision=reference_revision,
        )
        manifest_path = _verified_reference(
            "leaderboard_manifest.jsonl",
            CONTRACT["leaderboard_manifest_sha256"],
            reference_token,
            revision=reference_revision,
        )
        gold = _index_jsonl(gold_path, "gold")
        _manifest_bytes, loaded_manifest = load_jsonl_snapshot(manifest_path)
        if not loaded_manifest.valid:
            raise ContractError("E_REFERENCE_MANIFEST")
        report = score_submission(questions, gold, archive.records, loaded_manifest.records)
        if report.seen_fac is None or report.seen_checkpoint is None:
            raise ContractError("E_SCORING_CONTRACT")
        score_report_sha256 = hashlib.sha256(
            canonical_json_bytes(report.as_dict())
        ).hexdigest()
        archive_bytes = _read_archive_bytes(upload_path)
        if hashlib.sha256(archive_bytes).hexdigest() != archive.archive_sha256:
            raise ContractError("E_SUBMISSION_CHANGED")
        # Re-read immediately before the only state-changing call.
        final_head, final_events, final_paths = _load_events(vault_token)
        commit_time = datetime.now(timezone.utc).replace(microsecond=0)
        accepted_at = commit_time.strftime("%Y-%m-%dT%H:%M:%SZ")
        final_state = quota_state(
            final_events,
            team_id=team["team_id"],
            accepted_at=accepted_at,
            idempotency_key=idempotency_key,
            logical_replay_key=logical_replay_key,
            per_utc_day=quota["per_utc_day"],
            total=quota["total"],
        )
        if final_state["duplicate"] is not None:
            return _success(final_state["duplicate"], final_events, replayed=True)
        if not _within_window(commit_time):
            raise ContractError("E_SUBMISSION_WINDOW_CLOSED")
        if not final_state["allowed"]:
            raise ContractError("E_QUOTA_EXCEEDED")
        submission_id = make_submission_id(
            accepted_at, team["team_id"], archive.archive_sha256
        )
        encrypted_archive = encrypt_submission_archive(
            archive_bytes,
            CONTRACT["vault_recipient_public_key_b64"],
            submission_id=submission_id,
            release_manifest_sha256=CONTRACT["release_manifest_sha256"],
            portal_contract_sha256=CONTRACT_SHA256,
        )
        event = build_submission_event(
            accepted_at=accepted_at,
            team=team,
            release_version=CONTRACT["release_version"],
            source_commit=CONTRACT["source_commit"],
            portal_contract_sha256=CONTRACT_SHA256,
            reference_revision=reference_revision,
            archive_sha256=archive.archive_sha256,
            predictions_jsonl_sha256=archive.predictions_jsonl_sha256,
            release_manifest_sha256=CONTRACT["release_manifest_sha256"],
            questions_jsonl_sha256=CONTRACT["leaderboard_questions_sha256"],
            expected_ids_json_sha256=CONTRACT["leaderboard_expected_ids_sha256"],
            gold_jsonl_sha256=CONTRACT["leaderboard_gold_sha256"],
            manifest_jsonl_sha256=CONTRACT["leaderboard_manifest_sha256"],
            score_report_sha256=score_report_sha256,
            encryption_public_key_sha256=ENCRYPTION_PUBLIC_KEY_SHA256,
            encrypted_archive_sha256=hashlib.sha256(encrypted_archive).hexdigest(),
            seen_fac=report.seen_fac,
            seen_checkpoint=report.seen_checkpoint,
        )
        stored, verified_events = _persist(
            event,
            encrypted_archive,
            vault_token,
            parent_commit=final_head,
            known_paths=final_paths,
        )
        return _success(
            stored,
            verified_events,
            replayed=stored["submission_id"] != event["submission_id"],
        )
    except ContractError as exc:
        code = str(exc)
        if not code.startswith("E_"):
            code = "E_SUBMISSION_REJECTED"
        return {"status": "not_accepted", "replayed": False, "error_code": code}
    except Exception:
        return {
            "status": "not_accepted",
            "replayed": False,
            "error_code": "E_PLATFORM_UNAVAILABLE",
        }


def leaderboard_payload() -> dict[str, Any]:
    _reference_token, vault_token = _tokens()
    _head, events, _paths = _load_events(vault_token)
    return _leaderboard_payload(events)


def leaderboard_rows() -> list[list[Any]]:
    try:
        projection = leaderboard_payload()
        return [
            [
                row["team_name"],
                row["rank"],
                row["final_answer_score"],
                row["reasoning_steps_score"],
                row["accepted_at"],
            ]
            for row in projection["rows"]
        ]
    except ContractError as exc:
        raise gr.Error("Leaderboard temporarily unavailable") from exc
    except Exception as exc:
        raise gr.Error("Leaderboard temporarily unavailable") from exc


LEADERBOARD_CORS_HEADERS = {
    "Access-Control-Allow-Origin": "https://the-finai.github.io",
    "Access-Control-Allow-Methods": "GET, HEAD",
    "Vary": "Origin",
}


web = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
web.add_middleware(
    CORSMiddleware,
    allow_origins=["https://the-finai.github.io"],
    allow_methods=["GET", "HEAD"],
    allow_headers=[],
    allow_credentials=False,
)


@web.get("/api/leaderboard")
@web.head("/api/leaderboard")
def api_leaderboard() -> JSONResponse:
    try:
        return JSONResponse(
            leaderboard_payload(), headers=LEADERBOARD_CORS_HEADERS
        )
    except Exception:
        return JSONResponse(
            {"detail": "Leaderboard temporarily unavailable"},
            status_code=503,
            headers=LEADERBOARD_CORS_HEADERS,
        )


with gr.Blocks(title="FinReason Task 1 Development Leaderboard") as demo:
    if CONTRACT["mode"] == "rehearsal":
        gr.Markdown(
            "## NON-PRODUCTION HOSTED REHEARSAL\n"
            "Organizer smoke test only. Results and submissions are not official."
        )
    public_dataset_url = immutable_public_dataset_url(CONTRACT)
    gr.Markdown(
        "# FinReason Task 1 Development Leaderboard\n"
        "Enter your Team Name and upload the canonical ZIP. A successful submission "
        "immediately returns Final answer score, Reasoning steps score, a private "
        "receipt ID, and current rank. No case-level feedback is exposed.\n\n"
        f"Submissions must arrive before {INTAKE_SCHEDULE['development_close']} "
        "(exclusive UTC cutoff).\n\n"
        f"[Immutable train + labeled dev + 580-case leaderboard-development participant dataset]({public_dataset_url})"
    )
    team_name = gr.Textbox(
        label="Team Name",
        placeholder="Your public leaderboard name",
        max_lines=1,
    )
    upload = gr.File(label="Canonical submission ZIP", file_types=[".zip"], type="filepath")
    submit_button = gr.Button("Submit", variant="primary")
    status = gr.JSON(label="Submission result")
    submit_button.click(
        submit,
        inputs=[team_name, upload],
        outputs=[status],
        api_name="submit",
    )
    table = gr.Dataframe(
        headers=[
            "Team Name",
            "Rank",
            "Final answer score",
            "Reasoning steps score",
            "Accepted at (UTC)",
        ],
        datatype=["str", "number", "str", "str", "str"],
        interactive=False,
    )
    refresh = gr.Button("Refresh leaderboard")
    refresh.click(leaderboard_rows, inputs=[], outputs=[table], api_name=False)
    demo.load(leaderboard_rows, inputs=[], outputs=[table], api_name=False)


app = gr.mount_gradio_app(
    web,
    demo.queue(default_concurrency_limit=1, max_size=32),
    path="/",
    max_file_size=MAX_ARCHIVE_BYTES,
    run_history=False,
)


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)