app.py

"""HVTracker web service.

Serves the pre-generated static site from the volume, exposes a dynamic JSON
API and live SVG badges sourced from data.json, accepts agent submissions /
corrections into Postgres, and runs the 4-hourly refresh in-process.
"""
from __future__ import annotations

import json
import os
import re
import sys
import threading
import hashlib
import time
from collections import deque
from contextlib import asynccontextmanager
from datetime import datetime, timezone, timedelta
from html import escape

from apscheduler.schedulers.background import BackgroundScheduler
from dotenv import load_dotenv
from fastapi import FastAPI, Form, Request
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse, RedirectResponse, Response
from fastapi.staticfiles import StaticFiles

import alerts
import db
import mcp_server
import mcp_trust
import usage
import verify_log

# Load local secrets/config from a gitignored .env (OAuth client ids/secrets,
# HVT_SESSION_SECRET, HVT_DEV_AUTH). No-op in production where env vars are set
# directly. Must run before `import auth` (~line 206), which reads these at load.
load_dotenv()


# ---- anti-spam helpers -----------------------------------------------------

_RATE_WINDOW = 600  # 10 minutes
_RATE_LIMIT = 5
_rate_log: dict[str, deque] = {}
_rate_lock = threading.Lock()


def _client_ip(request: Request) -> str:
    return (
        request.headers.get("cf-connecting-ip")
        or request.headers.get("x-forwarded-for", "").split(",")[0].strip()
        or (request.client.host if request.client else "unknown")
    )


def _is_rate_limited(request: Request) -> bool:
    ip = _client_ip(request)
    now = time.monotonic()
    with _rate_lock:
        q = _rate_log.setdefault(ip, deque())
        while q and q[0] < now - _RATE_WINDOW:
            q.popleft()
        if len(q) >= _RATE_LIMIT:
            return True
        q.append(now)
    return False


# Open-lookup (P2): a more generous, read-only limit than the form limiter,
# plus a short cache so repeat checks of the same repo don't re-hit GitHub.
_OLOOKUP_WINDOW = 600
_OLOOKUP_LIMIT = 20
_olookup_log: dict[str, deque] = {}
_olookup_cache: dict[str, tuple] = {}
_OLOOKUP_TTL = 86400  # cache a repo's verdict for the day; a daily job can refresh it


def _is_open_lookup_limited(request: Request) -> bool:
    ip = _client_ip(request)
    now = time.monotonic()
    with _rate_lock:
        q = _olookup_log.setdefault(ip, deque())
        while q and q[0] < now - _OLOOKUP_WINDOW:
            q.popleft()
        if len(q) >= _OLOOKUP_LIMIT:
            return True
        q.append(now)
    return False


def _olookup_cache_get(repo: str):
    hit = _olookup_cache.get(repo)
    if hit and time.monotonic() - hit[0] < _OLOOKUP_TTL:
        return hit[1]
    return None


def _olookup_cache_put(repo: str, verdict: dict) -> None:
    _olookup_cache[repo] = (time.monotonic(), verdict)


# MCP endpoint limiter: the trust tools are read-only in-memory lookups with no
# external calls, so this is deliberately generous — it only exists to stop a
# flood that bypasses the CDN and hits the origin directly. A real multi-check
# agent session stays well under it.
_MCP_WINDOW = 60
_MCP_LIMIT = 60
_mcp_log: dict[str, deque] = {}


def _is_mcp_rate_limited(request: Request) -> bool:
    ip = _client_ip(request)
    now = time.monotonic()
    with _rate_lock:
        q = _mcp_log.setdefault(ip, deque())
        while q and q[0] < now - _MCP_WINDOW:
            q.popleft()
        if len(q) >= _MCP_LIMIT:
            return True
        q.append(now)
    return False


def _mcp_enabled() -> bool:
    """Budget kill switch for the MCP endpoint.

    Set MCP_ENABLED=0 (Railway variable) to pause only the /mcp server — the
    website, badges, and the /api/v1/mcp/verify HTTP API keep serving. Use this
    if MCP traffic ever pushes usage toward the workspace spend cap.
    """
    return os.environ.get("MCP_ENABLED", "1").strip().lower() not in {"0", "false", "off", "no"}


_EMAIL_RE = re.compile(r".+@.+\..+")

_FIELD_LIMITS = {
    "repo": 200,
    "name": 120,
    "email": 254,
    "contact": 254,
    "message": 4000,
    "notes": 4000,
}
_DEFAULT_LIMIT = 500


def _check_field_lengths(**fields: str) -> str | None:
    for name, value in fields.items():
        limit = _FIELD_LIMITS.get(name, _DEFAULT_LIMIT)
        if len(value) > limit:
            return f"{name} exceeds the {limit}-character limit."
    return None


HONEYPOT_HTML = '<div style="position:absolute;left:-9999px;top:-9999px" aria-hidden="true"><label>Leave blank<input name="website" autocomplete="off" tabindex="-1"></label></div>'

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
OUTPUT_DIR = os.environ.get("OUTPUT_DIR", BASE_DIR)


def _asset_ver(name: str) -> str:
    """Short content hash for cache-busting unhashed static JS (auth.js etc.)."""
    try:
        with open(os.path.join(BASE_DIR, name), "rb") as f:
            return hashlib.sha256(f.read()).hexdigest()[:8]
    except OSError:
        return "0"


_AUTH_JS_VER = _asset_ver("auth.js")
os.makedirs(OUTPUT_DIR, exist_ok=True)  # volume subdir may not exist on first boot
DATA_PATH = os.path.join(OUTPUT_DIR, "data.json")
TEMPLATES_DIR = os.path.join(BASE_DIR, "templates")
COMPARE_TOOL_PATH = os.path.join(BASE_DIR, "compare", "index.html")
VERIFY_TOOL_PATH = os.path.join(BASE_DIR, "verify", "index.html")
LIVE_PAGE_PATH = os.path.join(BASE_DIR, "live", "index.html")
SCAN_TOOL_PATH = os.path.join(BASE_DIR, "scan", "index.html")
RENDER_FINGERPRINT_PATH = os.path.join(OUTPUT_DIR, ".render_fingerprint")
verify_log.init(OUTPUT_DIR)  # public "recently checked" feed (transparency)
usage.init(OUTPUT_DIR)  # machine-channel usage rollup behind /live/
REFRESH_STATUS_PATH = os.path.join(OUTPUT_DIR, ".refresh_status.json")
MAX_DATA_AGE = timedelta(hours=int(os.environ.get("MAX_DATA_AGE_HOURS", "6")))
# OSSF scores are baked into the image from the `data` branch at build time
# (see Dockerfile). Re-pull that file at runtime so live refreshes apply the
# latest daily scan instead of the deploy-time snapshot, which otherwise ages
# out to the unreliable deps.dev fallback after 48h.
SCORECARD_CACHE_URL = os.environ.get(
    "SCORECARD_CACHE_URL",
    "https://raw.githubusercontent.com/YugantM/hvtracker/data/scorecard-cache.json",
)
SCORECARD_CACHE_PATH = os.path.join(BASE_DIR, "scorecard-cache.json")
# The /registry/ feed's snapshot of official MCP registry entries, refreshed
# daily on the `data` branch by .github/workflows/registry-snapshot.yml.
REGISTRY_SNAPSHOT_URL = os.environ.get(
    "REGISTRY_SNAPSHOT_URL",
    "https://raw.githubusercontent.com/YugantM/hvtracker/data/mcp-registry-snapshot.json",
)
REGISTRY_SNAPSHOT_PATH = os.path.join(BASE_DIR, "mcp-registry-snapshot.json")

@asynccontextmanager
async def _lifespan(_app):
    # Run the MCP Streamable-HTTP session manager (required by the SDK even in
    # stateless mode) for the lifetime of the app, then the existing startup work.
    _install_mcp_route()
    async with mcp_server.mcp.session_manager.run():
        startup()
        try:
            yield
        finally:
            shutdown()


app = FastAPI(title="HVTracker", docs_url="/api/docs", openapi_url="/api/openapi.json",
              lifespan=_lifespan)

# Accounts: GitHub/Google OAuth + watchlist/claim/notifications. Registered
# before the catch-all StaticFiles mount so its /auth/* and /api/* routes win.
import auth as _auth  # noqa: E402
app.include_router(_auth.router)

_scheduler = None
_scheduler_error: str | None = None
_refresh_lock = threading.Lock()


# ---- cache headers -------------------------------------------------------
#
# StaticFiles ships responses with no Cache-Control, so Cloudflare returns
# `cf-cache-status: DYNAMIC` and every visit hits Railway directly.  Add
# sensible defaults: HTML pages get 5 min browser / 15 min CDN cache (with
# stale-while-revalidate so re-fetches feel instant), JSON data endpoints
# get longer s-maxage since they're consumed by API users.  Routes that
# already set Cache-Control (e.g. badge SVGs at app.py:172) win — we only
# fill in the gaps.
_HTML_CACHE = "public, max-age=300, s-maxage=900, stale-while-revalidate=86400"
_JSON_CACHE = "public, max-age=600, s-maxage=1800, stale-while-revalidate=86400"
# Static images/fonts (OG share cards, logos, favicons). Without a header these
# ship `cf-cache-status: BYPASS` and every social unfurl / crawler fetch of a
# ~50-60KB card hits Railway. Per-agent OG cards regenerate when a score moves,
# so a 1-day edge TTL (stale-while-revalidate serves instantly, revalidates in
# the background) keeps them fresh enough while cutting origin egress to ~zero.
_ASSET_CACHE = "public, max-age=3600, s-maxage=86400, stale-while-revalidate=604800"


_CSP = (
    "default-src 'self'; "
    "style-src 'self' 'unsafe-inline'; "
    "font-src 'self'; "
    "script-src 'self' 'unsafe-inline'; "
    "img-src 'self' data:"
)
_CANONICAL_HOST = "hvtracker.net"
_LOCAL_HOSTS = {"localhost", "127.0.0.1", "::1"}
_DYNAMIC_SLASH_PATHS = {
    "/alerts",
    "/badges",
    "/blog",
    "/changes",
    "/changelog",
    "/compare",
    "/verify",
    "/correct",
    "/data",
    "/data-api",
    "/ecosystem",
    "/movers",
    "/methodology",
    "/org",
    "/roadmap",
    "/spec",
    "/sponsor",
    "/submit",
    "/use-cases",
}
_HEALTHCHECK_PATHS = {"/healthz", "/healthz/"}

# Retired sections that still attract crawlers (GSC "Not found (404)" cleanup).
# 301 preserves link equity and fires before the static mount, so stale
# generated pages left on the prod volume can never shadow these.
_RETIRED_REDIRECTS = {
    "/score-lab/": "/methodology/#runtime-calibration",
    "/spec/runtime-trust/v0.1/": "/spec/runtime-trust/v0.3/",
    "/spec/runtime-trust/v0.2/": "/spec/runtime-trust/v0.3/",
    "/org/i-am-bee/": "/org/",
    "/use-cases/recently-active/": "/use-cases/",
    # Duplicate listings merged into the original via previous_repos.
    "/agents/headroomlabs-headroom/": "/agents/headroom/",
    "/agents/zylon-ai-privategpt/": "/agents/privategpt/",
    "/agents/notfair-plugin/": "/agents/notfair/",
}
# Agents hard-deleted from agents.json (never in data/retired.json, which only
# carries `legacy` rows the renderer delisted).
_HARD_RETIRED_AGENT_SLUGS = frozenset({"bee-agent-framework"})
_retired_cache = {"mtime": None, "slugs": _HARD_RETIRED_AGENT_SLUGS}


def _retired_slugs() -> frozenset:
    """Slugs whose /agents/<slug>/ page is gone for good → 410 (cached by mtime)."""
    path = os.path.join(OUTPUT_DIR, "data", "retired.json")
    try:
        mtime = os.path.getmtime(path)
    except OSError:
        return _HARD_RETIRED_AGENT_SLUGS
    if _retired_cache["mtime"] != mtime:
        try:
            with open(path, encoding="utf-8") as f:
                slugs = json.load(f).get("agents", [])
        except (OSError, json.JSONDecodeError, AttributeError):
            return _HARD_RETIRED_AGENT_SLUGS
        _retired_cache["slugs"] = frozenset(slugs) | _HARD_RETIRED_AGENT_SLUGS
        _retired_cache["mtime"] = mtime
    return _retired_cache["slugs"]


def _retired_response(path: str) -> Response | None:
    """301 for retired sections, 410 Gone for retired agent/compare pages."""
    lookup = path if path.endswith("/") else f"{path}/"
    target = _RETIRED_REDIRECTS.get(lookup)
    if target is not None:
        return RedirectResponse(target, status_code=301)
    parts = [p for p in lookup.split("/") if p]
    if len(parts) != 2:
        return None
    gone = False
    if parts[0] == "agents":
        gone = parts[1] in _retired_slugs()
    elif parts[0] == "compare" and "-vs-" in parts[1]:
        gone = any(s in _retired_slugs() for s in parts[1].split("-vs-"))
    if gone:
        return HTMLResponse(
            "<h1>410 Gone</h1><p>This agent has been retired from the HVTracker "
            'registry. See the <a href="/">current leaderboard</a>.</p>',
            status_code=410,
        )
    return None


def _external_scheme(request: Request) -> str:
    cf_visitor = request.headers.get("cf-visitor")
    if cf_visitor:
        try:
            scheme = json.loads(cf_visitor).get("scheme")
            if scheme in {"http", "https"}:
                return scheme
        except (json.JSONDecodeError, TypeError, AttributeError):
            pass
    forwarded = request.headers.get("x-forwarded-proto", "")
    if forwarded:
        scheme = forwarded.split(",", 1)[0].strip().lower()
        if scheme in {"http", "https"}:
            return scheme
    return request.url.scheme


def _external_host(request: Request) -> str:
    forwarded_host = request.headers.get("x-forwarded-host")
    raw_host = forwarded_host.split(",", 1)[0].strip() if forwarded_host else request.headers.get("host", "")
    return raw_host.split(":", 1)[0].lower()


def _path_needs_trailing_slash(path: str) -> bool:
    if path in ("", "/") or path.endswith("/"):
        return False
    if path.startswith("/spec/"):
        parts = [part for part in path.split("/") if part]
        if len(parts) >= 3:
            return True
    if "." in path.rsplit("/", 1)[-1]:
        return False
    if path in _DYNAMIC_SLASH_PATHS or path.startswith("/track/"):
        return True
    candidate = os.path.join(OUTPUT_DIR, path.lstrip("/"))
    return os.path.isdir(candidate) and os.path.isfile(os.path.join(candidate, "index.html"))


def _canonical_redirect_target(request: Request) -> str | None:
    scheme = _external_scheme(request)
    host = _external_host(request)
    orig_path = request.url.path or "/"
    # Collapse repeated slashes: /agents/composio// currently 200s (StaticFiles
    # resolves the doubled path) and Google indexes it as a duplicate of the
    # single-slash canonical. Redirect to the collapsed path so each page has
    # exactly one crawlable URL. Query string is untouched. The "already
    # canonical" guards below compare the target against orig_path (not the
    # collapsed path), or a doubled-slash URL would match itself and never
    # redirect.
    path = re.sub(r"/{2,}", "/", orig_path) if "//" in orig_path else orig_path
    target_path = f"{path}/" if _path_needs_trailing_slash(path) else path
    if host in _LOCAL_HOSTS:
        if target_path == orig_path:
            return None
        query = request.url.query
        suffix = f"?{query}" if query else ""
        return f"{request.url.scheme}://{request.headers.get('host', host)}{target_path}{suffix}"
    if scheme == "https" and host == _CANONICAL_HOST and target_path == orig_path:
        return None
    query = request.url.query
    suffix = f"?{query}" if query else ""
    return f"https://{_CANONICAL_HOST}{target_path}{suffix}"


# ---- machine-surface usage counters (master plan 1.2) ---------------------
# In-memory since last process start — enough to establish the API/MCP usage
# baseline KPI without any storage. Exposed in /healthz as machine_usage.
_usage_counters = {"api_v1": 0, "mcp": 0, "data_json": 0, "agent_records": 0, "exports": 0, "registry": 0}
_USAGE_SINCE = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

# /mcp requests by protocol era: 2026-07-28 clients name the method in an
# Mcp-Method header; 2025-era clients send none ("legacy"). Shows how fast
# clients move to the new revision. Unknown header values fold into "other"
# so arbitrary input can't grow the dict.
_MCP_METHODS = frozenset({
    "server/discover", "tools/list", "tools/call", "resources/list",
    "resources/read", "prompts/list", "subscriptions/listen", "ping",
})
_mcp_method_counters: dict[str, int] = {}
# What the "other" bucket actually holds: the first 20 distinct unknown method
# names (truncated), so a flood of junk headers can't grow it without bound.
_MCP_OTHER_LIMIT = 20
_mcp_other_methods: dict[str, int] = {}

# Badge SVG fetches per agent slug. READMEs embed badges via GitHub's camo
# proxy, which caches the SVG and strips referrers — so GA never sees this
# reach at all, and each fetch here represents MANY actual README views
# (camo re-fetches only when its cached copy expires). Counted only for
# badges that resolve to a real agent, so scanner noise can't inflate it.
_badge_counters: dict[str, int] = {}


def _count_badge(slug: str) -> None:
    _badge_counters[slug] = _badge_counters.get(slug, 0) + 1


# The /live/ page polls /api/v1/usage, and the badge/status widgets poll their
# own endpoints. Counting those would let our own site inflate the "machine
# consumer" numbers it is reporting — the same self-counting mistake the
# verify feed made. Excluded from machine_usage by path.
# The homepage's own deferred board rows are a page asset fetched by people
# searching the leaderboard, not machine use of the data surface.
_USAGE_EXCLUDED_PATHS = frozenset({"/api/v1/usage", "/data/board-rest.json", "/data/search-index.json"})

# Daily snapshots are the registry's irreplaceable asset: one 4MB file per day
# holding every row with all 62 fields, including trust_breakdown and
# scorecard_checks — the scoring internals, not just the published scores. They
# were never deliberately published; the catch-all StaticFiles mount over
# OUTPUT_DIR simply exposed them, and the date-based filenames make the whole
# corpus enumerable with a loop. The site itself reads these from DISK, never
# over HTTP, so refusing them here costs nothing.
#
# The curated public history surface stays open and unaffected:
# GET /api/v1/agents/<slug>/history (90-day window, whitelisted fields).
_PRIVATE_SNAPSHOT_PREFIXES = ("/output/history/", "/output/history")
# The generator's API-response cache lives on the volume beside the site
# (see _refresh_cache_dir); it is never meant to be served.
_PRIVATE_CACHE_PREFIX = "/.cache"


_PUBLIC_SNAPSHOT_RE = re.compile(r"^/data/history/(\d{4}-\d{2}-\d{2})\.json$")


def _is_expired_public_snapshot(path: str) -> bool:
    """/data/history/<date>.json copies older than the public 90-day window.
    The same window as /api/v1/agents/<slug>/history and the open-core plan
    (extended history is the reserved future tier); the files stay on disk,
    and the quarterly CC BY export remains the path to the full series."""
    m = _PUBLIC_SNAPSHOT_RE.match(path)
    if not m:
        return False
    cutoff = (datetime.now(timezone.utc).date() - timedelta(days=_HISTORY_PUBLIC_DAYS - 1)).isoformat()
    return m.group(1) < cutoff


def _is_private_snapshot_path(path: str) -> bool:
    """True for raw daily-snapshot and cache paths, which must not be served publicly."""
    return (path.startswith(_PRIVATE_SNAPSHOT_PREFIXES[0]) or path == _PRIVATE_SNAPSHOT_PREFIXES[1]
            or path == _PRIVATE_CACHE_PREFIX or path.startswith(_PRIVATE_CACHE_PREFIX + "/"))


def _count_machine_usage(path: str, mcp_method: str | None = None) -> None:
    if path in _USAGE_EXCLUDED_PATHS:
        return
    channel = None
    if path.startswith("/api/v1/"):
        channel = "api_v1"
    elif path == "/mcp" or path.startswith("/mcp/"):
        channel = "mcp"
    elif path.startswith("/data/exports/"):
        channel = "exports"
    elif path.startswith("/registry/") and "/v0.1/" in path:
        channel = "registry"
    elif path.startswith("/data/agents/") and path.endswith(".json"):
        # Per-agent records carry the signed trust_credential, so their
        # fetches are the closest measure of anyone verifying a record.
        channel = "agent_records"
    elif path.startswith("/data/") and path.endswith(".json"):
        channel = "data_json"
    if channel is None:
        return
    _usage_counters[channel] += 1
    usage.bump(channel)  # durable rollup behind /live/
    if channel == "mcp":
        key = mcp_method if mcp_method in _MCP_METHODS else ("other" if mcp_method else "legacy")
        _mcp_method_counters[key] = _mcp_method_counters.get(key, 0) + 1
        if key == "other":
            name = mcp_method[:40]
            if name in _mcp_other_methods or len(_mcp_other_methods) < _MCP_OTHER_LIMIT:
                _mcp_other_methods[name] = _mcp_other_methods.get(name, 0) + 1


@app.middleware("http")
async def _cache_headers(request, call_next):
    path = request.url.path
    _count_machine_usage(path, request.headers.get("mcp-method"))
    if path not in _HEALTHCHECK_PATHS:
        redirect_target = _canonical_redirect_target(request)
        if redirect_target is not None:
            status_code = 301 if request.method in {"GET", "HEAD"} else 308
            return RedirectResponse(redirect_target, status_code=status_code)
        retired = _retired_response(path)
        if retired is not None:
            return retired
        if _is_private_snapshot_path(path) or _is_expired_public_snapshot(path):
            # 404, not 403: don't confirm that a given date's snapshot exists.
            return Response("Not Found", status_code=404, media_type="text/plain")

    if path == "/mcp" and request.method == "GET":
        # A GET opens the optional server-to-client stream, which this
        # stateless JSON server doesn't offer (the spec allows 405). Under MCP
        # SDK v2 the request would otherwise hang open until the client gives up.
        return Response(status_code=405, headers={"Allow": "POST"})
    if path == "/mcp" and request.method == "POST":
        if not _mcp_enabled():
            return JSONResponse(
                {"jsonrpc": "2.0", "id": None, "error": {"code": -32000,
                 "message": "HVTracker MCP is temporarily paused to stay within budget. "
                            "The website and the HTTP API at /api/v1/mcp/verify remain available."}},
                status_code=503,
                headers={"Retry-After": "3600"},
            )
        if _is_mcp_rate_limited(request):
            return JSONResponse(
                {"jsonrpc": "2.0", "id": None, "error": {"code": -32029,
                 "message": "Rate limit exceeded — too many MCP requests from your IP. Retry in a minute."}},
                status_code=429,
                headers={"Retry-After": "60"},
            )

    response = await call_next(request)

    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
    response.headers["Permissions-Policy"] = "camera=(), microphone=(), geolocation=()"
    response.headers["Content-Security-Policy-Report-Only"] = _CSP
    if _external_scheme(request) == "https":
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains; preload"
    if not path.startswith("/badge/"):
        response.headers["X-Frame-Options"] = "SAMEORIGIN"

    if response.status_code != 200 or "cache-control" in {k.lower() for k in response.headers}:
        return response
    # User-specific / auth / API responses must never be cached (or shared by a
    # CDN): /account shows the signed-in user's data, /login varies by auth state.
    if (path in {"/account", "/account/", "/login", "/login/"}
            or path.startswith("/api/") or path.startswith("/auth/")):
        response.headers["Cache-Control"] = "no-store"
        return response
    if path.startswith("/data/") and path.endswith(".json"):
        response.headers["Cache-Control"] = _JSON_CACHE
    elif path in ("/sitemap.xml", "/feed.json") or path.endswith("/feed.xml"):
        # Feeds + sitemap regenerate at most once per render; without a header
        # Cloudflare's cache rule marks them BYPASS and every crawler poll hits
        # the origin.
        response.headers["Cache-Control"] = _JSON_CACHE
    elif path.endswith("/") or path.endswith(".html"):
        response.headers["Cache-Control"] = _HTML_CACHE
    elif path in ("",) or path == "/":
        response.headers["Cache-Control"] = _HTML_CACHE
    elif path.endswith((".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg",
                        ".ico", ".woff2", ".woff", ".ttf")):
        # OG cards, logos, favicons, fonts — without this they're BYPASS.
        response.headers["Cache-Control"] = _ASSET_CACHE
    return response


# Compress on the way out. Cloudflare compresses for visitors, so the site
# always *looked* fine, but nothing compressed the origin->edge hop and Railway
# bills that hop uncompressed: the homepage left the container at 6.4MB instead
# of 0.54MB, and with /data/latest.json (9.0MB raw) the two accounted for ~94%
# of a ~98GB/month egress bill. StaticFiles never compresses on its own.
# Registered LAST so it is the outermost middleware (add_middleware inserts at
# index 0 and the stack is built inside-out), which puts it around both
# _cache_headers and the StaticFiles mount.
# compresslevel=6, not the library default of 9: measured on the real homepage,
# 9 costs 2.1x the CPU of 6 to gain 6.6% (30.6ms/538KB vs 65.3ms/503KB).
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=6)

# ---- data.json access (mtime-cached) -------------------------------------

_cache: dict = {"mtime": None, "data": None}


def load_data() -> dict:
    """Return the current leaderboard from the volume's data.json (cached by mtime)."""
    try:
        mtime = os.path.getmtime(DATA_PATH)
    except OSError:
        return {"agents": [], "total": 0, "updated": None}
    if _cache["mtime"] != mtime:
        for _ in range(3):
            try:
                with open(DATA_PATH, encoding="utf-8") as f:
                    _cache["data"] = json.load(f)
                _cache["mtime"] = mtime
                break
            except json.JSONDecodeError:
                if _cache["data"] is not None:
                    return _cache["data"]
                time.sleep(0.05)
        else:
            return {"agents": [], "total": 0, "updated": None}
    return _cache["data"]


def _catalog_agent_count() -> int:
    try:
        with open(os.path.join(BASE_DIR, "agents.json"), encoding="utf-8") as f:
            return len(json.load(f))
    except (OSError, json.JSONDecodeError, TypeError):
        return 0


def _source_render_count() -> int:
    catalog_count = 0
    try:
        with open(os.path.join(BASE_DIR, "agents.json"), encoding="utf-8") as f:
            agents = json.load(f)
        catalog_count = sum(
            1 for agent in agents
            if agent.get("status") != "legacy" and agent.get("listing_status") != "legacy"
        )
    except (OSError, json.JSONDecodeError, TypeError):
        pass
    path = os.path.join(BASE_DIR, "data", "render_state.json")
    try:
        with open(path, encoding="utf-8") as f:
            payload = json.load(f)
        return max(catalog_count, len(payload.get("rows") or []))
    except (OSError, json.JSONDecodeError, TypeError):
        return catalog_count


def _normalize_github_repo(value: str | None) -> str | None:
    if not value:
        return None
    text = value.strip()
    if not text or text.lower().startswith("pkg:"):
        return None  # a package URL is never a repo ("pkg:npm/x" has two parts)
    text = text.replace("git@github.com:", "https://github.com/")
    text = text.replace("ssh://git@github.com/", "https://github.com/")
    text = text.replace("git+https://github.com/", "https://github.com/")
    text = text.replace("git://github.com/", "https://github.com/")
    if "github.com/" in text.lower():
        m = re.search(r"github\.com[:/]+([^/\s]+)/([^/\s?#]+)", text, re.IGNORECASE)
        if not m:
            return None
        owner, repo = m.group(1), m.group(2)
    else:
        parts = [part for part in text.strip("/").split("/") if part]
        if len(parts) != 2:
            return None
        owner, repo = parts
    repo = repo.removesuffix(".git")
    return f"{owner.lower()}/{repo.lower()}" if owner and repo else None


def _tracked_repo_lookup() -> dict[str, dict]:
    return {
        (agent.get("repo") or "").lower(): agent
        for agent in db.load_agents()
        if agent.get("repo")
    }


def load_manual_candidates() -> list[dict]:
    path = os.path.join(BASE_DIR, "docs", "import-candidates.json")
    try:
        with open(path, encoding="utf-8") as f:
            payload = json.load(f)
    except (OSError, json.JSONDecodeError, TypeError):
        return []
    if not isinstance(payload, list):
        return []

    tracked = _tracked_repo_lookup()
    items = []
    for entry in payload:
        if isinstance(entry, str):
            source_repo = entry
            item = {"repo": entry}
        elif isinstance(entry, dict):
            source_repo = entry.get("repo") or entry.get("url") or entry.get("github")
            item = dict(entry)
        else:
            continue
        normalized_repo = _normalize_github_repo(source_repo)
        if not normalized_repo:
            continue
        tracked_agent = tracked.get(normalized_repo)
        item["repo"] = normalized_repo
        item["url"] = f"https://github.com/{normalized_repo}"
        item["tracked"] = tracked_agent is not None
        item["tracked_name"] = tracked_agent.get("name") if tracked_agent else None
        items.append(item)
    return items


_HEADER_ENV = None
SITE_HEADER_MARKER = "<!--#site-header-->"


def _site_header_html(updated: str) -> str:
    """The site header, rendered from templates/_site_header.html.j2: the same
    partial every generated page includes, so marketing/login pages and the
    hand-written tools (/scan/, /verify/, /live/, /compare/) can't drift from
    it (they each carried a hand-copied header with stale links before)."""
    global _HEADER_ENV
    if _HEADER_ENV is None:
        from jinja2 import Environment, FileSystemLoader
        _HEADER_ENV = Environment(loader=FileSystemLoader(os.path.join(BASE_DIR, "templates")),
                                  autoescape=True)
    return _HEADER_ENV.get_template("_site_header.html.j2").render(
        updated=updated, auth_js_hash=_AUTH_JS_VER)


def _with_site_header(html: str) -> str:
    """Fill a hand-written page's <!--#site-header--> slot."""
    return html.replace(SITE_HEADER_MARKER, _site_header_html(""), 1)


def _runtime_git_sha() -> str | None:
    for key in (
        "RAILWAY_GIT_COMMIT_SHA",
        "GIT_COMMIT_SHA",
        "GITHUB_SHA",
        "SOURCE_COMMIT",
        "COMMIT_SHA",
    ):
        value = (os.environ.get(key) or "").strip()
        if value:
            return value
    return None


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _parse_updated_timestamp(value: str | None) -> datetime | None:
    if not value:
        return None
    text = value.strip()
    for fmt in ("%Y-%m-%d %H:%M UTC", "%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%dT%H:%M:%S.%f%z"):
        try:
            parsed = datetime.strptime(text, fmt)
            if parsed.tzinfo is None:
                parsed = parsed.replace(tzinfo=timezone.utc)
            return parsed.astimezone(timezone.utc)
        except ValueError:
            continue
    try:
        parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=timezone.utc)
        return parsed.astimezone(timezone.utc)
    except ValueError:
        return None


def _read_refresh_status() -> dict:
    try:
        with open(REFRESH_STATUS_PATH, encoding="utf-8") as f:
            payload = json.load(f)
        return payload if isinstance(payload, dict) else {}
    except (OSError, json.JSONDecodeError, TypeError):
        return {}


def _write_refresh_status(status: dict) -> None:
    with open(REFRESH_STATUS_PATH, "w", encoding="utf-8") as f:
        json.dump(status, f, indent=2, sort_keys=True)


def _mark_refresh_status(**updates) -> dict:
    status = _read_refresh_status()
    status.update(updates)
    _write_refresh_status(status)
    return status


def find_agent(repo: str) -> dict | None:
    repo = repo.lower()
    for a in load_data().get("agents", []):
        if a["repo"].lower() == repo:
            return a
    return None


def find_agent_by_slug(slug: str) -> dict | None:
    slug = slug.lower()
    for a in load_data().get("agents", []):
        if (a.get("slug") or "").lower() == slug:
            return a
    return None


_identifiers_cache: dict = {"mtime": None, "purl": {}, "registry": {}}


def _identifier_slugs(identifier: str) -> list[str]:
    """Listing slugs that data/identifiers.json (written each render) ties to a
    package URL (pkg:npm/..., pkg:pypi/..., pkg:cargo/...) or an MCP registry
    name. Registry names match case-insensitively."""
    path = os.path.join(OUTPUT_DIR, "data", "identifiers.json")
    try:
        mtime = os.path.getmtime(path)
        if _identifiers_cache["mtime"] != mtime:
            with open(path, encoding="utf-8") as f:
                ids = json.load(f).get("identifiers") or {}
            _identifiers_cache.update(
                mtime=mtime, purl=ids.get("purl") or {},
                registry={k.lower(): v for k, v in (ids.get("mcp-registry") or {}).items()})
    except (OSError, ValueError):
        return []
    purl = mcp_trust.parse_purl(identifier)
    if purl:
        return _identifiers_cache["purl"].get(purl, [])
    return _identifiers_cache["registry"].get(identifier.lower(), [])


def find_agent_by_key(key: str) -> dict | None:
    """A tracked agent by exact slug, then exact display name, then a package
    URL or MCP registry name from the identifier index, then npm/PyPI package
    id. Slugs and names are unique; package ids are not across
    ecosystems (npm `fastmcp` is punkpeye/fastmcp, PyPI `fastmcp` is
    PrefectHQ/fastmcp, slug `fastmcp`). A single first-match pass in board
    order returned whichever ranked higher, so "fastmcp" resolved to the
    TypeScript port. An indexed identifier resolves only when it names one
    listing; with several there is nothing to choose by."""
    raw = (key or "").strip()
    key = raw.lower()
    if not key:
        return None
    agents = load_data().get("agents", [])
    for field in ("slug", "name"):
        for a in agents:
            if (a.get(field) or "").strip().lower() == key:
                return a
    slugs = _identifier_slugs(raw)
    if len(slugs) == 1:
        indexed = find_agent_by_slug(slugs[0])
        if indexed is not None:
            return indexed
    for a in agents:
        if key in ((a.get("npm_package") or "").lower(), (a.get("pypi_package") or "").lower()):
            return a
    return None


def _resolve_registry_agent(identifier: str) -> dict | None:
    """Resolve an identifier to a tracked agent: GitHub repo/URL first, then
    find_agent_by_key (slug, display name, npm/pypi package). Shared by
    /api/v1/mcp/verify, scan_stack and the MCP server so all agree on what a
    string maps to."""
    identifier = (identifier or "").strip()
    if not identifier:
        return None
    repo = _normalize_github_repo(identifier)
    agent = find_agent(repo) if repo else None
    if agent is None:
        agent = find_agent_by_key(identifier)
    return agent


_SCAN_MAX_ITEMS = 60


def _parse_scan_input(text: str) -> list[str]:
    """Extract candidate identifiers from pasted requirements.txt, package.json,
    an MCP client config, or a plain (newline/comma) list. Best-effort and
    tolerant — unknown shapes fall back to line parsing."""
    text = (text or "").strip()
    if not text:
        return []
    ids: list[str] = []
    try:
        obj = json.loads(text)
    except (json.JSONDecodeError, ValueError):
        obj = None
    if isinstance(obj, dict):
        for key in ("dependencies", "devDependencies", "peerDependencies", "optionalDependencies"):
            dep = obj.get(key)
            if isinstance(dep, dict):
                ids.extend(dep.keys())
        servers = obj.get("mcpServers")
        if isinstance(servers, dict):
            for name, cfg in servers.items():
                ids.append(name)
                if isinstance(cfg, dict) and isinstance(cfg.get("url"), str):
                    ids.append(cfg["url"])
    if not ids:
        # requirements.txt / plain list: strip version specifiers, extras, markers, comments.
        for raw in re.split(r"[\n,]", text):
            line = raw.strip()
            if not line or line.startswith("#") or line.startswith("-"):
                continue
            token = re.split(r"[<>=!~;\[ @]", line, maxsplit=1)[0].strip()
            if token:
                ids.append(token)
    seen, out = set(), []
    for i in ids:
        i = i.strip()
        k = i.lower()
        if i and k not in seen:
            seen.add(k)
            out.append(i)
    return out[:_SCAN_MAX_ITEMS]


# ---- JSON API ------------------------------------------------------------

@app.api_route(
    "/.well-known/mcp/server-card.json",
    methods=["GET", "HEAD"],
    include_in_schema=False,
)
def smithery_mcp_server_card():
    return JSONResponse(mcp_server.server_card(), headers={"Cache-Control": _JSON_CACHE})


@app.api_route("/healthz", methods=["GET", "HEAD"])
@app.api_route("/healthz/", methods=["GET", "HEAD"])
def healthz():
    d = load_data()
    source_fingerprint = _compute_render_fingerprint()
    stored_fingerprint = _read_render_fingerprint()
    refresh_status = _read_refresh_status()
    updated_at = _parse_updated_timestamp(d.get("updated"))
    data_age_seconds = None
    data_fresh = None
    if updated_at is not None:
        data_age_seconds = max(0, int((datetime.now(timezone.utc) - updated_at).total_seconds()))
        data_fresh = data_age_seconds <= int(MAX_DATA_AGE.total_seconds())
    return {
        "status": "ok",
        "agents": d.get("total", 0),
        "catalog_agents": _catalog_agent_count(),
        "source_render_agents": _source_render_count(),
        "updated": d.get("updated"),
        "max_data_age_seconds": int(MAX_DATA_AGE.total_seconds()),
        "data_age_seconds": data_age_seconds,
        "data_fresh": data_fresh,
        "git_sha": _runtime_git_sha(),
        "source_render_fingerprint": source_fingerprint,
        "stored_render_fingerprint": stored_fingerprint,
        "render_in_sync": bool(stored_fingerprint and stored_fingerprint == source_fingerprint),
        "refresh_in_progress": bool(refresh_status.get("in_progress")),
        "last_refresh_started_at": refresh_status.get("last_started_at"),
        "last_refresh_completed_at": refresh_status.get("last_completed_at"),
        "last_refresh_succeeded": refresh_status.get("last_succeeded"),
        "last_refresh_mode": refresh_status.get("last_mode"),
        "last_refresh_error": refresh_status.get("last_error"),
        "scheduler_running": bool(_scheduler is not None and _scheduler.running),
        "scheduler_error": _scheduler_error,
        "scheduled_jobs": _scheduled_jobs(),
        "process_rss_mb": _process_rss_mb(),
        "api_cache_entries": _api_cache_entries(),
        "machine_usage": {"since": _USAGE_SINCE, **_usage_counters,
                          "mcp_by_method": dict(_mcp_method_counters),
                          "mcp_other_methods": dict(_mcp_other_methods)},
        "badge_fetches": {
            "since": _USAGE_SINCE,
            "total": sum(_badge_counters.values()),
            "top": dict(sorted(_badge_counters.items(), key=lambda kv: -kv[1])[:20]),
        },
    }


@app.api_route("/api/agents", methods=["GET", "HEAD"])
def api_agents(q: str = "", category: str = "", sort: str = "rank",
               order: str = "asc", limit: int = 50, offset: int = 0):
    """Search/filter/sort/paginate the leaderboard."""
    agents = load_data().get("agents", [])
    if q:
        ql = q.lower()
        agents = [a for a in agents
                  if ql in a["name"].lower() or ql in a["repo"].lower()
                  or ql in (a.get("description") or "").lower()]
    if category:
        cl = category.lower()
        agents = [a for a in agents if (a.get("category") or "").lower() == cl]
    reverse = order == "desc"
    agents = sorted(agents, key=lambda a: (a.get(sort) is None, a.get(sort)), reverse=reverse)
    total = len(agents)
    page = agents[offset:offset + max(0, min(limit, 200))]
    return {"total": total, "count": len(page), "offset": offset, "agents": page}


@app.api_route("/api/agents/{owner}/{repo}", methods=["GET", "HEAD"])
def api_agent(owner: str, repo: str):
    agent = find_agent(f"{owner}/{repo}")
    if not agent:
        return JSONResponse({"error": "not found"}, status_code=404)
    return agent


@app.api_route("/api/feed", methods=["GET", "HEAD"])
def api_feed():
    path = os.path.join(OUTPUT_DIR, "feed.json")
    if not os.path.isfile(path):
        return JSONResponse({"error": "feed not built yet"}, status_code=503)
    with open(path, encoding="utf-8") as f:
        return JSONResponse(json.load(f))


@app.api_route("/api/import-candidates", methods=["GET", "HEAD"])
def api_import_candidates(status: str = "", category: str = "", tracked: str = "",
                          limit: int = 100, offset: int = 0):
    candidates = load_manual_candidates()
    if status:
        status_l = status.lower()
        candidates = [c for c in candidates if (c.get("status") or "").lower() == status_l]
    if category:
        category_l = category.lower()
        candidates = [c for c in candidates if (c.get("category") or "").lower() == category_l]
    if tracked:
        tracked_flag = tracked.lower()
        want_tracked = tracked_flag in ("1", "true", "yes")
        if tracked_flag in ("1", "true", "yes", "0", "false", "no"):
            candidates = [c for c in candidates if bool(c.get("tracked")) is want_tracked]
    total = len(candidates)
    page = candidates[offset:offset + max(0, min(limit, 500))]
    return {"total": total, "count": len(page), "offset": offset, "candidates": page}


_API_V1_CACHE = "public, max-age=900"
_API_V1_CORS = "*"


@app.get("/api/v1/graph")
def api_v1_graph():
    path = os.path.join(OUTPUT_DIR, "data", "graph.json")
    if not os.path.isfile(path):
        return JSONResponse({"error": "graph not built yet"}, status_code=503)
    with open(path, encoding="utf-8") as f:
        data = json.load(f)
    return JSONResponse(data, headers={
        "Cache-Control": _API_V1_CACHE,
        "Access-Control-Allow-Origin": _API_V1_CORS,
    })


@app.get("/api/v1/agents")
def api_v1_agents():
    """The agent board. Agent-class rows only — see the note below.

    The v1 stability promise is that fields are add-only. Returning a second
    listing class here would change *which rows* the endpoint serves, not just
    their fields, and this is the dominant machine channel (the MCP server
    pulls and caches this board). Skills have their own rank space, so their
    `rank` is not comparable with an agent's anyway. The complete multi-class
    snapshot stays available at /data/latest.json.
    """
    if not os.path.isfile(DATA_PATH):
        return JSONResponse({"error": "data not built yet"}, status_code=503)
    with open(DATA_PATH, encoding="utf-8") as f:
        data = json.load(f)
    rows = data.get("agents")
    if isinstance(rows, list):
        data = {**data, "agents": [
            r for r in rows if (r.get("listing_class") or "agent") == "agent"
        ]}
    return JSONResponse(data, headers={
        "Cache-Control": _API_V1_CACHE,
        "Access-Control-Allow-Origin": _API_V1_CORS,
    })


# Public history window. The full time series is reserved future paid
# surface per docs/open-core.md ("90-day public history: Yes ... Extended
# history: paid"); this endpoint never serves beyond it.
_HISTORY_PUBLIC_DAYS = 90
# Snapshots never served as history: 2026-09-21 holds 441 rows ranked among
# themselves (production briefly ran main's stale roster). Mirrors
# fetch_and_build.PARTIAL_SNAPSHOT_DATES — inlined to keep the generator out
# of the web process; tests/test_partial_snapshots.py keeps the two equal.
_PARTIAL_SNAPSHOT_DATES = frozenset({"2026-09-21"})
# Mirrors fetch_and_build.snapshot_is_degraded (same test pins them): a board
# with >20% provisional rows (2026-09-22) is never served as history.
_DEGRADED_PENDING_SHARE = 0.2


def _snapshot_is_degraded(snap: dict) -> bool:
    agents = snap.get("agents") or []
    if not agents:
        return False
    pending = sum(1 for a in agents if a.get("pending_signals"))
    return pending > _DEGRADED_PENDING_SHARE * len(agents)
# Per-day public fields — the subset already public on the agent page and
# the CC BY 4.0 dataset export. NOT the full snapshot row.
_HISTORY_PUBLIC_FIELDS = (
    "rank", "trust_score", "evidence_grade", "coverage_grade",
    "trust_confidence", "has_provenance", "scorecard_score",
    "signed_commits_ratio", "stars", "weekly_downloads", "days_ago",
    "listing_status", "methodology_version",
)


@app.get("/api/v1/agents/{slug}/history")
def api_v1_agent_history(slug: str):
    """Last 90 days of an agent's public trust fields, one entry per daily
    snapshot (oldest first). Extended history stays reserved for a future
    paid tier, so the window is capped regardless of how much we hold."""
    agent = find_agent_by_slug(slug)
    if agent is None:
        return JSONResponse({"error": "agent not found"}, status_code=404,
                            headers={"Access-Control-Allow-Origin": _API_V1_CORS})
    repo_key = (agent.get("repo") or "").lower()
    # The per-repo file each render writes, shared with the MCP
    # get_agent_history tool. This endpoint used to re-parse every daily
    # snapshot (~5 MB each, up to 90) on every request: ~4 s per call.
    entries = mcp_server._history_points(repo_key)

    return JSONResponse({
        "slug": agent.get("slug"),
        "repo": agent.get("repo"),
        "name": agent.get("name"),
        "window_days": _HISTORY_PUBLIC_DAYS,
        "count": len(entries),
        "note": ("Public 90-day window. Full history is available under the "
                 "CC BY 4.0 quarterly dataset export and future extended-history access."),
        "license": "CC BY 4.0 — https://creativecommons.org/licenses/by/4.0/",
        "history": entries,
    }, headers={
        "Cache-Control": _API_V1_CACHE,
        "Access-Control-Allow-Origin": _API_V1_CORS,
    })


@app.get("/api/v1/mcp/verify")
def api_v1_mcp_verify(request: Request, server: str = ""):
    """Pre-connect trust verdict for an agent or MCP server.

    `server` may be a GitHub repo (owner/name or URL) or an npm/pypi package.
    Curated agents return a full signed verdict. An unlisted GitHub repo falls
    through to open lookup (P2): a free PROVISIONAL verdict if it is an AI
    project with >= 1,000 stars, else it is routed to /submit. The attestation
    is signed so the verdict is verifiable; verification is always free.
    """
    import mcp_trust
    server = (server or "").strip()
    cors = {"Access-Control-Allow-Origin": _API_V1_CORS}
    if not server:
        return JSONResponse({"error": "missing required query param: server"},
                            status_code=400, headers=cors)

    # 1) Curated registry (the gold "Verified" tier).
    agent = None
    repo = _normalize_github_repo(server)
    if repo:
        agent = find_agent(repo)
    if agent is None:
        # Users naturally type the name they see on the leaderboard
        # ("headroom") rather than the owner/repo or package id.
        agent = find_agent_by_key(server)
    if agent is not None:
        verdict = mcp_trust.evaluate(agent, server)
        verdict["attestation"] = mcp_trust.build_attestation(verdict)
        verify_log.record(verdict.get("resolved") or repo or server, agent.get("name"),
                          verdict.get("grade"), verdict.get("trusted"), False, agent.get("stars"))
        return JSONResponse(verdict, headers={**cors, "Cache-Control": _API_V1_CACHE})

    # 2) Unlisted and not a GitHub repo -> submit funnel (no scoring).
    if not repo:
        verdict = mcp_trust.evaluate(None, server)
        verdict["attestation"] = mcp_trust.build_attestation(verdict)
        return JSONResponse(verdict, headers=cors)

    # 3) Unlisted GitHub repo -> open lookup (cached, rate-limited).
    cached = _olookup_cache_get(repo)
    if cached is not None:
        return JSONResponse(cached, headers=cors)
    if _is_open_lookup_limited(request):
        return JSONResponse(
            {"server": server, "tracked": False, "trusted": False,
             "eligibility": "rate_limited", "submit_url": "https://hvtracker.net/submit",
             "reasons": ["Too many instant checks from your IP — try again in a few minutes."]},
            status_code=429, headers=cors)
    import open_lookup
    verdict = open_lookup.evaluate_open(server, repo, os.environ.get("GITHUB_TOKEN", ""))
    if verdict.get("eligibility") == "ok":
        verdict["attestation"] = mcp_trust.build_attestation(verdict)
        verify_log.record(verdict.get("resolved") or repo, None, verdict.get("grade"),
                          verdict.get("trusted"), True, verdict.get("stars"))
    _olookup_cache_put(repo, verdict)
    return JSONResponse(verdict, headers=cors)


@app.get("/api/v1/verify/recent")
def api_v1_verify_recent():
    """Public feed: the last successfully checked projects (transparency).

    Every free check is public by default and appears here. Private checks will
    require the paid tier or a public submission.
    """
    return JSONResponse({"checks": verify_log.recent()}, headers={
        "Cache-Control": "public, max-age=30, s-maxage=60",
        "Access-Control-Allow-Origin": _API_V1_CORS,
    })


@app.get("/api/v1/usage")
def api_v1_usage(hours: int = 24):
    """Public machine-channel usage: how much the trust layer is actually used.

    Powers /live/. Counts only channel/tool names and the hour they happened
    in — no IPs, arguments, or anything user-identifying. Excluded from
    machine_usage itself so the page cannot inflate what it reports.
    """
    hours = max(1, min(int(hours or 24), 168))
    # Copy: snapshot() hands back its own cached dict, and this response is
    # per-request. The header widget on every page reads freshness AND activity
    # from this one small, edge-cached response — it previously pulled the
    # multi-megabyte /data/latest.json just to read this string.
    payload = dict(usage.snapshot(hours))
    payload["data_updated"] = load_data().get("updated")
    return JSONResponse(payload, headers={
        # Short edge TTL: enough to absorb many viewers polling at once while
        # still feeling live. Matches the client poll interval.
        "Cache-Control": "public, max-age=10, s-maxage=10",
        "Access-Control-Allow-Origin": _API_V1_CORS,
    })


@app.post("/api/v1/scan")
async def api_v1_scan(request: Request):
    """Bulk pre-connect trust check for a whole dependency set.

    POST JSON {"input": "<requirements.txt | package.json | MCP config | list>"}.
    Each identifier is resolved against the curated registry and returned with a
    per-item verdict plus a summary. Registry-only (no GitHub open lookup) so it
    stays cheap and fast — the same engine that backs /api/v1/mcp/verify."""
    import mcp_trust
    cors = {"Access-Control-Allow-Origin": _API_V1_CORS}
    if _is_mcp_rate_limited(request):
        return JSONResponse({"error": "rate limited — slow down and retry shortly"},
                            status_code=429, headers=cors)
    try:
        payload = await request.json()
    except Exception:
        payload = None
    text = payload.get("input") if isinstance(payload, dict) else None
    if not isinstance(text, str) or not text.strip():
        return JSONResponse(
            {"error": 'POST JSON {"input": "<requirements.txt / package.json / mcp config / list>"}'},
            status_code=400, headers=cors)
    if len(text) > 20000:
        return JSONResponse({"error": "input too large (max 20000 chars)"},
                            status_code=400, headers=cors)
    identifiers = _parse_scan_input(text)
    results = []
    summary = {"total": 0, "tracked": 0, "untracked": 0, "trusted": 0, "avg_trust": None}
    scores = []
    for ident in identifiers:
        v = mcp_trust.evaluate(_resolve_registry_agent(ident), ident)
        results.append({
            "input": ident,
            "tracked": v["tracked"],
            "trusted": v["trusted"],
            "grade": v["grade"],
            "trust_score": v["trust_score"],
            "resolved": v.get("resolved"),
            "slug": v.get("slug"),
        })
        summary["total"] += 1
        summary["tracked" if v["tracked"] else "untracked"] += 1
        if v["trusted"]:
            summary["trusted"] += 1
        if v["trust_score"] is not None:
            scores.append(v["trust_score"])
    # Average HVTrust across the tracked (scored) items in the stack.
    if scores:
        summary["avg_trust"] = round(sum(scores) / len(scores), 1)
    return JSONResponse({"summary": summary, "results": results}, headers=cors)


@app.api_route("/scan", methods=["GET", "HEAD"], response_class=HTMLResponse)
@app.api_route("/scan/", methods=["GET", "HEAD"], response_class=HTMLResponse)
def scan_tool():
    if not os.path.isfile(SCAN_TOOL_PATH):
        return HTMLResponse("<p>Scan tool is not available yet.</p>", status_code=503)
    with open(SCAN_TOOL_PATH, encoding="utf-8") as f:
        return HTMLResponse(_with_site_header(f.read()))


@app.api_route("/compare", methods=["GET", "HEAD"], response_class=HTMLResponse)
@app.api_route("/compare/", methods=["GET", "HEAD"], response_class=HTMLResponse)
def compare_tool():
    if not os.path.isfile(COMPARE_TOOL_PATH):
        return HTMLResponse("<p>Compare tool is not available yet.</p>", status_code=503)
    with open(COMPARE_TOOL_PATH, encoding="utf-8") as f:
        return HTMLResponse(_with_site_header(f.read()))


@app.api_route("/compare/{pair}", methods=["GET", "HEAD"], include_in_schema=False)
def compare_pair_noslash(pair: str):
    return RedirectResponse(f"/compare/{pair}/", status_code=301)


@app.api_route("/compare/{pair}/", methods=["GET", "HEAD"], response_class=HTMLResponse)
def compare_pair(pair: str):
    # /compare/<a>-vs-<b>/ serves the static pre-rendered comparison page when
    # one exists (see below), else the interactive tool with both agents
    # preselected. Validate the pair and keep one canonical URL per pair —
    # alphabetical by slug — so it never 404s or splits for two tracked agents.
    parts = pair.split("-vs-")
    if len(parts) != 2 or not parts[0] or not parts[1]:
        return JSONResponse({"detail": "Not Found"}, status_code=404)
    slug_a, slug_b = parts

    slugs = {r["slug"] for r in load_data().get("agents", [])}
    if slug_a not in slugs or slug_b not in slugs:
        return JSONResponse({"detail": "Not Found"}, status_code=404)

    if slug_a > slug_b:
        return RedirectResponse(f"/compare/{slug_b}-vs-{slug_a}/", status_code=301)

    # Serve the static pre-rendered comparison page when it exists (crawlable
    # SEO content with the full header/auth widget); fall back to the
    # interactive tool for ad-hoc pairs that have no generated page.
    static_pair = os.path.join(OUTPUT_DIR, "compare", f"{slug_a}-vs-{slug_b}", "index.html")
    if os.path.isfile(static_pair):
        with open(static_pair, encoding="utf-8") as f:
            return HTMLResponse(f.read())
    # The fallback serves the hub (canonical /compare/) under an unbounded
    # /compare/<a>-vs-<b>/ URL space — noindex so crawlers drop these URLs.
    resp = compare_tool()
    resp.headers["X-Robots-Tag"] = "noindex"
    return resp


@app.api_route("/verify", methods=["GET", "HEAD"], response_class=HTMLResponse)
@app.api_route("/verify/", methods=["GET", "HEAD"], response_class=HTMLResponse)
def verify_tool():
    if not os.path.isfile(VERIFY_TOOL_PATH):
        return HTMLResponse("<p>Verify tool is not available yet.</p>", status_code=503)
    with open(VERIFY_TOOL_PATH, encoding="utf-8") as f:
        return HTMLResponse(_with_site_header(f.read()))


@app.api_route("/live", methods=["GET", "HEAD"], response_class=HTMLResponse)
@app.api_route("/live/", methods=["GET", "HEAD"], response_class=HTMLResponse)
def live_usage_page():
    """Public machine-channel usage. Hand-written source shipped in the image
    (like /verify/), so it reads from BASE_DIR — the numbers themselves are
    fetched client-side from /api/v1/usage."""
    if not os.path.isfile(LIVE_PAGE_PATH):
        return HTMLResponse("<p>Live usage page is not available yet.</p>", status_code=503)
    with open(LIVE_PAGE_PATH, encoding="utf-8") as f:
        return HTMLResponse(_with_site_header(f.read()))


# ---- MCP registry feed (Phase 9) -------------------------------------------
# A subregistry of the official MCP registry (v0.1 API): the entries whose
# source repo HVTracker scores, with HVTrust under server._meta. Each policy is
# its own base URL, so an admin points Copilot's "MCP Registry URL" at e.g.
# https://hvtracker.net/registry/grade-b and "Registry only" admits exactly
# that list. Copilot appends /v0.1/... itself and needs these CORS headers.
_REGISTRY_CORS = {"Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "GET, OPTIONS",
                  "Access-Control-Allow-Headers": "Authorization, Content-Type"}
_registry_cache: dict = {"mtime": None, "doc": None}


def _registry_entries(policy: str) -> list[dict] | None:
    """Entries in `policy`, from data/registry.json (written each render); None
    for an unknown policy."""
    import fetch_and_build
    if policy != "all" and policy not in fetch_and_build.REGISTRY_POLICIES:
        return None
    path = os.path.join(OUTPUT_DIR, "data", "registry.json")
    try:
        mtime = os.path.getmtime(path)
        if _registry_cache["mtime"] != mtime:
            with open(path, encoding="utf-8") as f:
                _registry_cache.update(mtime=mtime, doc=json.load(f))
    except (OSError, ValueError):
        return []
    return [e for e in _registry_cache["doc"].get("entries", []) if policy in e["policies"]]


def _registry_json(body: dict, status: int = 200) -> JSONResponse:
    return JSONResponse(body, status_code=status,
                        headers={**_REGISTRY_CORS, "Cache-Control": "public, max-age=300, s-maxage=900"})


@app.get("/registry/{policy}/v0.1/servers")
def registry_servers(policy: str, cursor: str = "", limit: int = 30, search: str = "",
                     updated_since: str = "", version: str = ""):
    entries = _registry_entries(policy)
    if entries is None:
        return _registry_json({"error": f"Unknown registry '{policy}'"}, 404)
    if updated_since:
        try:
            since = datetime.fromisoformat(updated_since)
        except ValueError:
            return _registry_json({"error": "updated_since must be an RFC 3339 timestamp"}, 400)
        if since.tzinfo is None:
            since = since.replace(tzinfo=timezone.utc)
        entries = [e for e in entries if e["updated_at"] and datetime.fromisoformat(e["updated_at"]) >= since]
    if search:
        entries = [e for e in entries if search.lower() in e["name"].lower()]
    if version and version != "latest":
        entries = [e for e in entries if e["version"] == version]
    if cursor:  # opaque to clients: the last name of the previous page
        entries = [e for e in entries if e["name"] > cursor]
    page = entries[:max(1, min(limit, 100))]
    metadata = {"count": len(page)}
    if len(entries) > len(page):
        metadata["nextCursor"] = page[-1]["name"]
    return _registry_json({"servers": [e["entry"] for e in page], "metadata": metadata})


def _registry_entry(policy: str, server_name: str) -> dict | None:
    return next((e for e in _registry_entries(policy) or [] if e["name"] == server_name), None)


@app.get("/registry/{policy}/v0.1/servers/{server_name:path}/versions/{version}")
def registry_server_version(policy: str, server_name: str, version: str):
    e = _registry_entry(policy, server_name)
    if not e or version not in ("latest", e["version"]):
        return _registry_json({"error": "Server not found"}, 404)
    return _registry_json(e["entry"])


@app.get("/registry/{policy}/v0.1/servers/{server_name:path}/versions")
def registry_server_versions(policy: str, server_name: str):
    """We mirror each server's latest version only."""
    e = _registry_entry(policy, server_name)
    if not e:
        return _registry_json({"error": "Server not found"}, 404)
    return _registry_json({"servers": [e["entry"]], "metadata": {"count": 1}})


@app.options("/registry/{rest:path}")
def registry_preflight(rest: str):
    return Response(status_code=204, headers=_REGISTRY_CORS)


@app.get("/registry", response_class=HTMLResponse)
@app.get("/registry/", response_class=HTMLResponse, include_in_schema=False)
def registry_page():
    counts = {p: len(_registry_entries(p) or []) for p in ("all", "grade-a", "grade-b", "grade-c")}
    pulled = (_registry_cache.get("doc") or {}).get("snapshot_pulled_at") or "not yet"
    rows = "".join(
        f"<tr><td style='word-break:break-all'><code>https://hvtracker.net/registry/{p}</code></td><td>{escape(what)}</td>"
        f"<td style='text-align:right;font-family:var(--font-mono)'>{counts[p]:,}</td></tr>"
        for p, what in (("grade-a", "Grade A only"), ("grade-b", "Grade A or B"),
                        ("grade-c", "Grade A to C"), ("all", "Every entry we score, any grade")))
    body = f"""
    <div class='card' style='min-width:0'>
      <h2>Base URLs</h2>
      <p>Each URL is a complete registry implementing the official MCP Registry API v0.1. Point a client at the base URL; it appends <code>/v0.1/servers</code> itself.</p>
      <div style='overflow-x:auto'><table style='width:100%;border-collapse:collapse;font-size:14px;margin-top:8px'>
        <thead><tr><th style='text-align:left'>Base URL</th><th style='text-align:left'>Admits</th><th style='text-align:right'>Servers</th></tr></thead>
        <tbody>{rows}</tbody>
      </table></div>
      <p style='margin-top:10px;color:var(--muted);font-size:13px'>The graded lists admit an entry only when its publisher is tied to the repository we scored: an <code>io.github.&lt;owner&gt;</code> namespace that matches the repository owner (the registry verifies it by GitHub login), or a domain namespace (verified by DNS) that matches the repository's homepage or its owner's GitHub profile website, both of which only the owner can set. An entry anyone else publishes pointing at a well-scored repository doesn't inherit its grade. Provisional and review-flagged projects are left out too. <code>all</code> keeps every entry and says which check each one passed.</p>
    </div>
    <div class='card' style='min-width:0'>
      <h2>Use it with GitHub Copilot</h2>
      <p>Enterprise: <strong>Settings → AI controls → MCP → MCP Registry URL</strong>. Organization: <strong>Settings → Copilot → Policies → MCP Registry URL</strong>. Paste a base URL above with no <code>/v0.1</code> suffix, then choose <strong>Registry only</strong> to admit just that list, or <strong>Allow all</strong> to show it as recommendations. Copilot matches servers by their registry name, so a developer's locally configured server must use the same name as its registry entry. See <a href='https://docs.github.com/en/copilot/how-tos/administer-copilot/manage-mcp-usage/configure-mcp-registry'>GitHub's guide</a>.</p>
      <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>curl -s "https://hvtracker.net/registry/grade-b/v0.1/servers?limit=3"
curl -s "https://hvtracker.net/registry/all/v0.1/servers/io.github.github%2Fgithub-mcp-server/versions/latest"</code></pre>
    </div>
    <div class='card' style='min-width:0'>
      <h2>What each entry carries</h2>
      <p>The official registry entry unchanged, plus <code>server._meta["net.hvtracker/trust"]</code>: <code>trust_score</code>, <code>grade</code>, <code>coverage_grade</code>, <code>rank</code>, <code>board</code>, <code>provisional</code>, <code>scored_repo</code>, <code>publisher_verified</code> and <code>publisher_check</code>, <code>known_advisory</code> (worst unfixed severity), <code>package_source</code> (does the package link back to the repository), and links to the <code>profile</code> and the signed <code>credential</code>.</p>
      <p style='color:var(--muted);font-size:13px'>Limits: the score is the source repository's provenance (who ships it and whether that changed), not a scan of what the server does, so pair it with a content scanner. Servers that share one repository share its score. We mirror each server's latest version. Registry entries refresh daily (last pull: {escape(str(pulled))}); scores refresh every 4 hours. Copilot's own enforcement matches names only, which GitHub documents as bypassable by editing local configuration.</p>
    </div>"""
    return HTMLResponse(_marketing_page(
        "MCP registry feed with trust scores — HVTracker", "Registry",
        "An MCP registry that only admits servers you can trace", body,
        description="Official MCP Registry entries for the servers HVTracker scores, with HVTrust in _meta and grade-filtered base URLs for GitHub Copilot's Registry only policy.",
        path="/registry/",
        lede="The official MCP Registry entries for the servers HVTracker scores, with each one's trust score attached, as registries your MCP client or GitHub Copilot can point at directly."))


@app.get("/og-v2.png")
def og_v2():
    # Every full render regenerates the site card on the volume with current
    # totals. Serving the copy baked into the image instead left every social
    # preview showing June's "272 active projects" until Oct 2026; that copy is
    # now only the fallback for a volume that hasn't rendered one yet.
    fresh = os.path.join(OUTPUT_DIR, "og-v2.png")
    path = fresh if os.path.isfile(fresh) else os.path.join(BASE_DIR, "og-v2.png")
    return FileResponse(path, media_type="image/png")


@app.get("/og-verify.png")
def og_verify():
    return FileResponse(os.path.join(BASE_DIR, "og-verify.png"), media_type="image/png")


@app.get("/og-mcp.png")
def og_mcp():
    return FileResponse(os.path.join(BASE_DIR, "og-mcp.png"), media_type="image/png")


@app.get("/og-scan.png")
def og_scan():
    return FileResponse(os.path.join(BASE_DIR, "og-scan.png"), media_type="image/png")


@app.get("/favicon.svg")
def favicon_svg():
    return FileResponse(os.path.join(BASE_DIR, "favicon.svg"), media_type="image/svg+xml")


# Crawlers auto-request these regardless of <link> tags, and Bing's favicon
# fetcher in particular wants a real file at the root /favicon.ico convention —
# a 301 to a different format is a known way to end up with a blank globe in the
# results. Raster sources come from scripts/generate_favicons.py.
@app.get("/favicon.ico")
def favicon_ico():
    return FileResponse(os.path.join(BASE_DIR, "favicon.ico"), media_type="image/x-icon")


@app.get("/apple-touch-icon.png")
@app.get("/apple-touch-icon-precomposed.png")
def apple_touch_icon():
    return FileResponse(
        os.path.join(BASE_DIR, "apple-touch-icon.png"), media_type="image/png"
    )


@app.get("/haystack-logo.png")
def haystack_logo():
    return FileResponse(os.path.join(BASE_DIR, "haystack-logo.png"), media_type="image/png")


@app.get("/aipass-logo.png")
def aipass_logo():
    return FileResponse(os.path.join(BASE_DIR, "aipass-logo.png"), media_type="image/png")


@app.get("/composio-logo.svg")
def composio_logo():
    return FileResponse(os.path.join(BASE_DIR, "composio-logo.svg"), media_type="image/svg+xml")


@app.get("/lightrag-logo.png")
def lightrag_logo():
    return FileResponse(os.path.join(BASE_DIR, "lightrag-logo.png"), media_type="image/png")


@app.get("/threadplane-logo.png")
def threadplane_logo():
    return FileResponse(os.path.join(BASE_DIR, "threadplane-logo.png"), media_type="image/png")


_ADOPTER_AVATAR_SLUGS = frozenset({
    "skillspector", "haystack", "lightrag", "composio",
    "aipass", "threadplane", "reversecore-mcp",
})


@app.get("/adopters/{slug}.png")
def adopter_avatar(slug: str):
    # Square avatars for the homepage "Projects carrying the HVTracker badge"
    # cards. Like the *-logo.png routes above, each BASE_DIR file needs an
    # explicit route — the StaticFiles mount only serves the volume.
    if slug not in _ADOPTER_AVATAR_SLUGS:
        return Response(status_code=404)
    return FileResponse(
        os.path.join(BASE_DIR, "adopters", f"{slug}.png"), media_type="image/png"
    )


@app.get("/hex-bg.svg")
def hex_bg():
    return FileResponse(os.path.join(BASE_DIR, "hex-bg.svg"), media_type="image/svg+xml")


# ---- Dynamic SVG badges --------------------------------------------------

def _badge_svg(label: str, value: str, color: str) -> str:
    char_w = 6.1
    label_w = len(label) * char_w + 12
    value_w = len(value) * char_w + 12
    total_w = label_w + value_w
    return f'''<svg xmlns="http://www.w3.org/2000/svg" width="{total_w:.0f}" height="20" role="img" aria-label="{label}: {value}">
  <title>{label}: {value}</title>
  <linearGradient id="s" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient>
  <clipPath id="r"><rect width="{total_w:.0f}" height="20" rx="3" fill="#fff"/></clipPath>
  <g clip-path="url(#r)">
    <rect width="{label_w:.0f}" height="20" fill="#555"/>
    <rect x="{label_w:.0f}" width="{value_w:.0f}" height="20" fill="#{color}"/>
    <rect width="{total_w:.0f}" height="20" fill="url(#s)"/>
  </g>
  <g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" text-rendering="geometricPrecision" font-size="11">
    <text x="{label_w / 2:.0f}" y="15" fill="#010101" fill-opacity=".3">{label}</text>
    <text x="{label_w / 2:.0f}" y="14">{label}</text>
    <text x="{label_w + value_w / 2:.0f}" y="15" fill="#010101" fill-opacity=".3">{value}</text>
    <text x="{label_w + value_w / 2:.0f}" y="14">{value}</text>
  </g>
</svg>'''


@app.get("/badge/{owner}/{repo}/{kind}.svg")
def badge(owner: str, repo: str, kind: str):
    agent = find_agent(f"{owner}/{repo}")
    # Counts both direct hits and badge_by_slug delegations (which all pass
    # through here except -trend, counted at its own file-serving branch).
    if agent:
        _count_badge(agent.get("slug") or f"{owner}/{repo}")
    if kind == "trust":
        score = (agent or {}).get("trust_score", 0) or 0
        color = "34d399" if score >= 55 else "60a5fa" if score >= 30 else "f87171"
        svg = _badge_svg("HVTrust", str(score), color)
    elif kind == "grade":
        grade = (agent or {}).get("evidence_grade", "D")
        color = {"A": "34d399", "B": "60a5fa", "C": "fbbf24", "D": "f87171"}.get(grade, "9ca3af")
        svg = _badge_svg("Grade", grade, color)
    else:
        return JSONResponse({"error": "unknown badge kind"}, status_code=404)
    headers = {"Cache-Control": "public, max-age=3600"}
    if not agent:
        headers["Cache-Control"] = "public, max-age=60"
    return Response(svg, media_type="image/svg+xml", headers=headers)


@app.get("/badge/{slug}.svg")
def badge_by_slug(slug: str):
    kind = "trust"
    if slug.endswith("-grade"):
        slug = slug[:-6]
        kind = "grade"
    elif slug.endswith("-trend"):
        # Trend badges need history (the era-aware rank series), so they are
        # pre-rendered by generate_badges() each render rather than built
        # per-request like the score/grade badges. Serve the file from the
        # render output root (the volume in prod) — NOT BASE_DIR, which is
        # the immutable code dir and never receives rendered badges.
        base_slug = slug[:-6]
        if not find_agent_by_slug(base_slug):
            return JSONResponse({"error": "not found"}, status_code=404)
        path = os.path.join(OUTPUT_DIR, "badge", f"{base_slug}-trend.svg")
        if not os.path.isfile(path):
            return JSONResponse({"error": "not found"}, status_code=404)
        _count_badge(base_slug)
        return FileResponse(path, media_type="image/svg+xml")
    agent = find_agent_by_slug(slug)
    if not agent:
        return JSONResponse({"error": "not found"}, status_code=404)
    owner, repo = agent["repo"].split("/", 1)
    return badge(owner, repo, kind)


# ---- Submission / correction intake --------------------------------------

def _page(name: str) -> str:
    with open(os.path.join(TEMPLATES_DIR, name), encoding="utf-8") as f:
        return f.read()


def _marketing_page(
    title: str,
    eyebrow: str,
    heading: str,
    body_html: str,
    *,
    description: str = "HVTracker validates demand for alerts, data access, sponsorship, submissions, and corrections before building heavier workflows.",
    path: str = "/",
    noindex: bool = False,
    lede: str = "HVTracker is still early, so these pages are intentionally lightweight. The goal is to validate who wants alerts, data access, and sponsorship before building heavier workflows.",
) -> str:
    canonical = f"https://hvtracker.net{path}"
    robots_meta = '\n  <meta name="robots" content="noindex">' if noindex else ""
    return f"""<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>{escape(title)}</title>
  <meta name="description" content="{escape(description)}">
  <link rel="canonical" href="{escape(canonical)}">{robots_meta}
  <meta property="og:type" content="website">
  <meta property="og:title" content="{escape(title)}">
  <meta property="og:description" content="{escape(description)}">
  <meta property="og:url" content="{escape(canonical)}">
  <meta property="og:image" content="https://hvtracker.net/og-v2.png">
  <meta property="og:image:secure_url" content="https://hvtracker.net/og-v2.png">
  <meta property="og:image:type" content="image/png">
  <meta property="og:image:width" content="1200">
  <meta property="og:image:height" content="630">
  <meta property="og:image:alt" content="HVTracker AI trust registry preview">
  <meta name="twitter:card" content="summary_large_image">
  <meta name="twitter:title" content="{escape(title)}">
  <meta name="twitter:description" content="{escape(description)}">
  <meta name="twitter:image" content="https://hvtracker.net/og-v2.png">
  <meta name="twitter:image:alt" content="HVTracker AI trust registry preview">
  <link rel="icon" href="/favicon.ico" sizes="32x32">
  <link rel="icon" href="/favicon.svg" type="image/svg+xml">
  <link rel="apple-touch-icon" href="/apple-touch-icon.png">
  <link rel="stylesheet" href="/static/site.css">
  <style>
    /* Palette and fonts come from site.css. This page used to redefine them
       with older, paler values (--lobster #c67c6d read at 2.5-2.9:1). */
    * {{ box-sizing:border-box; margin:0; padding:0; }}
    body {{
      min-height:100vh; color:var(--ink); font:15px/1.65 var(--font-sans);
      background:var(--paper);
    }}
    a {{ color:inherit; text-decoration:none; }}
    a:hover {{ text-decoration:underline; }}
    .page {{ max-width:1120px; margin:0 auto; padding:24px 24px 48px; background:var(--paper); min-height:100vh; }}
    .shell {{
      max-width:780px; margin:0 auto;
    }}
    .hero {{ padding:28px 0 20px; border-bottom:1px solid var(--line); }}
    .eyebrow {{
      display:inline-block; margin-bottom:12px; color:var(--lobster); font:11px var(--font-mono);
      text-transform:uppercase; letter-spacing:0.08em;
    }}
    h1 {{ margin:0 0 10px; font-size:34px; line-height:1.1; letter-spacing:-0.03em; }}
    .lede {{ margin:0; max-width:680px; color:var(--muted); }}
    .content {{ padding:24px 0 30px; display:grid; gap:22px; }}
    .card {{
      border-top:2px solid var(--line-strong); padding:18px;
      background:var(--paper-2);
    }}
    .card h2 {{ margin:0 0 10px; font-size:15px; }}
    .card p {{ margin:0 0 10px; color:var(--muted); }}
    .grid {{ display:grid; grid-template-columns:repeat(auto-fit, minmax(190px, 1fr)); gap:14px; }}
    .pill {{
      display:inline-block; padding:4px 9px; border-radius:999px; font:11px var(--font-mono);
      background:var(--lobster-soft); color:var(--lobster);
    }}
    ul {{ margin:10px 0 0 18px; padding:0; color:var(--muted); }}
    li + li {{ margin-top:6px; }}
    form {{ display:grid; gap:14px; }}
    label {{ display:grid; gap:6px; font-weight:600; }}
    input, textarea {{
      width:100%; border:1px solid var(--line); border-radius:10px; padding:12px 13px;
      background:var(--paper); color:var(--ink); font:14px var(--font-sans);
    }}
    textarea {{ min-height:120px; resize:vertical; }}
    .actions {{ display:flex; gap:10px; flex-wrap:wrap; align-items:center; }}
    .button {{
      display:inline-flex; align-items:center; justify-content:center; padding:11px 16px; border-radius:10px;
      background:var(--paper); color:var(--ink); border:1px solid var(--line);
      font:700 12px var(--font-mono); text-decoration:none;
    }}
    .button.secondary {{
      background:var(--paper-2); color:var(--muted); border-color:var(--line);
    }}
    .button:hover {{ text-decoration:none; border-color:var(--lobster); color:var(--lobster); }}
    .button.secondary:hover {{ color:var(--ink); border-color:var(--lobster); }}
    .ok {{ color:var(--green); font-weight:700; }}
    footer {{
      margin-top:28px; padding-top:16px; border-top:1px solid var(--line);
      font:11px var(--font-mono); color:var(--muted); text-align:center;
    }}
    .footer-sep {{ color:var(--line-strong); }}
    footer a {{ color:var(--blue-strong); }}
    @media (max-width:760px) {{
      .page {{ padding:24px 20px 40px; }}
      h1 {{ font-size:28px; }}
    }}
  </style>
  <!-- opt out: localStorage.setItem('hvt_notrack','1') -->
  <script>
    (function(){{try{{if(localStorage.getItem("hvt_notrack")==="1"){{window["ga-disable-G-TZ8921LR0K"]=true;return}}}}catch(_){{}}
    var ua=navigator.userAgent||"";if(navigator.webdriver||/HeadlessChrome|Puppeteer|Playwright|Claude\\/[\\d.]+.*Electron\\/|bot|crawl|spider|curl|wget|python-requests/i.test(ua)){{window["ga-disable-G-TZ8921LR0K"]=true}}}})();
  </script>
  <script async src="https://www.googletagmanager.com/gtag/js?id=G-TZ8921LR0K"></script>
  <script>
    window.dataLayer = window.dataLayer || [];
    function gtag(){{dataLayer.push(arguments);}}
    gtag('js', new Date());
    gtag('config', 'G-TZ8921LR0K');
  </script>
  <script defer src="/analytics.js"></script>
</head>
<body>
  {_site_header_html(datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC'))}
  <div class="page">
    <div class="shell">
      <section class="hero">
        <div class="eyebrow">{escape(eyebrow)}</div>
        <h1>{escape(heading)}</h1>
        <p class="lede">{escape(lede)}</p>
      </section>
      <section class="content">
        {body_html}
      </section>
      <footer>
        <a href="/methodology">Methodology</a>
        <span class="footer-sep">&middot;</span>
        <a href="/spec/">Specifications</a>
        <span class="footer-sep">&middot;</span>
        <a href="/data/">Data API</a>
        <span class="footer-sep">&middot;</span>
        <a href="/compare/">Compare</a>
        <span class="footer-sep">&middot;</span>
        <a href="/badges/">Badges</a>
        <span class="footer-sep">&middot;</span>
        <a href="/changelog/">Changelog</a>
        <span class="footer-sep">&middot;</span>
        <a href="/blog/">Blog</a>
        <span class="footer-sep">&middot;</span>
        <a href="https://github.com/YugantM/hvtracker/issues/new?template=agent-listing.yml" target="_blank" rel="noopener">Submit Agent</a>
        <span class="footer-sep">&middot;</span>
        <a href="https://github.com/YugantM/hvtracker" target="_blank" rel="noopener">GitHub</a>
      </footer>
    </div>
  </div>
</body>
</html>"""


def _interest_unavailable() -> HTMLResponse:
    return HTMLResponse(
        _marketing_page(
            "Interest capture unavailable — HVTracker",
            "Temporarily unavailable",
            "Interest capture is offline right now.",
            "<div class='card'><p>The site can still be browsed, but the lead queue is unavailable on this environment. Try again later or reach out through GitHub issues.</p></div>",
            description="HVTracker interest capture is temporarily unavailable.",
        ),
        status_code=503,
    )


def _interest_thanks(title: str, message: str, repo: str | None = None) -> HTMLResponse:
    next_action = "<a class='button secondary' href='/alerts/'>Trust alerts</a>"
    if repo:
        agent = find_agent(repo)
        if agent:
            next_action = f"<a class='button secondary' href='/agents/{agent['slug']}'>Back to {escape(agent['name'])}</a>"
    return HTMLResponse(
        _marketing_page(
            title,
            "Request saved",
            "You are on the list.",
            f"<div class='card'><p class='ok'>{escape(message)}</p><p>For now this is a human-reviewed queue, not an automated onboarding flow. That is intentional: it helps validate which alerts, exports, and sponsor offers are worth building first.</p></div><div class='actions'><a class='button' href='/'>Open leaderboard</a>{next_action}</div>",
            description=message,
        )
    )


def _queue_table_html(title: str, rows: list[dict], empty_copy: str) -> str:
    if not rows:
        return (
            "<div class='card'>"
            f"<h2>{escape(title)}</h2>"
            f"<p>{escape(empty_copy)}</p>"
            "</div>"
        )
    items: list[str] = []
    for row in rows:
        payload = row.get("payload") or {}
        payload_html = escape(json.dumps(payload, indent=2, sort_keys=True))
        created = row.get("created_at")
        created_text = created.isoformat() if hasattr(created, "isoformat") else escape(str(created))
        contact = escape(row.get("contact") or "—")
        items.append(
            "<div class='card'>"
            f"<h2>{escape(title[:-1] if title.endswith('s') else title)} #{row.get('id', '—')}</h2>"
            f"<p><strong>Repo:</strong> {escape(row.get('repo') or '—')}</p>"
            f"<p><strong>Contact:</strong> {contact}</p>"
            f"<p><strong>Status:</strong> {escape(row.get('status') or '—')}</p>"
            f"<p><strong>Created:</strong> {created_text}</p>"
            f"<pre style='white-space:pre-wrap;word-break:break-word;background:#f6f2e9;border:1px solid #d7d0c3;padding:12px;font-size:12px;line-height:1.5;overflow:auto'>{payload_html}</pre>"
            "</div>"
        )
    return f"<h2>{escape(title)}</h2>" + "".join(items)


@app.get("/admin", response_class=HTMLResponse)
@app.get("/admin/", response_class=HTMLResponse, include_in_schema=False)
def admin_panel():
    submissions = db.list_queue("submissions", status="pending")
    corrections = db.list_queue("corrections", status="pending")
    queue_note = (
        "<div class='card'><p>This is a lightweight local review screen for the moderation queue. "
        "It is read-only for now: useful for opening the pending submissions and corrections that "
        "are already stored through the public forms.</p></div>"
    )
    if not db.enabled():
        queue_note = (
            "<div class='card'><p>The moderation queue needs <code>DATABASE_URL</code> to show real data. "
            "The route is now live, but this environment is running without the Postgres-backed queue.</p></div>"
        )
    body = (
        queue_note
        + _queue_table_html("Pending submissions", submissions, "No pending submissions.")
        + _queue_table_html("Pending corrections", corrections, "No pending corrections.")
    )
    return _marketing_page(
        "Admin queue — HVTracker",
        "Admin queue",
        "Review pending submissions and corrections.",
        body,
        description="HVTracker moderation queue for pending submissions and corrections.",
        path="/admin/",
    )


@app.get("/submit", response_class=HTMLResponse)
@app.get("/submit/", response_class=HTMLResponse, include_in_schema=False)
def submit_form():
    body = """
    <div class='card'>
      <h2>Submit an agent for listing</h2>
      <p>Submissions enter a moderation queue and appear after review. Keep it simple: repo, display name, optional category, and a contact email if you want follow-up.</p>
      <form method='post' action='/submit'>
        <label>GitHub repo (owner/name)
          <input name='repo' placeholder='owner/name' required>
        </label>
        <label>Display name
          <input name='name' placeholder='My Agent' required>
        </label>
        <label>Category (optional)
          <input name='category' placeholder='Coding Agents'>
        </label>
        <label>Contact email (optional)
          <input name='contact' type='email' placeholder='you@example.com'>
        </label>
        """ + HONEYPOT_HTML + """
        <div class='actions'>
          <button class='button' type='submit'>Submit for review</button>
        </div>
      </form>
    </div>
    """
    return _marketing_page(
        "Submit an agent — HVTracker",
        "Submission",
        "Suggest a project for the trust registry.",
        body,
        description="Submit an AI agent project for review and possible inclusion in the HVTracker trust registry.",
        path="/submit/",
    )


@app.post("/submit", response_class=HTMLResponse)
@app.post("/submit/", response_class=HTMLResponse, include_in_schema=False)
def submit_post(request: Request, repo: str = Form(...), name: str = Form(...),
                category: str = Form(""), contact: str = Form(""),
                website: str = Form("")):
    if website:
        return HTMLResponse(
            _marketing_page("Submission received — HVTracker", "Submission saved", "Your project is queued for review.",
                "<div class='card'><p class='ok'>Thanks. The submission is in the review queue.</p></div>",
                path="/submit/"))
    if _is_rate_limited(request):
        return HTMLResponse(
            _marketing_page("Too many requests — HVTracker", "Slow down", "Please wait before submitting again.",
                "<div class='card'><p>Too many submissions from this address. Try again in a few minutes.</p></div>",
                path="/submit/"), status_code=429)
    over = _check_field_lengths(repo=repo, name=name, category=category, contact=contact)
    if over:
        return HTMLResponse(
            _marketing_page("Input too long — HVTracker", "Submission", over,
                f"<div class='card'><p>{escape(over)}</p></div><div class='actions'><a class='button' href='/submit/'>Back to form</a></div>",
                path="/submit/"), status_code=400)
    if contact.strip() and not _EMAIL_RE.match(contact.strip()):
        return HTMLResponse(
            _marketing_page("Invalid email — HVTracker", "Submission", "Please provide a valid email address.",
                "<div class='card'><p>The contact email does not look valid.</p></div><div class='actions'><a class='button' href='/submit/'>Back to form</a></div>",
                path="/submit/"), status_code=400)
    repo = repo.strip().removeprefix("https://github.com/").strip("/")
    if repo.count("/") != 1:
        return HTMLResponse(
            _marketing_page(
                "Invalid repo — HVTracker",
                "Submission",
                "Use the GitHub owner/name format.",
                "<div class='card'><p>Please submit the repository as <code>owner/name</code>, for example <code>openai/codex</code>.</p></div><div class='actions'><a class='button' href='/submit/'>Back to form</a></div>",
                description="Submit an AI agent project for review and possible inclusion in the HVTracker trust registry.",
                path="/submit/",
            ),
            status_code=400,
        )
    if not db.enabled():
        return _interest_unavailable()
    db.add_submission(repo, {"name": name.strip(), "category": category.strip()}, contact.strip() or None)
    return HTMLResponse(
        _marketing_page(
            "Submission received — HVTracker",
            "Submission saved",
            "Your project is queued for review.",
            "<div class='card'><p class='ok'>Thanks. The submission is in the review queue.</p><p>HVTracker reviews repos manually before adding them so the registry stays evidence-backed and comparable.</p></div><div class='actions'><a class='button' href='/'>Open leaderboard</a><a class='button secondary' href='/submit/'>Submit another</a></div>",
            description="Your AI agent project submission has been received by HVTracker.",
            path="/submit/",
        )
    )


@app.get("/correct", response_class=HTMLResponse)
@app.get("/correct/", response_class=HTMLResponse, include_in_schema=False)
def correct_form():
    body = """
    <div class='card'>
      <h2>Corrections &amp; appeals — how it works</h2>
      <p>Any listed project (or anyone else) can dispute anything we publish: a trust score, a grade, a capability field, provenance drift, category, or listing status. This is the process, in full:</p>
      <ul>
        <li><strong>What counts as evidence.</strong> Scores are computed only from public, checkable signals, so disputes are settled the same way: point at the public source (repo, registry, manifest, commit) that contradicts what we show. "Our score should be higher" isn't reviewable; "your provenance check missed our PEP 740 attestation, here's the PyPI link" is.</li>
        <li><strong>What happens next.</strong> We re-check the signal against the source. If we're wrong, the fix ships as a data or detector change and takes effect on the next render — and if the error was in the methodology itself, that's said plainly in the <a href='/methodology/#changelog'>methodology changelog</a>. If we're right, you get the reasoning, not silence.</li>
        <li><strong>Turnaround.</strong> HVTracker is independently run; expect a response within a week, usually sooner.</li>
        <li><strong>Public appeal.</strong> Prefer a public record, or disagree with the outcome? Open a <a href='https://github.com/YugantM/hvtracker/issues' target='_blank' rel='noopener'>GitHub issue</a> — disputes and their resolutions stay visible there.</li>
      </ul>
      <p style='color:var(--muted);font-size:13px'>Scores are never changed by request — only by evidence. You can also <a href='/methodology/#verify-yourself'>verify any score yourself</a> against its signed credential.</p>
    </div>
    <div class='card'>
      <h2>Request a correction</h2>
      <p>Spotted wrong data on a listing? Send the repo, explain what is wrong, and optionally leave a contact email for follow-up.</p>
      <form method='post' action='/correct'>
        <label>GitHub repo (owner/name)
          <input name='repo' placeholder='owner/name' required>
        </label>
        <label>What's wrong?
          <textarea name='message' rows='5' placeholder='Describe the issue...' required></textarea>
        </label>
        <label>Contact email (optional)
          <input name='contact' type='email' placeholder='you@example.com'>
        </label>
        """ + HONEYPOT_HTML + """
        <div class='actions'>
          <button class='button' type='submit'>Send correction</button>
        </div>
      </form>
    </div>
    """
    return _marketing_page(
        "Request a correction — HVTracker",
        "Corrections",
        "Flag data that needs review.",
        body,
        description="Request a correction to an HVTracker project listing.",
        path="/correct/",
    )


@app.post("/correct", response_class=HTMLResponse)
@app.post("/correct/", response_class=HTMLResponse, include_in_schema=False)
def correct_post(request: Request, repo: str = Form(...), message: str = Form(...),
                 contact: str = Form(""), website: str = Form("")):
    if website:
        return HTMLResponse(
            _marketing_page("Correction received — HVTracker", "Correction saved", "The correction request is queued for review.",
                "<div class='card'><p class='ok'>Thanks.</p></div>", path="/correct/"))
    if _is_rate_limited(request):
        return HTMLResponse(
            _marketing_page("Too many requests — HVTracker", "Slow down", "Please wait before submitting again.",
                "<div class='card'><p>Too many submissions from this address. Try again in a few minutes.</p></div>",
                path="/correct/"), status_code=429)
    over = _check_field_lengths(repo=repo, message=message, contact=contact)
    if over:
        return HTMLResponse(
            _marketing_page("Input too long — HVTracker", "Correction", over,
                f"<div class='card'><p>{escape(over)}</p></div>", path="/correct/"), status_code=400)
    if not db.enabled():
        return _interest_unavailable()
    repo = repo.strip().removeprefix("https://github.com/").strip("/")
    db.add_correction(repo, {"message": message.strip()}, contact.strip() or None)
    return HTMLResponse(
        _marketing_page(
            "Correction received — HVTracker",
            "Correction saved",
            "The correction request is queued for review.",
            "<div class='card'><p class='ok'>Thanks. The correction request is now in the review queue.</p><p>HVTracker reviews correction requests manually so trust signals stay accurate and auditable.</p></div><div class='actions'><a class='button' href='/'>Open leaderboard</a><a class='button secondary' href='/correct/'>Send another correction</a></div>",
            description="Your HVTracker correction request has been received.",
            path="/correct/",
        )
    )


@app.get("/alerts", response_class=HTMLResponse)
@app.get("/alerts/", response_class=HTMLResponse, include_in_schema=False)
def alerts_page(request: Request):
    # Was a waitlist (kind=alerts in interest_signups) until accounts and
    # tracking were real; the captured rows are kept.
    if _auth.current_user(request):
        cta = "<a class='button' href='/account/'>Open your tracked projects</a>"
    else:
        cta = "<a class='button' href='/login?next=/account/'>Sign in to start tracking</a>"
    email_html = ("\n        <p>Prefer email? Turn on alert emails on your account page: a daily or weekly "
                  "digest of grade flips, HVTrust moves of 3+ points, lost provenance and supply-chain "
                  "warnings, sent only when something changed.</p>") if alerts.enabled() else ""
    body = f"""
    <div class='grid'>
      <div class='card'>
        <span class='pill'>What you hear about</span>
        <h2 style='margin-top:10px'>The changes on a tool's public timeline</h2>
        <ul>
          <li>Trust grade flips, such as B &rarr; C</li>
          <li>HVTrust moves of 3 points or more, and rank moves of 10 places or more</li>
          <li>Package provenance lost, or a provenance-drift warning raised</li>
          <li>Eligibility warnings</li>
        </ul>
        <p style='margin-top:12px'>Moves caused by a methodology update are not reported as changes.</p>
      </div>
      <div class='card'>
        <span class='pill'>How it works</span>
        <h2 style='margin-top:10px'>Track the tools in your stack</h2>
        <p>Sign in, then choose <em>Track</em> on any agent, MCP server or skill page, or paste your dependency file into <a href='/scan/'>Scan your stack</a> and choose <em>Watch all</em>.</p>
        <p>Changes show up in the bell at the top of every page and on your account page. A tool that leaves the registry stays on your list, marked as no longer listed.</p>{email_html}
      </div>
    </div>
    <div class='actions'>
      {cta}
      <a class='button secondary' href='/scan/'>Scan your stack</a>
      <a class='button secondary' href='/changes/'>All recent changes</a>
    </div>
    """
    return HTMLResponse(_marketing_page(
        "Trust alerts — HVTracker", "Alerts", "Know when a tool you depend on changes.", body,
        description="Track AI agents, MCP servers and skills on HVTracker and get notified when their trust grade, HVTrust score or provenance changes.",
        path="/alerts/",
        lede="Free. The registry stays fully public; an account only adds tracking and notifications."))


@app.post("/alerts", include_in_schema=False)
@app.post("/alerts/", include_in_schema=False)
def alerts_post():
    # The old waitlist form, from a page still in someone's cache.
    return RedirectResponse("/alerts/", status_code=303)


def _track(request: Request, slug: str, status_code: int):
    """Signed in: add the project to the account's tracked projects and land on
    /account. Signed out: sign in, then come back to GET /track/<slug>/."""
    agent = find_agent_by_slug(slug)
    if not agent:
        return HTMLResponse("<p>Agent not found.</p>", status_code=404)
    user = _auth.current_user(request)
    if not user:
        return RedirectResponse(f"/login?next=/track/{agent['slug']}/", status_code=status_code)
    db.add_watch(user["id"], agent["slug"])
    return RedirectResponse(f"/account/?tracked={agent['slug']}", status_code=status_code)


@app.get("/track/{slug}", include_in_schema=False)
@app.get("/track/{slug}/", include_in_schema=False)
def track_agent(request: Request, slug: str):
    """Where sign-in returns to (next=/track/<slug>/), plus old links and caches.

    Profiles no longer link here: a crawler that ignores robots.txt walked every
    profile's <a href="/track/<slug>/"> into /login (~2,500 GA4 /login views on
    17-18 Oct 2026), so "Get email alerts for X" POSTs instead.
    """
    return _track(request, slug, 302)


@app.post("/track/{slug}", include_in_schema=False)
@app.post("/track/{slug}/", include_in_schema=False)
def track_agent_post(request: Request, slug: str):
    # The profile's "Get email alerts for X" button (and the old waitlist form,
    # from a page still in someone's cache).
    return _track(request, slug, 303)


@app.get("/sponsor", response_class=HTMLResponse)
@app.get("/sponsor/", response_class=HTMLResponse, include_in_schema=False)
def sponsor_page():
    body = """
    <div class='grid'>
      <div class='card'>
        <span class='pill'>Sponsor options</span>
        <h2 style='margin-top:10px'>Low-noise ways to partner</h2>
        <ul>
          <li>Category or report sponsorships</li>
          <li>Launch-week or research sponsorships</li>
        </ul>
      </div>
      <div class='card'>
        <span class='pill'>Good fit</span>
        <h2 style='margin-top:10px'>Who this is for</h2>
        <p>Agent infrastructure, observability, security, evaluation, and devtools companies that want to reach technical buyers without generic ad inventory.</p>
      </div>
    </div>
    <div class='card'>
      <h2>Start a sponsor conversation</h2>
      <p>Keep it short. I mostly need to know who you are, what audience you want, and whether you want sponsorship, research, or a custom data relationship.</p>
      <form method='post' action='/sponsor'>
        <label>Name
          <input type='text' name='name' placeholder='Your name' required>
        </label>
        <label>Company
          <input type='text' name='company' placeholder='Company name' required>
        </label>
        <label>Work email
          <input type='email' name='email' placeholder='you@company.com' required>
        </label>
        <label>What are you interested in?
          <textarea name='message' placeholder='Example: we sell agent observability and want to sponsor a category roundup or trust report aimed at platform and security teams.' required></textarea>
        </label>
        """ + HONEYPOT_HTML + """
        <div class='actions'>
          <button class='button' type='submit'>Send sponsor interest</button>
        </div>
      </form>
    </div>
    """
    return HTMLResponse(_marketing_page("Sponsor HVTracker", "Commercial", "Reach teams evaluating AI agents with context, not banner spam.", body, description="Sponsor HVTracker to reach teams evaluating AI agents with trust and comparison context.", path="/sponsor/"))


@app.post("/sponsor", response_class=HTMLResponse)
@app.post("/sponsor/", response_class=HTMLResponse, include_in_schema=False)
def sponsor_post(request: Request, name: str = Form(...), company: str = Form(...),
                 email: str = Form(...), message: str = Form(...), website: str = Form("")):
    if website:
        return HTMLResponse(
            _marketing_page("Sponsor HVTracker", "Thanks", "Thanks. I saved your sponsorship inquiry.",
                "<div class='card'><p class='ok'>Thanks.</p></div>", path="/sponsor/"))
    if _is_rate_limited(request):
        return HTMLResponse(
            _marketing_page("Too many requests — HVTracker", "Slow down", "Please wait before submitting again.",
                "<div class='card'><p>Too many submissions from this address. Try again in a few minutes.</p></div>",
                path="/sponsor/"), status_code=429)
    over = _check_field_lengths(name=name, email=email, message=message)
    if over:
        return HTMLResponse(
            _marketing_page("Input too long — HVTracker", "Sponsor", over,
                f"<div class='card'><p>{escape(over)}</p></div>", path="/sponsor/"), status_code=400)
    if not _EMAIL_RE.match(email.strip()):
        return HTMLResponse(
            _marketing_page("Invalid email — HVTracker", "Sponsor", "Please provide a valid email address.",
                "<div class='card'><p>That email does not look valid.</p></div>", path="/sponsor/"), status_code=400)
    if not db.enabled():
        return _interest_unavailable()
    db.add_interest_signup(
        "sponsor",
        email.strip().lower(),
        None,
        {
            "name": name.strip(),
            "company": company.strip(),
            "message": message.strip(),
            "source": "sponsor-page",
        },
    )
    return _interest_thanks("Sponsor HVTracker", "Thanks. I saved your sponsorship inquiry.")


@app.get("/data-api", response_class=HTMLResponse)
@app.get("/data-api/", response_class=HTMLResponse, include_in_schema=False)
def data_api_page():
    _now = datetime.now(timezone.utc)
    quarter = f"{_now.year}-Q{(_now.month - 1) // 3 + 1}"
    body = """
    <div class='grid'>
      <div class='card'>
        <span class='pill'>Available now</span>
        <h2 style='margin-top:10px'>Public layer</h2>
        <ul>
          <li>Free leaderboard browsing</li>
          <li>Public JSON snapshot</li>
          <li>Open methodology and specs</li>
          <li>Read-only REST API (v1)</li>
        </ul>
      </div>
      <div class='card'>
        <span class='pill'>Planned paid layer</span>
        <h2 style='margin-top:10px'>Commercial access</h2>
        <ul>
          <li>Higher-rate data access</li>
          <li>Historical exports and change feeds</li>
          <li>Watchlists, alerts, and shared team usage</li>
        </ul>
      </div>
    </div>
    <div class='card'>
      <h2>REST API v1</h2>
      <p>Two read-only endpoints are available now, no auth required. CORS-enabled for browser use.</p>
      <div class='card' style='margin-top:12px'>
        <h2 style='font-family:var(--font-mono);font-size:13px'><code>GET /api/v1/agents</code></h2>
        <p>Full agent leaderboard with trust scores, evidence grades, and metadata.</p>
        <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>curl -s https://hvtracker.net/api/v1/agents | python -m json.tool | head</code></pre>
      </div>
      <div class='card' style='margin-top:12px'>
        <h2 style='font-family:var(--font-mono);font-size:13px'><code>GET /api/v1/graph</code></h2>
        <p>Dependency and ecosystem graph data for all tracked agents.</p>
        <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>curl -s https://hvtracker.net/api/v1/graph | python -m json.tool | head</code></pre>
      </div>
      <div class='card' style='margin-top:12px'>
        <h2 style='font-family:var(--font-mono);font-size:13px'><code>POST /api/v1/scan</code></h2>
        <p>Bulk trust check for a whole dependency set — paste a requirements.txt, package.json, or MCP config and get a verdict per item. Powers the <a href='/scan/'>Scan your stack</a> tool.</p>
        <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>curl -s https://hvtracker.net/api/v1/scan -H 'Content-Type: application/json' \
  -d '{"input": "langchain\ncrewai\nautogen"}' | python -m json.tool</code></pre>
      </div>
      <div class='card' style='margin-top:12px'>
        <h2 style='font-family:var(--font-mono);font-size:13px'><code>GET /api/v1/mcp/verify?server=&lt;repo|url|npm|pypi|purl|registry-name&gt;</code></h2>
        <p>Pre-connect trust verdict for an MCP server — a signed attestation when the server is tracked, an explicit "not tracked, unverified" verdict when it isn't. A package URL (<code>pkg:npm/…</code>, <code>pkg:pypi/…</code>, <code>pkg:cargo/…</code>) or an MCP registry name resolves through the <a href='/data/identifiers.json'>identifier index</a>, which lists only identifiers our evidence ties to a listing. Spec: <a href='/spec/mcp-server-trust/v0.1'>MCP Server Trust v0.1</a>.</p>
        <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>curl -s 'https://hvtracker.net/api/v1/mcp/verify?server=deepset-ai/haystack' | python -m json.tool</code></pre>
      </div>
      <div class='card' style='margin-top:12px'>
        <h2 style='font-family:var(--font-mono);font-size:13px'><code>GET /data/agents/&lt;slug&gt;.json</code> · <code>GET /data/latest.json</code></h2>
        <p>Per-agent record with the full signal set and an <strong>Ed25519-signed <code>trust_credential</code></strong> (verify it offline — see <a href='/methodology/#verify-yourself'>verify a score yourself</a>), and the full-registry snapshot. Machine discovery starts at <a href='/.well-known/hvtracker.json'><code>/.well-known/hvtracker.json</code></a>.</p>
      </div>
      <div class='card' style='margin-top:12px'>
        <h2 style='font-family:var(--font-mono);font-size:13px'><code>GET /api/v1/agents/&lt;slug&gt;/history</code></h2>
        <p>An agent's public trust fields over the <strong>last 90 days</strong>, one entry per daily snapshot (rank, HVTrust, grades, provenance, Scorecard, downloads, stars). Beyond 90 days, use the <a href='#'>quarterly dataset export</a> below; deeper continuous history is reserved for a future tier.</p>
        <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>curl -s https://hvtracker.net/api/v1/agents/haystack/history | python -m json.tool | head</code></pre>
      </div>
      <p style='margin-top:12px;color:var(--muted);font-size:13px'><strong>Stability promise:</strong> within v1, fields are add-only — existing fields are never renamed or removed; breaking changes ship under a new path version. Data license: <strong>CC BY 4.0</strong>, attribution "HVTracker (hvtracker.net)". Auth and rate quotas for a paid tier are intentionally out of scope for now.</p>
    </div>
    <div class='card'>
      <h2>Quarterly dataset export</h2>
      <p>A citable, versioned snapshot of every public field for research and press — refreshed through the quarter, frozen when the quarter ends. <strong>CC BY 4.0</strong>; attribution: "HVTracker (hvtracker.net)".</p>
      <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>curl -sO https://hvtracker.net/data/exports/hvtrust-QUARTER_LABEL.json.gz
curl -sO https://hvtracker.net/data/exports/hvtrust-QUARTER_LABEL.csv</code></pre>
      <p style='margin-top:12px;color:var(--muted);font-size:13px'>Filename pattern: <code>hvtrust-&lt;year&gt;-Q&lt;n&gt;</code>. Past quarters stay at their end-of-quarter state.</p>
      <p style='margin-top:8px;color:var(--muted);font-size:13px'>Ended quarters are archived on Zenodo with a DOI, so a citation always points at a fixed copy: 2026-Q3 is <a href='https://doi.org/10.5281/zenodo.23084548'>doi:10.5281/zenodo.23084548</a> (every quarter: <a href='https://doi.org/10.5281/zenodo.23084547'>10.5281/zenodo.23084547</a>). Agents and skills are ranked on separate boards, each from 1; the <code>listing_class</code> column says which (the archived 2026-Q3 copy adds it).</p>
    </div>
    <div class='card'>
      <h2>MCP registry feed</h2>
      <p>The official MCP Registry entries for every server we score, served as a v0.1 registry with HVTrust in <code>_meta</code>, plus Grade A, B and C allowlists for GitHub Copilot's <em>Registry only</em> policy. <a href='/registry/'>Base URLs and setup →</a></p>
    </div>
    <div class='card'>
      <h2>MCP server — trust layer for agents</h2>
      <p>Add HVTracker as a <a href='https://modelcontextprotocol.io'>Model Context Protocol</a> server so a coding agent can check supply-chain trust <em>before</em> installing a dependency or connecting to an MCP server. Streamable HTTP, no auth, no install.</p>
      <pre style='background:var(--paper);border:1px solid var(--line);padding:12px;overflow-x:auto;font:13px var(--font-mono);border-radius:6px;margin-top:8px'><code>{
  "mcpServers": {
    "hvtracker": { "url": "https://hvtracker.net/mcp" }
  }
}</code></pre>
      <p style='margin-top:12px;color:var(--muted);font-size:13px'>Tools: <code>verify_mcp_server</code> (pre-connect "Safe Browsing for MCP" verdict), <code>check_agent_trust</code> (incl. runtime capabilities + signed-credential URL), <code>scan_stack</code> (verdict for a whole requirements.txt / package.json / mcp config), <code>compare_agents</code>, <code>get_leaderboard</code>, <code>list_categories</code>, <code>get_agent_history</code> (90-day trust trend), <code>search_agents</code>.</p>
    </div>
    <div class='card'>
      <h2>Early pricing stub</h2>
      <div class='grid'>
        <div class='card'>
          <span class='pill'>Free</span>
          <h2 style='margin-top:10px'>Public registry</h2>
          <p>Leaderboard, agent profiles, compare pages, methodology, and the public JSON layer.</p>
        </div>
        <div class='card'>
          <span class='pill'>Planned from $29/mo</span>
          <h2 style='margin-top:10px'>Builder</h2>
          <p>Email alerts, small watchlists, expanded exports, and early access to new comparison workflows.</p>
        </div>
        <div class='card'>
          <span class='pill'>Planned from $149/mo</span>
          <h2 style='margin-top:10px'>Team</h2>
          <p>Shared watchlists, deeper history, bulk exports, and commercial usage support.</p>
        </div>
      </div>
    </div>
    <div class='card'>
      <h2>Request API or data access</h2>
      <p>If you want to use HVTracker data commercially, tell me how. The point of this page is to validate which export shapes and usage rights are worth productizing first.</p>
      <form method='post' action='/data-api'>
        <label>Work email
          <input type='email' name='email' placeholder='you@company.com' required>
        </label>
        <label>Company or team
          <input type='text' name='company' placeholder='Company or project name'>
        </label>
        <label>What do you need?
          <textarea name='message' placeholder='Example: daily JSON export for internal evaluations, historical trust changes, compare data for content, or commercial redistribution rights.' required></textarea>
        </label>
        """ + HONEYPOT_HTML + """
        <div class='actions'>
          <button class='button' type='submit'>Request access</button>
          <a class='button secondary' href='/data.json'>Open public data</a>
        </div>
      </form>
    </div>
    """
    body = body.replace("QUARTER_LABEL", quarter)
    return HTMLResponse(_marketing_page("Data API and pricing — HVTracker", "Data", "Public data today. Commercial access next.", body, description="Explore the HVTracker data API, exports, and future commercial access options.", path="/data-api/"))


@app.post("/data-api", response_class=HTMLResponse)
@app.post("/data-api/", response_class=HTMLResponse, include_in_schema=False)
def data_api_post(request: Request, email: str = Form(...), company: str = Form(""),
                  message: str = Form(...), website: str = Form("")):
    if website:
        return HTMLResponse(
            _marketing_page("Data API — HVTracker", "Thanks", "Thanks. I saved your request.",
                "<div class='card'><p class='ok'>Thanks.</p></div>", path="/data-api/"))
    if _is_rate_limited(request):
        return HTMLResponse(
            _marketing_page("Too many requests — HVTracker", "Slow down", "Please wait before submitting again.",
                "<div class='card'><p>Too many submissions from this address. Try again in a few minutes.</p></div>",
                path="/data-api/"), status_code=429)
    over = _check_field_lengths(email=email, message=message)
    if over:
        return HTMLResponse(
            _marketing_page("Input too long — HVTracker", "Data API", over,
                f"<div class='card'><p>{escape(over)}</p></div>", path="/data-api/"), status_code=400)
    if not _EMAIL_RE.match(email.strip()):
        return HTMLResponse(
            _marketing_page("Invalid email — HVTracker", "Data API", "Please provide a valid email address.",
                "<div class='card'><p>That email does not look valid.</p></div>", path="/data-api/"), status_code=400)
    if not db.enabled():
        return _interest_unavailable()
    db.add_interest_signup(
        "api-access",
        email.strip().lower(),
        None,
        {
            "company": company.strip(),
            "message": message.strip(),
            "source": "data-api-page",
        },
    )
    return _interest_thanks("Data API and pricing — HVTracker", "Thanks. I saved your API and data access request.")


# ---- Scheduler + startup --------------------------------------------------

def _pull_scorecard_cache() -> bool:
    """Refresh the baked scorecard cache from the `data` branch so live
    full/batch refreshes apply the latest OSSF scan rather than the image's
    deploy-time snapshot. Best-effort: on any failure the existing cache is
    left untouched. Returns True if the local cache was updated."""
    import urllib.request
    try:
        with urllib.request.urlopen(SCORECARD_CACHE_URL, timeout=20) as resp:
            raw = resp.read()
        data = json.loads(raw)
        agents = data.get("agents") if isinstance(data, dict) else None
        if not isinstance(agents, dict) or not agents:
            print("[scorecard] data-branch cache has no agents — keeping baked cache")
            return False
        tmp = SCORECARD_CACHE_PATH + ".tmp"
        with open(tmp, "wb") as f:
            f.write(raw)
        os.replace(tmp, SCORECARD_CACHE_PATH)
        print(f"[scorecard] pulled data-branch cache: {len(agents)} repos "
              f"(scanned {data.get('scanned_at', 'unknown')})")
        return True
    except Exception as e:
        print(f"[scorecard] data-branch cache pull failed ({e}) — keeping baked cache")
        return False


def _pull_registry_snapshot() -> bool:
    """Refresh the MCP registry snapshot from the `data` branch before a
    render. Best-effort like the scorecard pull: a failure keeps the current
    file, and an empty or malformed one is never written over it."""
    import urllib.request
    try:
        with urllib.request.urlopen(REGISTRY_SNAPSHOT_URL, timeout=20) as resp:
            raw = resp.read()
        servers = json.loads(raw).get("servers")
        if not isinstance(servers, list) or not servers:
            print("[registry] data-branch snapshot has no servers — keeping the current one")
            return False
        tmp = REGISTRY_SNAPSHOT_PATH + ".tmp"
        with open(tmp, "wb") as f:
            f.write(raw)
        os.replace(tmp, REGISTRY_SNAPSHOT_PATH)
        print(f"[registry] pulled data-branch snapshot: {len(servers)} entries")
        return True
    except Exception as e:
        print(f"[registry] data-branch snapshot pull failed ({e}) — keeping the current one")
        return False


def _refresh(mode: str) -> bool:
    """Run a refresh cycle. Returns True on success, False on failure.

    Runs fetch_and_build as a SUBPROCESS rather than in-process: the render's
    working set (history snapshots, graph build, ~500 rendered pages, Pillow
    OG images) is freed back to the OS when the child exits. In-process, that
    heap never shrank, ratcheting the always-on server to ~0.5GB average RAM
    — the dominant Railway cost. Child inherits env (OUTPUT_DIR, tokens) and
    stdout/stderr, so volume paths and Railway logs behave exactly as before.
    """
    import subprocess

    import fetch_and_build
    try:
        # Pull the latest OSSF scan from the data branch before every refresh —
        # including render-only, which now re-applies the cache to all agents so
        # fresh scores reach the live site on each deploy/restart, not just on a
        # GitHub-signal refresh cycle.
        _pull_scorecard_cache()
        _pull_registry_snapshot()
        subprocess.run(
            [sys.executable, os.path.join(BASE_DIR, "fetch_and_build.py"),
             *fetch_and_build.refresh_argv(mode)],
            check=True,
            cwd=BASE_DIR,
            # Stream the child's render logs live into the service log (piped
            # stdout is otherwise fully buffered until exit — useless for
            # watching a deploy's first render on Railway). HVT_CACHE_DIR turns
            # on cache.py's on-volume API cache for the child (replaced Redis).
            env={**os.environ, "PYTHONUNBUFFERED": "1",
                 "HVT_CACHE_DIR": os.environ.get("HVT_CACHE_DIR") or _refresh_cache_dir()},
        )
        return True
    except Exception as e:  # never let a build error kill the scheduler thread
        import traceback
        print(f"[scheduler] refresh ({mode}) failed: {e}")
        traceback.print_exc()
        return False


def _api_cache_entries() -> int | None:
    """Files in the refresh subprocess's API cache (None before the first
    cached fetch) — the only way to see the cache working without SSH."""
    try:
        return len(os.listdir(os.environ.get("HVT_CACHE_DIR") or _refresh_cache_dir()))
    except OSError:
        return None


def _refresh_cache_dir() -> str:
    """Where the refresh subprocess keeps its API-response cache: on the
    volume (survives restarts and deploys), under a path the middleware
    refuses to serve."""
    return os.path.join(OUTPUT_DIR, ".cache", "api")


def _compute_render_fingerprint() -> str:
    digest = hashlib.sha256()
    tracked_paths = [
        os.path.join(BASE_DIR, "fetch_and_build.py"),
        os.path.join(BASE_DIR, "template.html"),
        os.path.join(BASE_DIR, "compare", "index.html"),
        os.path.join(BASE_DIR, "og-v2.png"),
        os.path.join(BASE_DIR, "agents.json"),
    ]
    _css_path = os.path.join(BASE_DIR, "static", "site.css")
    if os.path.isfile(_css_path):
        tracked_paths.append(_css_path)
    # render_state.json lives on the volume in production (excluded from Docker
    # image by .dockerignore).  Include it when present so adding an agent
    # triggers a re-render; skip gracefully when the file doesn't exist yet.
    _rs_path = os.path.join(BASE_DIR, "data", "render_state.json")
    if os.path.isfile(_rs_path):
        tracked_paths.append(_rs_path)
    templates_dir = os.path.join(BASE_DIR, "templates")
    for name in sorted(os.listdir(templates_dir)):
        if name.endswith((".html", ".j2")):
            tracked_paths.append(os.path.join(templates_dir, name))
    blog_static_dir = os.path.join(BASE_DIR, "blog_static")
    if os.path.isdir(blog_static_dir):
        for name in sorted(os.listdir(blog_static_dir)):
            idx = os.path.join(blog_static_dir, name, "index.html")
            if os.path.isfile(idx):
                tracked_paths.append(idx)
    for path in tracked_paths:
        with open(path, "rb") as f:
            digest.update(path.removeprefix(BASE_DIR).encode("utf-8"))
            digest.update(b"\0")
            digest.update(f.read())
            digest.update(b"\0")
    return digest.hexdigest()


def _read_render_fingerprint() -> str | None:
    try:
        with open(RENDER_FINGERPRINT_PATH, encoding="utf-8") as f:
            return f.read().strip() or None
    except OSError:
        return None


def _write_render_fingerprint(fingerprint: str) -> None:
    with open(RENDER_FINGERPRINT_PATH, "w", encoding="utf-8") as f:
        f.write(fingerprint)


def _refresh_and_record(mode: str, fingerprint: str, trigger: str = "internal") -> bool:
    if not _refresh_lock.acquire(blocking=False):
        print(f"[scheduler] skipped refresh ({mode}) from {trigger}: another refresh is already running")
        _mark_refresh_status(
            in_progress=True,
            last_mode=mode,
            last_trigger=trigger,
            last_skipped_at=_utc_now_iso(),
            last_error="refresh skipped because another refresh is already running",
        )
        return False
    started_at = _utc_now_iso()
    _mark_refresh_status(
        in_progress=True,
        last_mode=mode,
        last_trigger=trigger,
        last_started_at=started_at,
        last_succeeded=None,
        last_error=None,
    )
    try:
        ok = _refresh(mode)
        completed_at = _utc_now_iso()
        if ok:
            _write_render_fingerprint(fingerprint)
        _mark_refresh_status(
            in_progress=False,
            last_mode=mode,
            last_trigger=trigger,
            last_completed_at=completed_at,
            last_succeeded=ok,
            last_error=None if ok else f"refresh failed in mode {mode}",
        )
        return ok
    finally:
        _refresh_lock.release()


def _seed_history_into_volume() -> int:
    """Copy daily history snapshots baked into the image into the volume if
    they're missing. Without prior days the leaderboard has no rank deltas,
    sparklines, or movers. Existing files on the volume are never overwritten,
    so today's freshly-written snapshot is preserved. Returns the number of
    files copied."""
    import shutil
    seed = os.path.join(BASE_DIR, "seed", "history")
    if not os.path.isdir(seed):
        return 0
    dest = os.path.join(OUTPUT_DIR, "output", "history")
    os.makedirs(dest, exist_ok=True)
    copied = 0
    for fn in sorted(os.listdir(seed)):
        if not fn.endswith(".json"):
            continue
        dst = os.path.join(dest, fn)
        if not os.path.exists(dst):
            shutil.copy2(os.path.join(seed, fn), dst)
            copied += 1
    if copied:
        print(f"[startup] seeded {copied} history snapshot(s) into volume")
    return copied


def _has_missing_commit_rows() -> bool:
    """Detect rows whose 4-week commit count needs (re)fetching, so startup
    kicks the repair-commits refresh. Must match fetch_and_build's
    _commit_count_suspect: None = never fetched; 0 alongside a push inside
    the window is near-contradictory (a push implies commits) and covers
    rows written before the None-seed fix plus GraphQL-prefetch misses."""
    # Inlined (not imported) to keep fetch_and_build out of the web process.
    def suspect(a: dict) -> bool:
        if not a.get("repo"):
            return False
        commits = a.get("weekly_commits")
        if commits is None:
            return True
        days = a.get("days_ago")
        return commits == 0 and days is not None and days <= 28

    try:
        with open(DATA_PATH, encoding="utf-8") as f:
            data = json.load(f)
        return any(suspect(a) for a in data.get("agents", []))
    except (OSError, json.JSONDecodeError, TypeError):
        return False


def _has_pending_signal_rows() -> bool:
    """Detect provisional rows that still need their first signal refresh."""
    try:
        with open(DATA_PATH, encoding="utf-8") as f:
            data = json.load(f)
        return any(a.get("pending_signals") for a in data.get("agents", []))
    except (OSError, json.JSONDecodeError, TypeError):
        return False


def _refresh_verify_feed():
    """Daily: re-verify the provisional (open-lookup) repos in the public feed
    so their verdicts stay fresh. Curated repos are refreshed by the main cron;
    only the open-lookup ones need this. Gentle on the GitHub budget.

    Uses verify_log.refresh, NOT .record: this is our own revalidation, so it
    must not count as a check or restamp `checked_at`. Recording it as a check
    pinned the same provisional repos to the top of the public feed every night
    with identical timestamps and inflated their `checks` counters."""
    if not db.enabled():
        return
    import open_lookup
    token = os.environ.get("GITHUB_TOKEN", "")
    refreshed = 0
    for repo in db.verify_check_targets(limit=200):
        try:
            v = open_lookup.evaluate_open(repo, repo, token)
            if v.get("eligibility") == "ok":
                # Key on the row we selected, not v["resolved"]: a transferred
                # repo resolves to its new name, and updating that instead
                # would no-op and leave this row permanently stalest-first.
                verify_log.refresh(repo, None, v.get("grade"),
                                   v.get("trusted"), v.get("stars"))
                refreshed += 1
        except Exception:
            pass
        time.sleep(0.6)  # ~100/min, well within the authenticated GitHub budget
        if refreshed >= 150:
            break
    print(f"[scheduler] verify feed refresh complete: {refreshed} repo(s)")


def _signals_cron(minutes: int) -> dict:
    """Cron fields for the signals refresh.

    The cron `minute` field only spans 0-59, so `minute="*/360"` raises
    ValueError at add_job — at startup, that takes the whole app down.
    Sub-hour cadences keep the minute step; hourly and slower ones step the
    hour and fire at :30, clear of the batch's :00 slot.
    """
    if minutes < 60:
        return {"minute": f"*/{minutes}"}
    return {"hour": f"*/{max(1, minutes // 60)}", "minute": 30}


def _start_scheduler() -> None:
    """Start the refresh scheduler.

    Called first in _startup, before any boot-time refresh thread, so nothing
    can race or skip it. On the 22 Sep 12:52 UTC boot the scheduler never
    started and the site served day-old data as "ok"; a failure is now
    recorded and surfaced in /healthz instead of passing silently.
    """
    global _scheduler, _scheduler_error
    if os.environ.get("DISABLE_SCHEDULER") == "1" or _scheduler is not None:
        return
    try:
        scheduler = BackgroundScheduler(timezone="UTC")
        # One job, one whole-site render every 4 h: the stalest sixth of the
        # board gets a full fetch (so every row daily) and every other row a
        # GitHub-signal refresh (stars/commits, ≤4 h old). Renders are most of
        # the Railway memory bill; a separate 6-hourly signals job used to add
        # four more a day.
        scheduler.add_job(
            lambda: _refresh_and_record("auto", _compute_render_fingerprint(), "scheduler"),
            "cron",
            hour="*/4",
            id="refresh",
            max_instances=1,
            coalesce=True,
        )
        # The standalone signals refresh stays available as an opt-in
        # (SIGNALS_REFRESH_MIN) without a deploy, but is off by default.
        signals_min = None
        if os.environ.get("SIGNALS_REFRESH_MIN"):
            signals_min = max(5, int(os.environ["SIGNALS_REFRESH_MIN"]))
            scheduler.add_job(
                lambda: _refresh_and_record("signals", _compute_render_fingerprint(), "scheduler"),
                "cron",
                id="signals-refresh",
                max_instances=1,
                coalesce=True,
                **_signals_cron(signals_min),
            )
        # Persist the in-memory machine-usage rollup. Runs regardless of
        # db.enabled() — usage.flush falls back to the volume JSON — and is
        # cheap: one upsert per (hour, channel) touched since the last run.
        scheduler.add_job(
            usage.flush,
            "cron",
            minute="*",
            id="usage-flush",
            max_instances=1,
            coalesce=True,
        )
        if db.enabled():
            scheduler.add_job(
                _refresh_verify_feed,
                "cron",
                hour=4,
                id="verify-feed-refresh",
                max_instances=1,
                coalesce=True,
            )
        scheduler.start()
        _scheduler = scheduler
        _scheduler_error = None
        extra = f", extra signals every {signals_min}m" if signals_min else ""
        print(f"[startup] scheduler started (batch + signals every 4h{extra})", flush=True)
    except Exception as e:  # never let a scheduler problem take the site down
        import traceback
        _scheduler_error = f"{type(e).__name__}: {e}"
        print(f"[startup] SCHEDULER FAILED TO START — no refreshes will run: {_scheduler_error}",
              flush=True)
        traceback.print_exc()


def _process_rss_mb() -> float | None:
    """Resident memory of the web process (Linux /proc), for /healthz.

    Memory is most of the Railway bill; this makes the web process's own share
    visible without SSH (render subprocesses are billed on top of it).
    """
    try:
        with open("/proc/self/status", encoding="ascii") as f:
            for line in f:
                if line.startswith("VmRSS:"):
                    return round(int(line.split()[1]) / 1024, 1)
    except (OSError, ValueError, IndexError):
        pass
    return None


def _scheduled_jobs() -> dict[str, str | None]:
    """Next run time per scheduled job, for /healthz."""
    if _scheduler is None:
        return {}
    out: dict[str, str | None] = {}
    for job in _scheduler.get_jobs():
        nrt = getattr(job, "next_run_time", None)
        out[job.id] = nrt.isoformat() if nrt else None
    return out


def _kick_boot_refresh(mode: str, fingerprint: str) -> None:
    """Start a boot-time refresh in the background.

    HVT_BOOT_REFRESH=0 skips it; the test suite sets that in conftest. Under
    pytest these threads ran real render subprocesses and scorecard pulls that
    outlived the tests: they rewrote the repo's scorecard-cache.json, raced
    module reloads ("release unlocked lock") and swallowed pytest's summary
    line. Tests that exercise the branch choice fake the thread and opt back in.
    """
    if os.environ.get("HVT_BOOT_REFRESH", "1") == "0":
        print(f"[startup] boot refresh ({mode}) skipped: HVT_BOOT_REFRESH=0", flush=True)
        return
    threading.Thread(target=_refresh_and_record, args=(mode, fingerprint, "startup"),
                     daemon=True).start()


def _startup():
    _start_scheduler()
    seeded = _seed_history_into_volume()
    fingerprint = _compute_render_fingerprint()
    stored_fingerprint = _read_render_fingerprint()
    db.init_schema()
    # Sync agents table from agents.json on every startup.  agents.json is
    # the source of truth; the table is rebuilt from it.  upsert_agent is
    # idempotent and only `agents.json` writes to this table (submissions
    # and corrections go to separate tables), so this is safe to run
    # unconditionally — it keeps category/legacy/license-override edits
    # in sync without waiting for a manual reseed.
    #
    # We also track a hash of the agents.json content (stored on the volume)
    # to detect when the *content* actually changed across deploys. The
    # fingerprint above hashes templates + agents.json bytes together; the
    # render-only rebuild reads agents from the DB. If only agents.json
    # changed (e.g. an agent's category was edited), template fingerprints
    # match, but the render is still stale unless we explicitly trigger one.
    agents_changed = False
    if db.enabled():
        with open(os.path.join(BASE_DIR, "agents.json"), "rb") as f:
            agents_bytes = f.read()
        agents_hash = hashlib.sha256(agents_bytes).hexdigest()
        agents_seed = json.loads(agents_bytes)
        agents_hash_path = os.path.join(OUTPUT_DIR, ".agents_hash")
        try:
            prev_hash = open(agents_hash_path).read().strip()
        except OSError:
            prev_hash = ""
        seed_repos = {a["repo"] for a in agents_seed}
        db_repos = {a.get("repo") for a in db.load_agents()}
        if prev_hash != agents_hash or db_repos != seed_repos:
            for a in agents_seed:
                db.upsert_agent(a)
            # Remove agents from DB that were deleted from agents.json
            valid_repos = list(seed_repos)
            pruned = db.delete_agents_not_in(valid_repos)
            with open(agents_hash_path, "w") as f:
                f.write(agents_hash)
            agents_changed = True
            print(f"[startup] agents.json sync needed → synced {len(agents_seed)} entries into DB, pruned {pruned}")
        else:
            print(f"[startup] agents.json unchanged ({len(agents_seed)} entries) — DB sync skipped")
    # If the volume has no site yet, build one in the background so the service
    # comes up immediately and the site appears shortly after.
    if not os.path.isfile(DATA_PATH):
        _kick_boot_refresh("full", fingerprint)
        print("[startup] no data.json on volume — kicked off initial full build")
    elif os.environ.get("DISABLE_SCHEDULER") != "1" and _has_pending_signal_rows():
        # Pending must outrank repair-commits: ~10 rows legitimately sit at
        # 0 commits with a recent push (default-branch-quiet repos, re-verified
        # each boot by design), so _has_missing_commit_rows() is true on nearly
        # every boot — with repair first, freshly added agents could never get
        # their first signal refresh from a restart. The pending refresh fetches
        # commit counts for its rows anyway; repair runs on the next boot.
        _kick_boot_refresh("pending", fingerprint)
        print("[startup] detected provisional rows — kicked off pending refresh")
    elif os.environ.get("DISABLE_SCHEDULER") != "1" and _has_missing_commit_rows():
        # Same DISABLE_SCHEDULER guard as the pending branch above. Without it
        # pytest on a roster-add branch spawned a real, un-tokened fetch
        # subprocess that outlived the suite in 403-retry loops (and raced
        # pytest's own summary line out of the log). No production effect:
        # DISABLE_SCHEDULER is set only by the tests and docker-compose.
        _kick_boot_refresh("repair-commits", fingerprint)
        print("[startup] detected rows with missing commit counts — kicked off targeted repair refresh")
    elif seeded > 0 or stored_fingerprint != fingerprint or agents_changed:
        # Re-render when:
        #   - we just dropped prior-day snapshots into a volume that already
        #     had a site rendered without them, or
        #   - templates/assets changed in the image, or
        #   - agents.json content changed since last deploy (DB resync above)
        _kick_boot_refresh("render", fingerprint)
        if seeded > 0:
            print("[startup] history seeded into existing volume — kicked off render-only rebuild")
        elif stored_fingerprint != fingerprint:
            print("[startup] template/assets fingerprint changed — kicked off render-only rebuild")
        else:
            print("[startup] agents.json changed — kicked off render-only rebuild")


def _shutdown():
    global _scheduler
    if _scheduler is not None:
        _scheduler.shutdown(wait=False)
        _scheduler = None
    usage.flush()  # don't lose the last minute of counts on redeploy


def startup():
    _startup()


def shutdown():
    _shutdown()


def _install_mcp_route():
    route = mcp_server.fresh_streamable_http_app().routes[0]
    routes = app.router.routes
    routes[:] = [r for r in routes if getattr(r, "path", None) != "/mcp"]
    insert_at = next(
        (i for i, r in enumerate(routes) if getattr(r, "name", None) == "site"),
        len(routes),
    )
    routes.insert(insert_at, route)


# Trust-layer MCP server (Streamable HTTP) — register the SDK route directly
# before the static catch-all so POST /mcp is handled without a slash redirect.
# session_manager runs in _lifespan above.
_install_mcp_route()

# Static site LAST so /api, /badge, /submit, /correct take precedence.
app.mount("/", StaticFiles(directory=OUTPUT_DIR, html=True), name="site")