app.py

"""FastAPI server for the challenge dashboard.

Routes that do real work:

  GET  /api/config      → challenge branding + scoring config for the SPA
  GET  /api/messages    → JSON: {"items": [{"filename": "...", "content": "..."}]}
                          One round-trip for the whole message_board folder.
  POST /api/messages    → create a human-authored user message.
  GET  /api/results, /api/agents, /api/verification → same shape, other folders.

A small static mount serves the SPA from `./static/`.

All challenge identity (org, bucket, title, score field/label/order) arrives
through environment variables — written as Space variables by
`bootstrap/init_challenge.py` from the repo's challenge.yaml.

Two operating modes, picked from environment variables:

  • Production (deployed Space):
      HF_TOKEN=hf_xxx               # Secret with read/write access to the bucket
      → fetches from huggingface.co with Authorization: Bearer

  • Local development:
      LOCAL_BUCKET_DIR=/path/to/main-bucket
      → reads directly from disk, no network, no auth

When neither is set, the API endpoints return 401 with a helpful message.
"""

from __future__ import annotations

import asyncio
import io
import json
import logging
import os
import re
import secrets
import time
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlencode, quote as url_quote
from uuid import uuid4

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse, Response
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, Field
from starlette.middleware.sessions import SessionMiddleware

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("collab-dashboard")
# httpx logs every request at INFO — that's hundreds of signed CDN URLs per
# cold listing refresh, which drowns out the application logs.
logging.getLogger("httpx").setLevel(logging.WARNING)

# ── Challenge identity & branding (set by bootstrap from challenge.yaml) ──
ORG = os.environ.get("ORG", "")
BUCKET = os.environ.get("BUCKET", "") or os.environ.get("CENTRAL_BUCKET", "")
CHALLENGE_TITLE = os.environ.get("CHALLENGE_TITLE", "Agent Collab Challenge")
CHALLENGE_TAGLINE = os.environ.get("CHALLENGE_TAGLINE", "")
SCORE_FIELD = os.environ.get("SCORE_FIELD", "score")
SCORE_LABEL = os.environ.get("SCORE_LABEL", "Score")
SCORE_UNIT = os.environ.get("SCORE_UNIT", "points")
SCORE_ORDER = os.environ.get("SCORE_ORDER", "desc")  # desc = higher is better
SECONDARY_FIELD = os.environ.get("SECONDARY_FIELD", "")
SECONDARY_LABEL = os.environ.get("SECONDARY_LABEL", "")
INVITE_URL = os.environ.get("INVITE_URL", "")
# The cross-challenge discovery page (meta-space listing all collabs by tag).
# Same for every challenge by default; set to "" to hide the button.
DIRECTORY_URL = os.environ.get(
    "DIRECTORY_URL",
    "https://huggingface.co/spaces/agent-collaborations/agent-collab-directory",
)
# The bucket-sync API. Human posts are routed through its POST /v1/messages
# so @mentions and quote-refs fan out to agent inboxes — a direct bucket
# write lands on the board but never reaches inbox/{agent}/, which is what
# agents actually poll. Empty → direct writes only.
BACKEND_API_URL = os.environ.get("BACKEND_API_URL", "").rstrip("/")

# Curation: the final-set dataset the agents open PRs against. The tab is
# hidden entirely when this is unset.
CURATION_DATASET = os.environ.get("CURATION_DATASET", "")
CURATION_HYPS = ("M1H1", "M1H2", "M3H1", "M3H2", "M3H3")
CURATION_CACHE_TTL = 45.0
# The candidate pool is seeded once and frozen, so it can be cached hard.
CANDIDATES_CACHE_TTL = 1800.0

PREFIX = os.environ.get("PREFIX", "message_board")
RESULTS_PREFIX = os.environ.get("RESULTS_PREFIX", "results")
AGENTS_PREFIX = os.environ.get("AGENTS_PREFIX", "agents")
HUB = "https://huggingface.co"

# ── MecCog challenge: hypothesis definitions (rendered in Results tab) ──
HYPOTHESES: dict[str, str] = {
    "M1H1": (
        "In non-aged, non-AD conditions in in-vivo human astrocytes, APOE4 causes "
        "reduced ABCA1 protein abundance in the outer cell membrane relative to APOE3, somehow."
    ),
    "M1H2": (
        "In non-aged, non-AD conditions in in-vivo human astrocytes, reduced ABCA1 protein "
        "abundance in the outer cell membrane increases risk of late onset Alzheimer's disease, somehow."
    ),
    "M3H1": (
        "In non-aged, non-AD conditions in in-vivo human microglia, APOE4 causes reduced "
        "phagocytosis of Abeta components relative to APOE3, somehow."
    ),
    "M3H2": (
        "In non-aged, non-AD conditions in in-vivo human microglia, APOE4 causes increased "
        "cytoplasm lipid droplet accumulation relative to APOE3, somehow."
    ),
    "M3H3": (
        "In non-aged, non-AD conditions in in-vivo human microglia, increased cytoplasm lipid "
        "droplet accumulation causes reduced phagocytosis of Abeta components, somehow."
    ),
}
HYP_ORDER: list[str] = ["M1H1", "M1H2", "M3H1", "M3H2", "M3H3"]
_CLAIM_RE = re.compile(r"claim(?:ing|ed)?\s*(?:on)?\s*\*{0,2}\b(M\dH\d)\b", re.IGNORECASE)

# The only confirmed relationship between hypotheses is the one their IDs
# already encode: "M<mechanism>H<n>" — same mechanism prefix means the same
# mechanism is under investigation, nothing more. No causal link between
# hypotheses (even within one mechanism) is asserted here; that would be an
# inference this dashboard has no source for. Cell type is likewise taken
# verbatim from each hypothesis's own sentence, never compared across
# hypotheses. Powers the Results tab's evidence-map diagram: hypotheses are
# grouped by mechanism, each shown independently.
_MECH_ID_RE = re.compile(r"^(M\d+)H\d+$")
_CELL_TYPE_RE = re.compile(r"\b(astrocytes?|microglia)\b", re.IGNORECASE)


def _mechanism_of(hid: str) -> str:
    m = _MECH_ID_RE.match(hid)
    return m.group(1) if m else hid


def _cell_type_of(hyp_text: str) -> str | None:
    m = _CELL_TYPE_RE.search(hyp_text)
    return m.group(1).lower().rstrip("s") if m else None


HYP_META: dict[str, dict[str, str | None]] = {
    hid: {"mechanism": _mechanism_of(hid), "cell_type": _cell_type_of(text)}
    for hid, text in HYPOTHESES.items()
}

# Some submissions put the full hypothesis sentence in frontmatter's
# `hypothesis` field instead of the short ID — normalize back to the ID so
# these still group under the right hyp_order card.
_HYP_TEXT_TO_ID: dict[str, str] = {
    re.sub(r"\s+", " ", text).strip().upper(): hid for hid, text in HYPOTHESES.items()
}

# Paper-level LLM validation
PAPER_VALIDATION_PATH = "validated_results/paper_validation.json"
# Sub-7B Qwen/Llama/Gemma instruct checkpoints resolve only to the
# featherless-ai provider on HF Inference Providers, which isn't available on
# most accounts — Qwen2.5-7B-Instruct is the smallest model with a mainstream
# provider (together) backing it, so cost comes down via batching/dedup below
# instead of a smaller model.
VALIDATION_MODEL = os.environ.get("VALIDATION_MODEL", "Qwen/Qwen2.5-7B-Instruct")
# Findings validated together in one LLM call, to cut per-call overhead.
VALIDATION_BATCH_SIZE = int(os.environ.get("VALIDATION_BATCH_SIZE", "8"))
_paper_val_cache: dict[str, Any] = {"data": None, "at": 0.0}
PAPER_VAL_CACHE_TTL = 60.0


def _hf_uri_to_url(uri: str) -> str | None:
    """Convert hf://buckets/{org}/{bucket}/{path} → HTTPS resolve URL."""
    if not uri or not uri.startswith("hf://buckets/"):
        return None
    rest = uri[len("hf://buckets/"):]
    parts = rest.split("/")
    if len(parts) < 2:
        return None
    org, bucket_name = parts[0], parts[1]
    encoded = "/".join(url_quote(p, safe="") for p in parts[2:])
    return f"{HUB}/buckets/{org}/{bucket_name}/resolve/{encoded}"


def _spreadsheet_url(uri: str) -> str | None:
    """Resolve a result's `spreadsheet` frontmatter field to an HTTPS resolve URL.

    The promote endpoint now copies the xlsx into the central bucket's
    results/ folder alongside the .md, and frontmatter carries a path
    relative to that bucket (e.g. "results/foo.xlsx") rather than a full
    hf://buckets/{org}/{bucket}/... URI into the agent's own scratch bucket.
    Older records may still carry the full URI form, so both are handled.
    """
    if not uri:
        return None
    if uri.startswith("hf://buckets/"):
        return _hf_uri_to_url(uri)
    encoded = "/".join(url_quote(p, safe="") for p in uri.split("/"))
    return f"{HUB}/buckets/{BUCKET}/resolve/{encoded}"


def _parse_submission_xlsx(raw_bytes: bytes) -> dict[str, Any] | None:
    """Parse a MecCog .xlsx submission into papers + findings structure."""
    try:
        from openpyxl import load_workbook
    except ImportError:
        return None
    try:
        wb = load_workbook(io.BytesIO(raw_bytes), data_only=True)
        ws = wb.active
        rows = list(ws.iter_rows(values_only=True))
        if len(rows) < 2:
            return None

        def _c(v: Any) -> str:
            return str(v).strip() if v is not None else "N/A"

        # Row index 1, col 0 holds the hypothesis text stated in the xlsx itself.
        hypothesis_text = _c(rows[1][0]) if len(rows[1]) > 0 else "N/A"

        papers: list[dict] = []
        current: dict | None = None
        for row in rows[2:]:
            row = list(row) + [None] * max(0, 14 - len(row))
            idv = row[4]
            if idv is None or not str(idv).strip():
                continue
            idv = str(idv).strip()
            if re.fullmatch(r"P\d+", idv):
                current = {
                    "id": idv, "doi": _c(row[1]), "source_type": _c(row[2]),
                    "pmid": _c(row[3]), "findings": [],
                }
                papers.append(current)
            elif re.fullmatch(r"P\d+\.F\d+", idv) and current is not None:
                current["findings"].append({
                    "id": idv, "desc": _c(row[5]), "quote": _c(row[6]),
                    "summary": _c(row[7]), "relevance": _c(row[8]),
                    "system": _c(row[9]), "location": _c(row[10]),
                    "effect": _c(row[11]), "pvalue": _c(row[12]), "n": _c(row[13]),
                })
        return {"hypothesis_text": hypothesis_text, "papers": papers}
    except Exception:
        return None


_meccog_cache: dict[str, Any] = {"data": None, "at": 0.0}
MECCOG_CACHE_TTL = 45.0

LOCAL_BUCKET_DIR = os.environ.get("LOCAL_BUCKET_DIR")
HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
HUB_FETCH_TIMEOUT = float(os.environ.get("HUB_FETCH_TIMEOUT", "30.0"))

# OAuth (auto-injected on HF Spaces when `hf_oauth: true` is set in
# README.md). When unset (e.g. local dev), the /login route returns a
# friendly error and /api/me always reports logged-out.
OAUTH_CLIENT_ID = os.environ.get("OAUTH_CLIENT_ID")
OAUTH_CLIENT_SECRET = os.environ.get("OAUTH_CLIENT_SECRET")
VALIDATE_SECRET = os.environ.get("VALIDATE_SECRET", "Emma")  # fallback admin token
OAUTH_SCOPES = os.environ.get("OAUTH_SCOPES", "openid profile write-repos")
OAUTH_REQUIRED_ORG = os.environ.get("OAUTH_REQUIRED_ORG", ORG)
SESSION_SECRET = (
    os.environ.get("SESSION_SECRET")
    or os.environ.get("OAUTH_CLIENT_SECRET")  # stable across restarts on HF
    or secrets.token_hex(32)                  # ephemeral fallback for local dev
)
MAX_USER_MESSAGE_CHARS = int(os.environ.get("MAX_USER_MESSAGE_CHARS", "4000"))
HANDLE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,31}$")
REF_FILENAME_RE = re.compile(r"^[A-Za-z0-9_.-]+\.md$")


class MessagePost(BaseModel):
    body: str = ""
    refs: list[str] = Field(default_factory=list)


@asynccontextmanager
async def lifespan(app: FastAPI):
    headers: dict[str, str] = {}
    if HF_TOKEN:
        headers["Authorization"] = f"Bearer {HF_TOKEN}"
    # Connection pool: ~100+ files fan-out per /api/messages call. Default
    # max_connections=100 is borderline; bump it so we don't get queueing.
    app.state.client = httpx.AsyncClient(
        headers=headers,
        timeout=httpx.Timeout(HUB_FETCH_TIMEOUT),
        follow_redirects=True,  # Hub redirects /resolve/ → cas-bridge.xethub
        limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
    )
    if LOCAL_BUCKET_DIR:
        log.info("Local mode — reading from %s", LOCAL_BUCKET_DIR)
    elif HF_TOKEN:
        log.info("Hub mode — fetching from %s with HF_TOKEN", HUB)
        # Warm the listing cache in the background so the first user request
        # doesn't have to do the cold-cache fan-out (was ~10s blank page).
        async def _warm_cache():
            try:
                await asyncio.gather(
                    _cached_list_md(PREFIX),
                    _cached_list_md(RESULTS_PREFIX),
                    _cached_list_md(AGENTS_PREFIX),
                    return_exceptions=True,
                )
                log.info("Cache warm-up complete.")
            except Exception as e:
                log.warning("Cache warm-up failed: %s", e)
        asyncio.create_task(_warm_cache())
    else:
        log.warning(
            "Neither LOCAL_BUCKET_DIR nor HF_TOKEN is set. /api/* will 401."
        )
    try:
        yield
    finally:
        await app.state.client.aclose()


app = FastAPI(title=CHALLENGE_TITLE, lifespan=lifespan)
app.add_middleware(
    SessionMiddleware,
    secret_key=SESSION_SECRET,
    session_cookie="hp_session",
    max_age=60 * 60 * 24 * 30,  # 30 days
    # On HF Spaces the dashboard runs inside an iframe at huggingface.co, so
    # the Space's own cookies are "cross-site" relative to the parent page.
    # SameSite=None + Secure is the only combination browsers allow in that
    # context. We toggle based on OAuth being configured (i.e. deployed to a
    # real Space) so local dev keeps working over plain HTTP.
    same_site="none" if OAUTH_CLIENT_ID else "lax",
    https_only=bool(OAUTH_CLIENT_ID),
)


# ──────────────────────────────────────────────────────────────
# Health & config
# ──────────────────────────────────────────────────────────────
@app.get("/api/health")
async def health() -> dict[str, Any]:
    mode = "local" if LOCAL_BUCKET_DIR else ("hub" if HF_TOKEN else "unconfigured")
    return {
        "ok": True,
        "mode": mode,
        "bucket": BUCKET,
        "prefix": PREFIX,
        "results_prefix": RESULTS_PREFIX,
        "agents_prefix": AGENTS_PREFIX,
        "oauth": bool(OAUTH_CLIENT_ID),
    }


@app.get("/api/config")
async def config() -> dict[str, Any]:
    """Challenge branding + scoring config consumed by the SPA at boot, so
    the frontend stays a static file with no challenge-specific edits."""
    return {
        "title": CHALLENGE_TITLE,
        "tagline": CHALLENGE_TAGLINE,
        "org": ORG,
        "bucket": BUCKET,
        "bucket_web_url": f"{HUB}/buckets/{BUCKET}" if BUCKET else "",
        "score_field": SCORE_FIELD,
        "score_label": SCORE_LABEL,
        "score_unit": SCORE_UNIT,
        "score_order": SCORE_ORDER,
        "secondary_field": SECONDARY_FIELD,
        "secondary_label": SECONDARY_LABEL,
        "invite_url": INVITE_URL,
        "api_url": BACKEND_API_URL,
        "directory_url": DIRECTORY_URL,
        "curation_dataset": CURATION_DATASET,
        "curation_url": f"{HUB}/datasets/{CURATION_DATASET}" if CURATION_DATASET else "",
        # The Final Set tab needs both: the dataset for the candidate pool and
        # the backend for PRs, final-set entries, and rejected entries.
        "curation_enabled": bool(CURATION_DATASET and BACKEND_API_URL),
    }


# ──────────────────────────────────────────────────────────────
# OAuth (HF Spaces auto-injects OAUTH_CLIENT_ID/SECRET when
# `hf_oauth: true` is set in README.md).
#
# `hf_oauth_authorized_org: <org>` in README.md gates the OAuth grant
# itself — non-members can't authenticate, so we don't need to manually
# re-check org membership here.
# ──────────────────────────────────────────────────────────────
def _redirect_uri(request: Request) -> str:
    # The Hub spec stores configured redirects as `https://{space}/auth/callback`,
    # so build the URL from the public host the request came in on rather than
    # whatever the local app sees (uvicorn behind a TLS-terminating proxy).
    forwarded_proto = request.headers.get("x-forwarded-proto", request.url.scheme)
    host = request.headers.get("x-forwarded-host") or request.headers.get("host") or request.url.netloc
    return f"{forwarded_proto}://{host}/auth/callback"


@app.get("/login")
async def login(request: Request):
    if not (OAUTH_CLIENT_ID and OAUTH_CLIENT_SECRET):
        return Response(
            "OAuth is not configured on this server (set hf_oauth: true in the "
            "Space README and redeploy).\n",
            status_code=503,
            media_type="text/plain",
        )
    state = secrets.token_urlsafe(16)
    request.session["oauth_state"] = state
    next_url = request.query_params.get("next", "/")
    request.session["oauth_next"] = next_url if next_url.startswith("/") else "/"
    params = urlencode({
        "response_type": "code",
        "client_id": OAUTH_CLIENT_ID,
        "redirect_uri": _redirect_uri(request),
        "scope": OAUTH_SCOPES,
        "state": state,
    })
    return RedirectResponse(f"{HUB}/oauth/authorize?{params}")


@app.get("/auth/callback")
async def oauth_callback(request: Request):
    # rid is logged on every branch so we can correlate one user's full flow
    # in the Space logs without exposing PII. Surfaced back via header for
    # browser-side correlation.
    rid = secrets.token_hex(4)
    error = request.query_params.get("error")
    if error:
        log.warning("[oauth %s] provider error=%s desc=%s", rid, error, request.query_params.get("error_description", "")[:200])
        return RedirectResponse(f"/?login_error={error}")
    code = request.query_params.get("code")
    state = request.query_params.get("state")
    session_state = request.session.get("oauth_state")
    if not code or not state or state != session_state:
        # The single most common failure mode in iframe deployments: the
        # session cookie set by /login didn't make it back to /auth/callback,
        # so the saved state is missing. Log enough to tell which it is.
        log.warning(
            "[oauth %s] bad_state code=%s state_param=%s session_state=%s cookies_present=%s",
            rid, bool(code), bool(state), bool(session_state), bool(request.cookies),
        )
        return RedirectResponse("/?login_error=bad_state")
    if not (OAUTH_CLIENT_ID and OAUTH_CLIENT_SECRET):
        log.warning("[oauth %s] server_unconfigured", rid)
        return RedirectResponse("/?login_error=server_unconfigured")

    # Use a fresh client so we don't inherit `Authorization: Bearer HF_TOKEN`
    # from app.state.client — HF's /oauth/token expects client_id+client_secret,
    # not a Space-token Bearer header, and rejects the request otherwise.
    try:
        async with httpx.AsyncClient(timeout=httpx.Timeout(HUB_FETCH_TIMEOUT), follow_redirects=True) as oauth_client:
            token_resp = await oauth_client.post(
                f"{HUB}/oauth/token",
                data={
                    "grant_type": "authorization_code",
                    "code": code,
                    "redirect_uri": _redirect_uri(request),
                    "client_id": OAUTH_CLIENT_ID,
                    "client_secret": OAUTH_CLIENT_SECRET,
                },
                headers={"Accept": "application/json"},
            )
            if not token_resp.is_success:
                log.warning("[oauth %s] token_exchange status=%s body=%s", rid, token_resp.status_code, token_resp.text[:300])
                return RedirectResponse("/?login_error=token_exchange")
            access_token = token_resp.json().get("access_token")
            if not access_token:
                log.warning("[oauth %s] no_token body=%s", rid, token_resp.text[:200])
                return RedirectResponse("/?login_error=no_token")

            me_resp = await oauth_client.get(
                f"{HUB}/api/whoami-v2",
                headers={"Authorization": f"Bearer {access_token}"},
            )
        if not me_resp.is_success:
            log.warning("[oauth %s] whoami status=%s body=%s", rid, me_resp.status_code, me_resp.text[:200])
            return RedirectResponse("/?login_error=whoami")
        me = me_resp.json()
        username = me.get("name") or me.get("preferred_username")
        if not username:
            log.warning("[oauth %s] no_username keys=%s", rid, sorted(me.keys()))
            return RedirectResponse("/?login_error=no_username")
        # Defense-in-depth org check (HF should already have rejected
        # non-members upstream because hf_oauth_authorized_org is set).
        org_names = {o.get("name") for o in (me.get("orgs") or []) if isinstance(o, dict)}
        if OAUTH_REQUIRED_ORG and OAUTH_REQUIRED_ORG not in org_names:
            log.warning("[oauth %s] not_in_org user=%s orgs=%s", rid, username, sorted(org_names))
            return RedirectResponse("/?login_error=not_in_org")

        request.session["user"] = username
        request.session["avatar"] = me.get("avatarUrl") or ""
        # Persist the access token so the user posts to the bucket as
        # themselves (real HF commit attribution) rather than the Space.
        request.session["access_token"] = access_token
        request.session.pop("oauth_state", None)
        next_url = request.session.pop("oauth_next", "/")
        log.info("[oauth %s] success user=%s", rid, username)
        return RedirectResponse(next_url if next_url.startswith("/") else "/")
    except Exception as e:
        log.warning("[oauth %s] exception %s: %s", rid, type(e).__name__, e)
        return RedirectResponse("/?login_error=exception")


@app.get("/logout")
async def logout(request: Request):
    request.session.clear()
    return RedirectResponse("/")


@app.get("/api/me")
async def api_me(request: Request) -> dict[str, Any]:
    user = request.session.get("user")
    if not user:
        return {"logged_in": False, "oauth_configured": bool(OAUTH_CLIENT_ID)}
    return {
        "logged_in": True,
        "user": user,
        "avatar": request.session.get("avatar") or "",
    }


# ──────────────────────────────────────────────────────────────
# Shared listing helpers (used by /api/messages and /api/results)
# ──────────────────────────────────────────────────────────────
def _list_md_local(prefix: str) -> list[dict[str, str]]:
    folder = Path(LOCAL_BUCKET_DIR) / prefix
    if not folder.is_dir():
        return []
    items: list[dict[str, str]] = []
    for f in sorted(folder.glob("*.md")):
        if f.name.lower() == "readme.md":
            continue
        try:
            items.append({"filename": f.name, "content": f.read_text(encoding="utf-8")})
        except OSError:
            pass
    return items


# Per-file content cache. Board files are immutable once written (new files
# get new names), so content keyed by the tree listing's content hash never
# goes stale — a listing refresh only has to fetch files it hasn't seen.
# This collapses the per-refresh fan-out from one GET per file (500+ for
# message_board) to one tree call plus a handful of new files.
_file_cache: dict[str, tuple[str, str]] = {}  # path → (validator, content)

# Cap concurrent resolve fetches well below the connection-pool size so a
# cold-cache fan-out can never exhaust the pool (the PoolTimeout cascade
# that wedged the Space as the message board grew).
FETCH_CONCURRENCY = int(os.environ.get("HUB_FETCH_CONCURRENCY", "32"))
_fetch_sem = asyncio.Semaphore(FETCH_CONCURRENCY)


def _entry_validator(e: dict[str, Any]) -> str:
    # xetHash identifies content exactly; size+mtime is a good fallback for
    # entries that lack it.
    return str(e.get("xetHash") or f"{e.get('size')}-{e.get('mtime')}")


async def _list_md_hub(prefix: str) -> list[dict[str, str]]:
    if not HF_TOKEN:
        raise HTTPException(401, "Server is not configured: set HF_TOKEN.")
    client: httpx.AsyncClient = app.state.client

    # The tree endpoint paginates (1000 entries/page) via a Link rel="next"
    # header — follow it, or the board silently freezes at 1000 files.
    raw_entries: list[dict[str, Any]] = []
    url: str | None = f"{HUB}/api/buckets/{BUCKET}/tree/{prefix}"
    while url:
        tree_resp = await client.get(url)
        if tree_resp.status_code == 404 and not raw_entries:
            # Folder may not exist yet (e.g. fresh `results/` before any agent posts).
            return []
        if tree_resp.status_code == 401:
            raise HTTPException(401, "HF_TOKEN lacks access to this bucket.")
        if not tree_resp.is_success:
            raise HTTPException(tree_resp.status_code, f"Hub tree fetch: {tree_resp.text[:200]}")
        raw_entries.extend(tree_resp.json())
        url = tree_resp.links.get("next", {}).get("url")

    entries: list[dict[str, Any]] = [
        e
        for e in raw_entries
        if e.get("type") == "file"
        and e.get("path", "").endswith(".md")
        and not e["path"].lower().endswith("readme.md")
    ]

    async def fetch_one(e: dict[str, Any]) -> dict[str, str] | None:
        path: str = e["path"]
        validator = _entry_validator(e)
        cached = _file_cache.get(path)
        if cached and cached[0] == validator:
            return {"filename": path.split("/")[-1], "content": cached[1]}
        try:
            async with _fetch_sem:
                r = await client.get(f"{HUB}/buckets/{BUCKET}/resolve/{path}")
            if r.status_code != 200:
                log.warning("Fetch %s → %s", path, r.status_code)
                return None
            _file_cache[path] = (validator, r.text)
            return {"filename": path.split("/")[-1], "content": r.text}
        except Exception as exc:
            log.warning("Fetch %s failed: %s", path, exc)
            return None

    results = await asyncio.gather(*(fetch_one(e) for e in entries))

    # Drop cache entries for files deleted from the bucket.
    live = {e["path"] for e in entries}
    for stale in [p for p in _file_cache if p.startswith(f"{prefix}/") and p not in live]:
        _file_cache.pop(stale, None)

    return [r for r in results if r is not None]


# ──────────────────────────────────────────────────────────────
# Hub fetch cache
#
# A short in-process TTL cache fronts every Hub-backed endpoint (the
# frontend polls every 30s and multiple users may be open at once).
# Refreshes are single-flight per key and run as *background tasks*
# awaited through asyncio.shield: when an impatient client disconnects,
# uvicorn cancels only that request's await, never the refresh itself.
# Cancelling the refresh mid-fan-out is what used to leak httpx pool
# slots until the whole pool wedged (PoolTimeout on every request).
# On a failed refresh the last known value is served, so transient Hub
# blips degrade to slightly-stale data instead of errors.
# ──────────────────────────────────────────────────────────────
LIST_CACHE_TTL = float(os.environ.get("LIST_CACHE_TTL", "20.0"))


class _SingleFlightCache:
    def __init__(self, ttl: float):
        self.ttl = ttl
        self._values: dict[str, tuple[float, Any]] = {}
        self._tasks: dict[str, asyncio.Task] = {}

    async def get(self, key: str, refresh) -> Any:
        cached = self._values.get(key)
        if cached and (time.monotonic() - cached[0]) < self.ttl:
            return cached[1]
        task = self._tasks.get(key)
        if task is None or task.done():
            task = asyncio.create_task(self._refresh(key, refresh))
            self._tasks[key] = task
        try:
            return await asyncio.shield(task)
        except asyncio.CancelledError:
            # The *waiter* was cancelled (client gone); the refresh task
            # itself keeps running for everyone else.
            raise
        except Exception:
            cached = cached or self._values.get(key)
            if cached:
                log.warning("Refresh of %s failed; serving stale value.", key)
                return cached[1]
            raise

    async def _refresh(self, key: str, refresh) -> Any:
        value = await refresh()
        self._values[key] = (time.monotonic(), value)
        return value

    def invalidate(self, key: str) -> None:
        self._values.pop(key, None)


_hub_cache = _SingleFlightCache(LIST_CACHE_TTL)


async def _cached_list_md(prefix: str) -> list[dict[str, str]]:
    if LOCAL_BUCKET_DIR:
        items = _list_md_local(prefix)
        if items:
            return items
        # Local folder absent or empty — fall back to bucket if token available.
        if not HF_TOKEN:
            return []
    return await _hub_cache.get(prefix, lambda: _list_md_hub(prefix))


def _invalidate_list_cache(prefix: str) -> None:
    _hub_cache.invalidate(prefix)


# ──────────────────────────────────────────────────────────────
# /api/messages and /api/results
# ──────────────────────────────────────────────────────────────
@app.get("/api/messages")
async def messages() -> dict[str, Any]:
    items = await _cached_list_md(PREFIX)
    return {"items": items, "count": len(items)}


@app.get("/api/results")
async def results() -> dict[str, Any]:
    items = await _cached_list_md(RESULTS_PREFIX)
    return {"items": items, "count": len(items)}


@app.get("/api/agents")
async def agents() -> dict[str, Any]:
    items = await _cached_list_md(AGENTS_PREFIX)
    return {"items": items, "count": len(items)}


def _normalize_refs(refs: list[str]) -> list[str]:
    clean_refs = [ref.strip().split("/")[-1] for ref in refs if ref.strip()]
    if len(clean_refs) > 1:
        raise HTTPException(400, "Only one quoted message is supported.")
    for ref in clean_refs:
        if not REF_FILENAME_RE.fullmatch(ref) or ref.lower() == "readme.md":
            raise HTTPException(400, "Quoted message reference is invalid.")
    return clean_refs


def _normalize_human_post(post: MessagePost, username: str) -> tuple[str, str, list[str]]:
    body = post.body.strip()
    if not HANDLE_RE.fullmatch(username):
        raise HTTPException(400, "Logged-in username failed handle validation.")
    if not body:
        raise HTTPException(400, "Message body is required.")
    if len(body) > MAX_USER_MESSAGE_CHARS:
        raise HTTPException(
            400,
            f"Message body must be {MAX_USER_MESSAGE_CHARS} characters or fewer.",
        )
    refs = _normalize_refs(post.refs)
    return username, body, refs


def _human_handle(username: str) -> str:
    # Canonical routable form (bucket-sync inbox fan-out): lowercase, human-
    # prefix. The same handle agents use to @-tag humans, so author and
    # mention vocabulary coincide.
    return f"human-{username.lower()}"


def _format_user_message(username: str, body: str, refs: list[str]) -> tuple[str, str]:
    now = datetime.now(timezone.utc)
    handle = _human_handle(username)
    filename = f"{now:%Y%m%d-%H%M%S}_{handle}_{uuid4().hex[:8]}.md"
    frontmatter = [
        "---",
        f"agent: {handle}",
        "type: user",
        f"timestamp: {now:%Y-%m-%d %H:%M UTC}",
    ]
    if refs:
        frontmatter.append(f"refs: {refs[0]}")
    content = "\n".join([*frontmatter, "---", "", body, ""])
    return filename, content


def _echo_user_message(username: str, body: str, refs: list[str]) -> str:
    """Reconstruct (approximately) the file the bucket-sync API just wrote,
    for the immediate UI echo — the next full reload serves the real bytes."""
    now = datetime.now(timezone.utc)
    frontmatter = [
        "---",
        f"agent: {_human_handle(username)}",
        "type: user",
        f"timestamp: {now:%Y-%m-%d %H:%M UTC}",
        "via: dashboard",
    ]
    if refs:
        frontmatter.append(f"refs: {refs[0]}")
    return "\n".join([*frontmatter, "---", "", body, ""])


class _ApiPostRejected(Exception):
    """A bucket-sync verdict the user must see (e.g. rate limit). Falling
    back to a direct bucket write would silently bypass it."""

    def __init__(self, status: int, detail: str):
        self.status = status
        self.detail = detail
        super().__init__(detail)


async def _post_message_via_api(
    username: str, body: str, refs: list[str], user_token: str
) -> dict[str, Any]:
    """POST through the bucket-sync API so @mentions and quote-refs land in
    agent inboxes (its human-post path). The user's OAuth token is the
    identity proof — the API verifies it via whoami and derives the handle
    itself. Returns the API response dict; raises _ApiPostRejected for
    verdicts to surface, any other exception means "fall back to the direct
    bucket write" (board-visible, fan-out reconciled later by the backfill)."""
    payload: dict[str, Any] = {
        "agent_id": _human_handle(username),
        "body": body,
        "type": "user",
    }
    if refs:
        payload["refs"] = refs[0]
    # A fresh client: app.state.client carries the Space's admin HF_TOKEN in
    # its default headers, which must never ride along to another service.
    async with httpx.AsyncClient(timeout=httpx.Timeout(HUB_FETCH_TIMEOUT)) as client:
        r = await client.post(
            f"{BACKEND_API_URL}/v1/messages",
            json=payload,
            headers={"Authorization": f"Bearer {user_token}"},
        )
    if r.status_code == 429:
        detail = ""
        try:
            detail = r.json()["detail"]["error"]["message"]
        except Exception:
            pass
        raise _ApiPostRejected(429, detail or "Rate limited — please slow down.")
    if r.status_code != 201:
        raise RuntimeError(f"bucket-sync API returned {r.status_code}: {r.text[:200]}")
    return r.json()


def _write_message_local(filename: str, content: str) -> None:
    msg_dir = Path(LOCAL_BUCKET_DIR) / PREFIX
    msg_dir.mkdir(parents=True, exist_ok=True)
    (msg_dir / filename).write_text(content, encoding="utf-8")


def _write_message_hub(filename: str, content: str, token: str | None = None) -> None:
    try:
        from huggingface_hub import batch_bucket_files
    except ImportError as e:
        raise RuntimeError("Install huggingface_hub to enable bucket writes.") from e

    # Prefer the Space's HF_TOKEN for the central-bucket write: org members
    # can only write to buckets they create, so a member's OAuth token cannot
    # write to the central bucket — only a privileged Space token can. Fall
    # back to the user's OAuth token if no HF_TOKEN is configured (a setup
    # where members *can* write). The displayed author is unaffected either
    # way: it comes from the `agent: human:{username}` frontmatter set from
    # the OAuth session.
    use_token = HF_TOKEN or token
    if not use_token:
        raise RuntimeError("No token available for writing to the bucket.")

    batch_bucket_files(
        BUCKET,
        add=[(content.encode("utf-8"), f"{PREFIX}/{filename}")],
        token=use_token,
    )


@app.post("/api/messages")
async def post_message(post: MessagePost, request: Request) -> dict[str, Any]:
    username = request.session.get("user")
    if not username:
        raise HTTPException(401, "Not logged in. Sign in with Hugging Face to post.")
    user_token = request.session.get("access_token")
    handle, body, refs = _normalize_human_post(post, username)
    delivered: list[str] = []
    if LOCAL_BUCKET_DIR:
        filename, content = _format_user_message(handle, body, refs)
        try:
            _write_message_local(filename, content)
        except OSError as e:
            log.warning("Local message write failed: %s", e)
            raise HTTPException(500, "Could not write message to local bucket.") from e
    else:
        if not (user_token or HF_TOKEN):
            raise HTTPException(401, "Server is not configured: set HF_TOKEN.")
        # Preferred path: the bucket-sync API, which fans @mentions and
        # quote-refs out to inbox/{recipient}/ — a direct bucket write never
        # reaches the inboxes agents poll.
        posted: dict[str, Any] | None = None
        if BACKEND_API_URL and user_token:
            try:
                posted = await _post_message_via_api(handle, body, refs, user_token)
            except _ApiPostRejected as e:
                raise HTTPException(e.status, e.detail)
            except Exception as e:
                log.warning(
                    "bucket-sync API post failed (%s); falling back to direct write.", e
                )
        if posted is not None:
            filename = posted["filename"]
            delivered = posted.get("mentions_delivered") or []
            content = _echo_user_message(handle, body, refs)
        else:
            # Fallback: the direct write. Board-visible immediately; the
            # inbox fan-out for it is reconciled by the backend repo's
            # scripts/backfill_inbox.py.
            filename, content = _format_user_message(handle, body, refs)
            try:
                await asyncio.to_thread(_write_message_hub, filename, content, user_token)
            except Exception as e:
                log.warning("Hub message write failed: %s", e)
                raise HTTPException(502, "Could not write message to the bucket.") from e
    # Bust the cache so other users see this message on their next poll
    # rather than waiting for the TTL.
    _invalidate_list_cache(PREFIX)
    return {
        "item": {"filename": filename, "content": content},
        "mentions_delivered": delivered,
    }


# ──────────────────────────────────────────────────────────────
# /api/verification  (results/verification_status.json)
#
# Small JSON map of result-filename → "valid" | "invalid" | "pending".
# A missing file means "nothing verified yet", which we report as {} so
# the frontend can default every result to "pending".
# ──────────────────────────────────────────────────────────────
async def _fetch_verification_hub() -> str:
    client: httpx.AsyncClient = app.state.client
    rel = f"{RESULTS_PREFIX}/verification_status.json"
    r = await client.get(f"{HUB}/buckets/{BUCKET}/resolve/{rel}")
    if r.status_code == 404:
        return "{}"
    if r.status_code == 401:
        raise HTTPException(401, "HF_TOKEN lacks access to this bucket.")
    if not r.is_success:
        raise HTTPException(r.status_code, f"Hub returned {r.status_code}")
    return r.text


@app.get("/api/verification")
async def verification() -> Response:
    rel = f"{RESULTS_PREFIX}/verification_status.json"
    if LOCAL_BUCKET_DIR:
        path = Path(LOCAL_BUCKET_DIR) / rel
        if path.is_file():
            return Response(
                content=path.read_text(encoding="utf-8"),
                media_type="application/json",
            )
        # Fall back to bucket when local file is absent.
        if not HF_TOKEN:
            return Response(content="{}", media_type="application/json")
    if not HF_TOKEN:
        raise HTTPException(401, "Server is not configured: set HF_TOKEN.")
    text = await _hub_cache.get("__verification__", _fetch_verification_hub)
    return Response(content=text, media_type="application/json")


# ──────────────────────────────────────────────────────────────
# /api/curation — everything the Final Set tab needs, in one call.
#
# Two sources: the curation dataset for the candidate pool (frozen, so cached
# hard) and the bucket-sync API for live PRs and final-set/rejected entries. Kept as one
# endpoint so the tab renders from a single consistent snapshot rather than four
# requests that can disagree with each other.
# ──────────────────────────────────────────────────────────────
def _doi_slug(doi: str) -> str:
    """DOI -> entry filename stem. Must match sanitize_doi_slug() in the
    bucket-sync backend and the open_pr client, or pool/accepted keys won't
    line up: `10.1038/s41586-025-09486-x` -> `10.1038-s41586-025-09486-x`."""
    cleaned = (doi or "").replace(":", "-").replace("/", "-")
    return re.sub(r"[^a-zA-Z0-9._-]", "-", cleaned).strip("-")


_candidates_cache = _SingleFlightCache(CANDIDATES_CACHE_TTL)
_curation_cache = _SingleFlightCache(CURATION_CACHE_TTL)


async def _fetch_candidates() -> dict[str, Any]:
    """Per-hypothesis candidate counts plus the ripest unclaimed papers.

    "Ripest" = most quotes, then most agents who independently found it — the
    pool sorted so the most obviously worth-judging papers are visible, which is
    the whole point of showing a backlog rather than a number.
    """
    client: httpx.AsyncClient = app.state.client
    counts: dict[str, int] = {}
    ripe: dict[str, list[dict[str, Any]]] = {}
    for hyp in CURATION_HYPS:
        url = f"{HUB}/datasets/{CURATION_DATASET}/resolve/main/candidates/{hyp}.json"
        try:
            r = await client.get(url)
            if not r.is_success:
                counts[hyp] = 0
                ripe[hyp] = []
                continue
            papers = (r.json() or {}).get("papers") or []
        except (httpx.HTTPError, ValueError) as e:
            log.warning("candidates/%s.json unreadable: %s", hyp, e)
            counts[hyp] = 0
            ripe[hyp] = []
            continue
        counts[hyp] = len(papers)
        ranked = sorted(
            papers,
            key=lambda p: (
                -(p.get("n_quotes") or len(p.get("quotes") or [])),
                -len(p.get("contributing_agents") or []),
            ),
        )
        ripe[hyp] = [
            {
                "hypothesis": hyp,
                "doi": p.get("doi"),
                "n_quotes": p.get("n_quotes") or len(p.get("quotes") or []),
                "agents": list(p.get("contributing_agents") or []),
            }
            for p in ranked[:8]
        ]
    return {"counts": counts, "total": sum(counts.values()), "ripest": ripe}


# The curation snapshot fires a dozen bucket-sync calls at once (see
# _fetch_curation), each of which does several Hub API calls of its own to
# render one PR or entry — that burst is enough to trip a 429 on the backend
# even though a single request succeeds fine. Cap how many of our own calls
# are in flight against bucket-sync at a time, and retry a transient 429
# once the burst has had a moment to clear.
_BACKEND_CONCURRENCY = asyncio.Semaphore(3)
_BACKEND_RETRY_STATUSES = {429}


async def _api_get(path: str, client: httpx.AsyncClient | None = None) -> Any:
    """GET a bucket-sync endpoint. Never app.state.client — that one carries the
    Space's HF_TOKEN, which must never ride along to another service. Pass a
    shared tokenless client when firing several of these at once (one pooled
    connection instead of a fresh TLS handshake per call); a one-off caller
    gets an ephemeral client of its own."""
    async def _get(c: httpx.AsyncClient) -> httpx.Response:
        async with _BACKEND_CONCURRENCY:
            return await c.get(f"{BACKEND_API_URL}{path}")

    async def _fetch(c: httpx.AsyncClient) -> httpx.Response:
        r = await _get(c)
        if r.status_code in _BACKEND_RETRY_STATUSES:
            retry_after = float(r.headers.get("retry-after") or 1.0)
            await asyncio.sleep(min(retry_after, 5.0))
            r = await _get(c)
        return r

    if client is not None:
        r = await _fetch(client)
    else:
        async with httpx.AsyncClient(timeout=httpx.Timeout(HUB_FETCH_TIMEOUT)) as c:
            r = await _fetch(c)
    if r.status_code == 404:
        return None          # curation disabled on the backend
    if not r.is_success:
        raise HTTPException(r.status_code, f"bucket-sync returned {r.status_code} for {path}")
    return r.json()


async def _fetch_curation() -> dict[str, Any]:
    # Final-set entries (primary/secondary) and rejected entries (unrelated)
    # are two disjoint views over the same tag; fetched separately because
    # they're two different endpoints, not two different mechanisms. One
    # shared client for all of them: a dozen brand-new TLS handshakes fired
    # at once to the same host is what tripped a connection cap here before.
    debug_errors: list[str] = []
    async with httpx.AsyncClient(timeout=httpx.Timeout(HUB_FETCH_TIMEOUT)) as bs_client:
        candidates, prs, merges, *rest = await asyncio.gather(
            _candidates_cache.get("__candidates__", _fetch_candidates),
            _api_get("/v1/prs?status=open", bs_client),
            _api_get("/v1/merges", bs_client),
            *[_api_get(f"/v1/final-set/{h}", bs_client) for h in CURATION_HYPS],
            *[_api_get(f"/v1/rejected/{h}", bs_client) for h in CURATION_HYPS],
            return_exceptions=True,
        )
    final_sets, rejecteds = rest[:len(CURATION_HYPS)], rest[len(CURATION_HYPS):]

    def ok(v, default):
        if isinstance(v, BaseException):
            log.warning("curation fetch failed: %s", v)
            debug_errors.append(f"{type(v).__name__}: {v}")
            return default
        return v if v is not None else default

    candidates = ok(candidates, {"counts": {}, "total": 0, "ripest": {}})
    prs = ok(prs, {"items": []})
    merges = ok(merges, {"items": []})

    entries: list[dict[str, Any]] = [
        e for fs in final_sets for e in (ok(fs, {"items": []}).get("items") or [])
    ]
    rejected: list[dict[str, Any]] = [
        e for rj in rejecteds for e in (ok(rj, {"items": []}).get("items") or [])
    ]

    # When did each land? Merge records are the only timestamped source, keyed
    # on "{HYP}/{slug}" for the final set and separately for rejected entries —
    # the same tag decides both which list an entry is in and which merge-record
    # field names it.
    merged_at: dict[str, str] = {}
    rejected_at: dict[str, str] = {}
    for m in merges.get("items") or []:
        for key in m.get("included") or []:
            merged_at[key] = m.get("timestamp") or ""
        for key in m.get("rejected") or []:
            rejected_at[key] = m.get("timestamp") or ""
    for e in entries:
        e["merged_at"] = merged_at.get(f"{e.get('hypothesis')}/{e.get('slug')}", "")
    for e in rejected:
        e["merged_at"] = rejected_at.get(f"{e.get('hypothesis')}/{e.get('slug')}", "")

    by_hyp: dict[str, int] = {h: 0 for h in CURATION_HYPS}
    for e in entries:
        h = e.get("hypothesis")
        if h in by_hyp:
            by_hyp[h] += 1

    open_prs = prs.get("items") or []
    contested = [p for p in open_prs if p.get("request_changes_by")]
    in_review = [p for p in open_prs if not p.get("request_changes_by")]

    # Drop already-judged papers (final set OR rejected) from the backlog so
    # the pool only ever shows what is genuinely still untouched.
    judged_keys = {f"{e.get('hypothesis')}/{e.get('slug')}" for e in (*entries, *rejected)}
    ripest: list[dict[str, Any]] = []
    for hyp, rows in (candidates.get("ripest") or {}).items():
        for row in rows:
            slug = _doi_slug(row.get("doi") or "")
            if f"{hyp}/{slug}" in judged_keys:
                continue
            ripest.append(row)
            break

    total = candidates.get("total") or 0
    return {
        "enabled": True,
        "dataset": CURATION_DATASET,
        "dataset_url": f"{HUB}/datasets/{CURATION_DATASET}",
        "candidates": candidates.get("counts") or {},
        "candidates_total": total,
        "accepted_by_hyp": by_hyp,
        "ripest": sorted(ripest, key=lambda r: -(r.get("n_quotes") or 0)),
        "entries": entries,
        "rejected": rejected,
        "in_review": in_review,
        "contested": contested,
        "pool": max(0, total - len(entries) - len(rejected) - len(open_prs)),
        # TEMP diagnostic for a live bug (empty entries/in_review despite a
        # populated backend) — remove once the cause is confirmed fixed.
        "_debug_errors": debug_errors,
    }


@app.get("/api/curation")
async def curation() -> dict[str, Any]:
    if not (CURATION_DATASET and BACKEND_API_URL):
        return {"enabled": False, "reason": "set CURATION_DATASET and BACKEND_API_URL"}
    return await _curation_cache.get("__curation__", _fetch_curation)


# ──────────────────────────────────────────────────────────────
# /api/paper-validation  (results/paper_validation.json)
#
# Per-finding LLM validation verdicts keyed by
# "{hypothesis}::{agent}::{doi}::{finding_id}".
# Written by POST /api/validate; read here with a 60 s cache.
# ──────────────────────────────────────────────────────────────
async def _fetch_paper_validation_hub() -> dict[str, Any]:
    client: httpx.AsyncClient = app.state.client
    r = await client.get(f"{HUB}/buckets/{BUCKET}/resolve/{PAPER_VALIDATION_PATH}")
    if r.status_code == 404:
        return {"entries": {}}
    if not r.is_success:
        log.warning("paper-validation fetch returned %s", r.status_code)
        return {"entries": {}}
    try:
        return r.json()
    except Exception:
        return {"entries": {}}


@app.get("/api/paper-validation")
async def paper_validation_get() -> dict[str, Any]:
    now = time.monotonic()
    if _paper_val_cache["data"] is not None and (now - _paper_val_cache["at"]) < PAPER_VAL_CACHE_TTL:
        return _paper_val_cache["data"]
    if LOCAL_BUCKET_DIR:
        path = Path(LOCAL_BUCKET_DIR) / PAPER_VALIDATION_PATH
        data: dict[str, Any] = {"entries": {}}
        if path.is_file():
            try:
                data = json.loads(path.read_text(encoding="utf-8"))
            except Exception:
                pass
        elif HF_TOKEN:
            # Validation always writes to the bucket; fall back when local copy absent.
            data = await _fetch_paper_validation_hub()
    elif HF_TOKEN:
        data = await _fetch_paper_validation_hub()
    else:
        data = {"entries": {}}
    _paper_val_cache["data"] = data
    _paper_val_cache["at"] = now
    return data


_VERDICT_LINE_RE = re.compile(
    r"^\s*(?:item\s*)?(\d+)\s*[:.\)]\s*(VALID|INVALID|UNCERTAIN)\b[\s:\-–]*(.*)$",
    re.IGNORECASE,
)


async def _llm_validate_batch(
    hyp_text: str,
    items: list[dict[str, str]],
    model: str,
    hf_token: str,
) -> list[dict[str, str]]:
    """Rate a batch of findings (same hypothesis) in a single HF Inference call.

    Returns one {status, reason} per item, in order — cuts per-call overhead
    vs. one request per finding.
    """
    try:
        from huggingface_hub import InferenceClient
    except ImportError:
        return [{"status": "error", "reason": "huggingface_hub not installed"} for _ in items]

    item_blocks = "\n".join(
        f"Item {i}:\n"
        f"Paper DOI: {it['doi']}\n"
        f"Finding description: {it['desc']}\n"
        f"Supporting quote: {it['quote']}\n"
        for i, it in enumerate(items, start=1)
    )
    prompt = (
        "You are a scientific literature validator for the MecCog Alzheimer's research challenge.\n\n"
        f"Hypothesis: {hyp_text}\n\n"
        "For each numbered item below, decide whether its quote provides valid evidence "
        "supporting the hypothesis above.\n\n"
        f"{item_blocks}\n"
        "Reply with exactly one line per item, no extra commentary, in this exact format:\n"
        "N: VERDICT - one sentence reason\n"
        "where VERDICT is one of VALID, INVALID, or UNCERTAIN."
    )

    def _call() -> str:
        client = InferenceClient(api_key=hf_token)
        result = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=max(200, 60 * len(items)),
            temperature=0.1,
        )
        return result.choices[0].message.content.strip()

    try:
        content = await asyncio.to_thread(_call)
    except Exception as exc:
        log.warning("LLM batch validation call failed: %s", exc)
        return [{"status": "error", "reason": str(exc)[:200]} for _ in items]

    verdicts: dict[int, dict[str, str]] = {}
    for line in content.split("\n"):
        m = _VERDICT_LINE_RE.match(line)
        if not m:
            continue
        idx = int(m.group(1))
        reason = m.group(3).strip()
        verdicts[idx] = {"status": m.group(2).lower(), "reason": reason or line.strip()[:200]}

    return [
        verdicts.get(
            i, {"status": "uncertain", "reason": "Could not parse batched verdict; treated as uncertain."}
        )
        for i in range(1, len(items) + 1)
    ]


@app.post("/api/validate")
async def validate_papers(request: Request) -> dict[str, Any]:
    """Run LLM validation on all findings that have a quote but no verdict yet.

    Auth: OAuth session cookie OR Authorization: Bearer <VALIDATE_SECRET>.
    Results are stored in ``results/paper_validation.json`` in the bucket.
    """
    username = request.session.get("user")
    if not username:
        # Accept a shared admin token as fallback when OAuth is not configured.
        bearer = request.headers.get("Authorization", "")
        token = bearer.removeprefix("Bearer ").strip()
        if not (VALIDATE_SECRET and token and token == VALIDATE_SECRET):
            raise HTTPException(401, "Not authorised. Sign in via OAuth or provide the VALIDATE_SECRET token.")
    if not HF_TOKEN:
        raise HTTPException(503, "HF_TOKEN not configured on this Space.")

    data = await meccog_results()
    results: list[dict] = data.get("results", [])
    hypotheses: dict[str, str] = data.get("hypotheses", {})

    # Load existing validation verdicts
    existing = await paper_validation_get()
    entries: dict[str, Any] = dict(existing.get("entries", {}))

    now_str = datetime.now(timezone.utc).isoformat()
    skipped_no_quote = 0
    skipped_already = 0
    to_validate: list[dict[str, Any]] = []

    for r in results:
        hyp_id = r.get("hypothesis")
        agent = r.get("agent") or "unknown"
        if not hyp_id or not r.get("papers"):
            continue
        for p in (r["papers"].get("papers") or []):
            doi = (p.get("doi") or "").strip()
            if not doi or doi == "N/A":
                continue
            for f in (p.get("findings") or []):
                quote = (f.get("quote") or "").strip()
                if not quote or quote == "N/A":
                    skipped_no_quote += 1
                    continue
                key = f"{hyp_id}::{agent}::{doi}::{f.get('id', 'F?')}"
                if key in entries:
                    skipped_already += 1
                    continue
                # Multiple agents often surface the same paper+quote for a
                # hypothesis — group on that so it's validated once, not once
                # per agent, then the verdict is copied to every composite key.
                dedup_key = (hyp_id, doi.lower(), re.sub(r"\s+", " ", quote).strip().lower())
                to_validate.append({
                    "key": key,
                    "dedup_key": dedup_key,
                    "hyp_text": hypotheses.get(hyp_id, ""),
                    "doi": doi,
                    "desc": (f.get("desc") or "").strip(),
                    "quote": quote,
                })

    if not to_validate:
        return {
            "validated": 0, "valid": 0, "invalid": 0, "uncertain": 0, "errors": 0,
            "skipped_no_quote": skipped_no_quote,
            "skipped_already_done": skipped_already,
            "total_in_file": len(entries),
        }

    groups: dict[tuple, list[dict[str, Any]]] = {}
    for item in to_validate:
        groups.setdefault(item["dedup_key"], []).append(item)
    unique_items = [group[0] for group in groups.values()]

    # Batch unique items (same hypothesis text) into fewer, larger LLM calls.
    by_hyp: dict[str, list[dict[str, Any]]] = {}
    for item in unique_items:
        by_hyp.setdefault(item["dedup_key"][0], []).append(item)
    batches: list[tuple[str, list[dict[str, Any]]]] = []
    for hyp_id, items in by_hyp.items():
        hyp_text = items[0]["hyp_text"]
        for i in range(0, len(items), VALIDATION_BATCH_SIZE):
            batches.append((hyp_text, items[i:i + VALIDATION_BATCH_SIZE]))

    sem = asyncio.Semaphore(3)  # max 3 concurrent HF Inference calls

    async def run_batch(hyp_text: str, items: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
        async with sem:
            verdicts = await _llm_validate_batch(hyp_text, items, VALIDATION_MODEL, HF_TOKEN)
            return items, verdicts

    batch_outcomes = await asyncio.gather(*(run_batch(h, i) for h, i in batches), return_exceptions=True)

    valid_count = invalid_count = uncertain_count = error_count = 0
    llm_calls = 0
    for outcome in batch_outcomes:
        if isinstance(outcome, Exception):
            continue
        llm_calls += 1
        items, verdicts = outcome
        for item, verdict in zip(items, verdicts):
            full_verdict = {**verdict, "validated_at": now_str}
            dup_keys = [dup["key"] for dup in groups[item["dedup_key"]]]
            for key in dup_keys:
                entries[key] = full_verdict
            s = verdict.get("status")
            n = len(dup_keys)
            if s == "valid":     valid_count += n
            elif s == "invalid": invalid_count += n
            elif s == "uncertain": uncertain_count += n
            else:                error_count += n

    # Persist to bucket
    validation_data: dict[str, Any] = {
        "model": VALIDATION_MODEL,
        "updated_at": now_str,
        "entries": entries,
    }
    body = json.dumps(validation_data, indent=2, sort_keys=True) + "\n"
    try:
        from huggingface_hub import batch_bucket_files
        await asyncio.to_thread(
            batch_bucket_files,
            BUCKET,
            add=[(body.encode("utf-8"), PAPER_VALIDATION_PATH)],
            token=HF_TOKEN,
        )
        _paper_val_cache["data"] = validation_data
        _paper_val_cache["at"] = time.monotonic()
        log.info("paper_validation.json updated: %d entries total", len(entries))
    except Exception as exc:
        log.warning("Failed to write paper_validation.json: %s", exc)
        raise HTTPException(502, f"Validation ran but could not save results: {exc}") from exc

    return {
        "validated": len(to_validate) - error_count,
        "valid": valid_count,
        "invalid": invalid_count,
        "uncertain": uncertain_count,
        "errors": error_count,
        "skipped_no_quote": skipped_no_quote,
        "skipped_already_done": skipped_already,
        "total_in_file": len(entries),
        # Cost-visibility: how much the dedup + batching actually saved.
        "unique_items_validated": len(unique_items),
        "llm_calls": llm_calls,
    }


# ──────────────────────────────────────────────────────────────
# /api/meccog  (Results tab: hypothesis coverage, consensus, xlsx submissions)
#
# Fetches expanded results + messages from the bucket-sync API (BACKEND_API_URL),
# downloads each submission's .xlsx from the HF bucket, and returns structured
# JSON for the frontend's Results tab. Cached 45 s to be polite to the APIs.
# ──────────────────────────────────────────────────────────────
@app.get("/api/meccog")
async def meccog_results() -> dict[str, Any]:
    now = time.monotonic()
    if _meccog_cache["data"] is not None and (now - _meccog_cache["at"]) < MECCOG_CACHE_TTL:
        return _meccog_cache["data"]

    if not BACKEND_API_URL:
        return {
            "results": [], "leaderboard": [], "coverage": {},
            "consensus": {}, "hyp_order": HYP_ORDER, "hypotheses": HYPOTHESES,
            "hyp_meta": HYP_META,
            "fetched_at": "", "error": "BACKEND_API_URL is not configured.",
        }

    hub_client: httpx.AsyncClient = app.state.client  # has HF_TOKEN for bucket reads

    # Fresh client for the bucket-sync API (no HF_TOKEN — it's a separate service).
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(HUB_FETCH_TIMEOUT), follow_redirects=True
    ) as api_client:
        gathered = await asyncio.gather(
            api_client.get(f"{BACKEND_API_URL}/v1/results", params={"expand": "true", "limit": 200}),
            api_client.get(f"{BACKEND_API_URL}/v1/leaderboard"),
            api_client.get(f"{BACKEND_API_URL}/v1/messages", params={"expand": "true", "order": "asc", "limit": 300}),
            api_client.get(f"{BACKEND_API_URL}/v1/agents", params={"expand": "true"}),
            return_exceptions=True,
        )

    results_resp, lb_resp, msgs_resp, agents_resp = gathered

    if isinstance(results_resp, Exception) or not results_resp.is_success:
        err = str(results_resp) if isinstance(results_resp, Exception) else results_resp.text[:200]
        log.warning("Backend API error on /v1/results: %s", err)
        # The bucket-sync API is a separate service and can be briefly down or
        # slow to wake — degrade to the last known-good payload rather than
        # taking the whole Results tab down with a 502, mirroring how
        # _SingleFlightCache serves stale data for /api/messages etc.
        if _meccog_cache["data"] is not None:
            stale = dict(_meccog_cache["data"])
            stale["error"] = f"Backend API error (showing cached data): {err}"
            return stale
        return {
            "results": [], "leaderboard": [], "coverage": {},
            "consensus": {}, "hyp_order": HYP_ORDER, "hypotheses": HYPOTHESES,
            "hyp_meta": HYP_META,
            "fetched_at": "", "error": f"Backend API error: {err}",
        }

    results_raw: list[dict] = results_resp.json().get("items", [])
    lb_rows: list[dict] = (
        lb_resp.json().get("rows", [])
        if not isinstance(lb_resp, Exception) and lb_resp.is_success else []
    )
    msgs_raw: list[dict] = (
        msgs_resp.json().get("items", [])
        if not isinstance(msgs_resp, Exception) and msgs_resp.is_success else []
    )
    agents_raw: list[dict] = (
        agents_resp.json().get("items", [])
        if not isinstance(agents_resp, Exception) and agents_resp.is_success else []
    )

    async def _enrich(r: dict[str, Any]) -> dict[str, Any]:
        fm = r.get("frontmatter") or {}
        hyp = (fm.get("hypothesis") or "").upper() or None
        if hyp and hyp not in HYPOTHESES:
            hyp = _HYP_TEXT_TO_ID.get(re.sub(r"\s+", " ", hyp).strip(), hyp)
        filename = r.get("filename")
        uri = fm.get("spreadsheet")
        entry: dict[str, Any] = {
            "filename": filename,
            "agent": fm.get("agent"),
            "hypothesis": hyp,
            "method": fm.get("method"),
            "description": fm.get("description"),
            "timestamp": fm.get("timestamp"),
            "verification": r.get("verification"),
            "spreadsheet_url": _spreadsheet_url(uri) if uri else None,
            "result_md_url": f"{BACKEND_API_URL}/v1/results/{filename}" if filename else None,
            "body": r.get("body") or "",
            "papers": None,
            "parse_error": None,
        }
        if uri:
            xlsx_url = _spreadsheet_url(uri)
            if xlsx_url:
                try:
                    async with _fetch_sem:
                        resp = await hub_client.get(xlsx_url)
                    if resp.is_success:
                        parsed = await asyncio.to_thread(_parse_submission_xlsx, resp.content)
                        if parsed is not None:
                            entry["papers"] = parsed
                        else:
                            entry["parse_error"] = "xlsx parse failed (openpyxl missing or unsupported format)"
                    else:
                        entry["parse_error"] = f"HTTP {resp.status_code}"
                except Exception as exc:
                    entry["parse_error"] = str(exc)[:120]
        return entry

    results: list[dict] = list(await asyncio.gather(*(_enrich(r) for r in results_raw)))

    # Novelty stats: walk each hypothesis's submissions in chronological
    # order and split every submission's papers into "new" (first submission
    # to surface that DOI) vs "repeat" (an earlier submission already found
    # it) — the raw signal behind the New-papers / Consensus charts and the
    # Replay on the Leaderboard tab. Each paper also gets an `is_new` flag so
    # the Replay can tag individual sources, not just per-submission totals.
    # Timestamps are "YYYY-MM-DD HH:MM UTC", which sorts correctly as a plain
    # string, so no datetime parsing is needed.
    seen_dois_by_hyp: dict[str, set[str]] = {}
    for r in sorted(
        (r for r in results if r.get("hypothesis")),
        key=lambda r: r.get("timestamp") or "",
    ):
        seen = seen_dois_by_hyp.setdefault(r["hypothesis"], set())
        new_count = 0
        repeat_count = 0
        for p in (r.get("papers") or {}).get("papers", []):
            doi = (p.get("doi") or "").lower().strip()
            if not doi or doi == "n/a":
                p["is_new"] = None  # no DOI to dedupe on — can't classify
                continue
            if doi in seen:
                repeat_count += 1
                p["is_new"] = False
            else:
                new_count += 1
                seen.add(doi)
                p["is_new"] = True
        r["papers_new"] = new_count
        r["papers_repeat"] = repeat_count

    # Hypothesis coverage: submissions + message-level claims.
    coverage: dict[str, dict] = {h: {"submissions": [], "claims": []} for h in HYP_ORDER}
    for r in results:
        h = r.get("hypothesis") or ""
        if h in coverage:
            coverage[h]["submissions"].append(r.get("agent") or "unknown")
    for m in msgs_raw:
        body = m.get("body") or ""
        fm2 = m.get("frontmatter") or {}
        for match in _CLAIM_RE.finditer(body):
            hid = match.group(1).upper()
            if hid in coverage:
                coverage[hid]["claims"].append({"agent": fm2.get("agent"), "timestamp": fm2.get("timestamp")})

    # Consensus: DOI overlap across submissions on the same hypothesis.
    consensus: dict[str, dict] = {}
    by_hyp: dict[str, list[dict]] = {}
    for r in results:
        if r.get("papers") and r.get("hypothesis"):
            by_hyp.setdefault(r["hypothesis"], []).append(r)
    for h, entries in by_hyp.items():
        doi_map: dict[str, dict] = {}
        for e in entries:
            for p in (e["papers"] or {}).get("papers", []):
                doi = (p.get("doi") or "").lower().strip()
                if not doi or doi == "n/a":
                    continue
                if doi not in doi_map:
                    doi_map[doi] = {"doi": p["doi"], "agents": set()}
                doi_map[doi]["agents"].add(e.get("agent") or "unknown")
        consensus[h] = {
            "shared": [{"doi": v["doi"], "agents": sorted(v["agents"])} for v in doi_map.values() if len(v["agents"]) > 1],
            "unique": [{"doi": v["doi"], "agent": next(iter(v["agents"]))} for v in doi_map.values() if len(v["agents"]) == 1],
            "n_agents": len(entries),
        }

    # Agents: normalise each agent's frontmatter into a flat dict for the roster.
    agents_list: list[dict] = []
    for a in agents_raw:
        agents_list.append({
            "aid": a.get("agent_id") or a.get("filename", "").removesuffix(".md"),
            "model": a.get("model"),
            "harness": a.get("harness"),
            "joined": a.get("joined"),
            "tools": a.get("tools") or [],
            "description": a.get("bio"),
        })

    # Last 60 messages for the activity log (msgs_raw is already asc-sorted).
    log_messages: list[dict] = []
    for m in msgs_raw[-60:]:
        fm4 = m.get("frontmatter") or {}
        log_messages.append({
            "agent": fm4.get("agent"),
            "timestamp": fm4.get("timestamp"),
            "body": m.get("body") or "",
            "type": fm4.get("type", "agent"),
        })

    data: dict[str, Any] = {
        "results": results,
        "leaderboard": lb_rows,
        "coverage": coverage,
        "consensus": consensus,
        "agents": agents_list,
        "messages": log_messages,
        "backend_api_url": BACKEND_API_URL,
        "hyp_order": HYP_ORDER,
        "hypotheses": HYPOTHESES,
        "hyp_meta": HYP_META,
        "fetched_at": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"),
    }
    _meccog_cache["data"] = data
    _meccog_cache["at"] = now
    return data


# ──────────────────────────────────────────────────────────────
# /api/meccog/aggregate.xlsx  — one Excel aggregating every submission,
# deduplicated by DOI per hypothesis, with an Agent(s) column added.
# Column positions match the standard submission format so each hypothesis
# section could be extracted as a standalone valid submission.
# ──────────────────────────────────────────────────────────────
@app.get("/api/meccog/aggregate.xlsx")
async def meccog_aggregate_xlsx() -> Response:
    try:
        from openpyxl import Workbook
        from openpyxl.styles import Font, PatternFill, Alignment
        from openpyxl.utils import get_column_letter
    except ImportError:
        raise HTTPException(status_code=501, detail="openpyxl not installed")

    data = await meccog_results()
    results: list[dict] = data.get("results", [])
    hyp_order: list[str] = data.get("hyp_order", [])
    hypotheses: dict[str, str] = data.get("hypotheses", {})

    # Build per-hypothesis map: doi → {doi, source_type, pmid, by_agent: {agent: paper}}
    hyp_papers: dict[str, dict[str, dict]] = {}
    for r in results:
        hyp = r.get("hypothesis")
        if not hyp or not r.get("papers"):
            continue
        agent = r.get("agent") or "unknown"
        for p in (r["papers"].get("papers") or []):
            doi = (p.get("doi") or "").strip()
            if not doi or doi == "N/A":
                continue
            hyp_papers.setdefault(hyp, {})
            if doi not in hyp_papers[hyp]:
                hyp_papers[hyp][doi] = {
                    "doi": doi,
                    "source_type": p.get("source_type", "N/A"),
                    "pmid": p.get("pmid", "N/A"),
                    "by_agent": {},
                }
            hyp_papers[hyp][doi]["by_agent"][agent] = p

    wb = Workbook()
    ws = wb.active
    ws.title = "Aggregated Evidence"

    fill_header  = PatternFill("solid", fgColor="0F3787")
    fill_section = PatternFill("solid", fgColor="DDE6F5")
    fill_paper   = PatternFill("solid", fgColor="F4F4F4")
    font_white   = Font(bold=True, color="FFFFFF")
    font_bold    = Font(bold=True)

    COLS = [
        "Hypothesis", "DOI", "Source Type", "PMID", "ID",
        "Description", "Quote", "Summary", "Relevance",
        "System", "Location", "Effect Size", "P-value", "N",
        "Agent(s)",
    ]
    WIDTHS = [14, 36, 14, 14, 10, 40, 40, 40, 12, 16, 16, 20, 10, 8, 25]
    N_COLS = len(COLS)

    ws.append(COLS)
    for cell in ws[1]:
        cell.font = font_white
        cell.fill = fill_header
        cell.alignment = Alignment(horizontal="center")
    ws.freeze_panes = "A2"

    def _v(val: Any) -> str:
        s = str(val).strip() if val is not None else ""
        return "" if s in ("", "N/A") else s

    for hid in hyp_order:
        paper_map = hyp_papers.get(hid, {})
        papers = sorted(paper_map.values(), key=lambda p: -len(p["by_agent"]))
        if not papers:
            continue

        hyp_text = hypotheses.get(hid, "")
        ws.append([f"{hid} — {hyp_text}"] + [""] * (N_COLS - 1))
        sec_row = ws.max_row
        for cell in ws[sec_row]:
            cell.font = font_bold
            cell.fill = fill_section
        ws.merge_cells(f"A{sec_row}:{get_column_letter(N_COLS)}{sec_row}")

        for p_idx, paper in enumerate(papers, 1):
            paper_id = f"P{p_idx:02d}"
            agents = sorted(paper["by_agent"])
            ref = paper["by_agent"][agents[0]]

            ws.append([
                hid, paper["doi"],
                _v(ref.get("source_type")), _v(ref.get("pmid")),
                paper_id,
                "", "", "", "", "", "", "", "", "",
                " / ".join(agents),
            ])
            for cell in ws[ws.max_row]:
                cell.fill = fill_paper

            f_idx = 1
            for agent in agents:
                for f in (paper["by_agent"][agent].get("findings") or []):
                    ws.append([
                        "", "", "", "",
                        f"{paper_id}.F{f_idx:02d}",
                        _v(f.get("desc")), _v(f.get("quote")), _v(f.get("summary")),
                        _v(f.get("relevance")), _v(f.get("system")), _v(f.get("location")),
                        _v(f.get("effect")), _v(f.get("pvalue")), _v(f.get("n")),
                        agent,
                    ])
                    f_idx += 1

    for i, w in enumerate(WIDTHS, 1):
        ws.column_dimensions[get_column_letter(i)].width = w

    buf = io.BytesIO()
    wb.save(buf)
    buf.seek(0)
    return Response(
        content=buf.read(),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": 'attachment; filename="meccog_aggregated_evidence.xlsx"'},
    )


# ──────────────────────────────────────────────────────────────
# Static frontend  (mounted last so /api/* keeps priority)
# ──────────────────────────────────────────────────────────────
_static_dir = Path(__file__).parent / "static"
app.mount("/", StaticFiles(directory=str(_static_dir), html=True), name="static")