"""Reporting over the bounty ledger — budget, attribution, SLA, researchers.

Four questions the security team actually asks, and the measurement
decision behind each one:

1. **"How much of the FY budget is gone?"** — counts money *committed*
   (pending + requested) alongside money paid. A burn number that only
   counts ``paid`` reads healthy right up to the moment finance settles
   six pending rewards. Charged to ``fiscal_year``, which is stamped when
   the reward is decided, so a reward decided in FY26 and paid in FY27
   stays in FY26.

2. **"Whose code produces the most findings?"** — grouped on the
   *attribution snapshot* (``assignee_department`` / ``assignee_name``),
   never on a live join, so the chart does not silently re-attribute
   two-year-old reports when somebody changes team.

3. **"Are we fixing them on time?"** — on-time is measured against
   ``fix_due_date``, the deadline the report was actually given, not
   against today's tier config.

4. **"Who are the useful researchers?"** — valid-rate over *decided*
   reports only. Counting still-open submissions in the denominator makes
   an active submitter look worse than a dormant one.

Denominator discipline, stated once because it governs every function
here: a report with no severity has not been graded, so it is not a
"valid finding" and it is not an "invalid" one either — it is undecided,
and it stays out of both numerator and denominator. Deleted reports are
excluded everywhere.
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime
from decimal import Decimal
from typing import Any

from sqlalchemy import Select, case, func, literal_column, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core import fiscal_year as fy
from app.domains.bounty import config_service as cfg
from app.domains.bounty import constants as const
from app.domains.bounty.models import BountyReport, BountyResearcher

#: TIMESTAMPDIFF's unit must reach MySQL/TiDB as a bare SQL keyword.
#: Passing the unit as a plain Python string binds it as a ``%s`` parameter
#: and the statement dies with a 1064 syntax error at runtime — it compiles
#: fine, so only a real-DB execution catches it, which is why
#: ``scripts/check_tidb_parity.py`` greps for that shape. Cached as module
#: constants because both are used inside aggregates.
_DAY: Any = literal_column("DAY")
_HOUR: Any = literal_column("HOUR")

#: Statuses that mean "the security team ruled on it". ``closed`` is in
#: both lanes, so validity is read off ``severity`` (only the valid lane
#: ever gets one) rather than off the status.
_DECIDED = (
    const.STATUS_CONFIRMED,
    const.STATUS_FALSE_POSITIVE,
    const.STATUS_OUT_OF_SCOPE,
    const.STATUS_DUPLICATE,
    const.STATUS_CLOSED,
)

#: How a report is bucketed for attribution when it was never assigned.
UNATTRIBUTED = "__unattributed__"


def _live() -> Any:
    return BountyReport.deleted_at.is_(None)


def _is_valid() -> Any:
    """A graded report in (or settled out of) the valid lane."""
    return (
        BountyReport.severity.is_not(None)
        & BountyReport.status.in_(
            (const.STATUS_CONFIRMED, const.STATUS_CLOSED),
        )
    )


def _money(value: Any) -> Decimal:
    return Decimal(str(value or 0))


def _f(value: Any) -> float:
    return float(_money(value))


# ════════════════════════════════════════════════════════════════════
# 1. Budget
# ════════════════════════════════════════════════════════════════════


async def budget_report(
    db: AsyncSession, *, fiscal_year: int | None = None,
    track: str | None = None,
) -> dict[str, Any]:
    """FY burn for one funding track, or for both combined.

    ``track`` ∈ ``constants.BUDGET_TRACKS``; ``None`` sums the pots. The
    two are approved by different owners, so "how much is left" is only a
    real answer per track — the combined figure is for the leadership view.
    """
    start_month = await fy.get_fiscal_year_start_month(db)
    year = fiscal_year if fiscal_year is not None else fy.current_fiscal_year(
        start_month,
    )
    if track is not None and track not in const.BUDGET_TRACKS:
        track = None
    budget = await cfg.budget_for_fy(db, year, track=track)

    def _scoped(stmt: Select[Any]) -> Select[Any]:
        return stmt if track is None else stmt.where(
            BountyReport.budget_track == track,
        )

    rows = (
        await db.execute(
            _scoped(
                select(
                    BountyReport.payment_status,
                    func.coalesce(func.sum(BountyReport.reward_amount_usd), 0),
                    func.count(),
                )
                .where(_live(), BountyReport.fiscal_year == year)
                .group_by(BountyReport.payment_status),
            ),
        )
    ).all()
    paid = Decimal("0")
    committed = Decimal("0")
    waived = Decimal("0")
    paid_count = committed_count = waived_count = 0
    for status, amount, count in rows:
        value = _money(amount)
        if status in const.PAYMENT_SETTLED_SPENT:
            paid += value
            paid_count += int(count)
        elif status in const.PAYMENT_COMMITTED:
            committed += value
            committed_count += int(count)
        elif status == const.PAYMENT_WAIVED:
            waived += value
            waived_count += int(count)

    total = paid + committed
    ratio = (
        float(total / budget) if budget and budget > 0
        else (1.0 if total > 0 else 0.0)
    )

    # Per-severity spend, so "where does the money go" is answerable
    # without exporting the ledger.
    severity_rows = (
        await db.execute(
            _scoped(
                select(
                    BountyReport.severity,
                    func.coalesce(func.sum(BountyReport.reward_amount_usd), 0),
                    func.count(),
                )
                .where(
                    _live(),
                    BountyReport.fiscal_year == year,
                    BountyReport.payment_status.in_(
                        tuple(
                            const.PAYMENT_SETTLED_SPENT | const.PAYMENT_COMMITTED,
                        ),
                    ),
                )
                .group_by(BountyReport.severity),
            ),
        )
    ).all()

    tiers = await cfg.tier_map(db)
    by_severity: list[dict[str, Any]] = [
        {
            "severity": key or "ungraded",
            "label_zh": tiers[key].label_zh if key in tiers else "未定级",
            "label_en": tiers[key].label_en if key in tiers else "Ungraded",
            "label_ja": tiers[key].label_ja if key in tiers else "未評定",
            "color": tiers[key].color if key in tiers else None,
            "amount_usd": _f(amount),
            "count": int(count),
        }
        for key, amount, count in severity_rows
    ]
    by_severity.sort(key=lambda r: float(r["amount_usd"]), reverse=True)

    # Paid-per-month, keyed on the actual payment date. A reward decided
    # in FY26 but paid later still belongs to FY26's budget (that is what
    # ``fiscal_year`` pins), so the series is filtered by FY and bucketed
    # by ``paid_at`` — the two are deliberately different axes.
    month_rows = (
        await db.execute(
            _scoped(
                select(
                    func.date_format(BountyReport.paid_at, "%Y-%m"),
                    func.coalesce(func.sum(BountyReport.reward_amount_usd), 0),
                    func.count(),
                )
                .where(
                    _live(),
                    BountyReport.fiscal_year == year,
                    BountyReport.payment_status == const.PAYMENT_PAID,
                    BountyReport.paid_at.is_not(None),
                )
                .group_by(func.date_format(BountyReport.paid_at, "%Y-%m"))
                .order_by(func.date_format(BountyReport.paid_at, "%Y-%m")),
            ),
        )
    ).all()
    monthly = [
        {"period": period, "paid_usd": _f(amount), "count": int(count)}
        for period, amount, count in month_rows
        if period
    ]

    return {
        "fiscal_year": year,
        "fiscal_year_label": fy.fy_label(year, start_month),
        "fiscal_year_short": fy.fy_short_label(year, start_month),
        # ``None`` = both pots combined.
        "track": track,
        "budget_usd": _f(budget),
        "paid_usd": _f(paid),
        "committed_usd": _f(committed),
        "waived_usd": _f(waived),
        "total_usd": _f(total),
        "remaining_usd": _f(budget - total),
        "burn_ratio": round(ratio, 4),
        "level": (
            "over" if ratio >= const.BUDGET_OVER_RATIO
            else "warn" if ratio >= const.BUDGET_WARN_RATIO
            else "ok"
        ),
        "paid_count": paid_count,
        "committed_count": committed_count,
        "waived_count": waived_count,
        "by_severity": by_severity,
        "monthly_paid": monthly,
        "warn_ratio": const.BUDGET_WARN_RATIO,
        "over_ratio": const.BUDGET_OVER_RATIO,
    }


async def budget_by_track(
    db: AsyncSession, *, fiscal_year: int | None = None,
) -> dict[str, Any]:
    """Both pots side by side plus the combined total.

    Rendered as two cards rather than one: the tracks have different
    owners and different approval paths, so "83% of product is gone" and
    "4% of IT is gone" are two separate operational facts. A single
    combined bar would average them into a number nobody can act on.
    """
    per_track = []
    for track in const.BUDGET_TRACKS:
        per_track.append(
            await budget_report(db, fiscal_year=fiscal_year, track=track),
        )
    combined = await budget_report(db, fiscal_year=fiscal_year, track=None)
    return {"tracks": per_track, "combined": combined}


async def attribution_by_track(
    db: AsyncSession, *, fiscal_year: int | None = None,
) -> dict[str, Any]:
    """Report counts and spend per funding track — the "what am I paying
    for" cut of the same ledger."""
    stmt = (
        select(
            BountyReport.budget_track,
            func.count().label("total"),
            func.sum(case((_is_valid(), 1), else_=0)).label("valid"),
            func.coalesce(
                func.sum(
                    case(
                        (
                            BountyReport.payment_status.in_(
                                tuple(
                                    const.PAYMENT_SETTLED_SPENT
                                    | const.PAYMENT_COMMITTED,
                                ),
                            ),
                            BountyReport.reward_amount_usd,
                        ),
                        else_=0,
                    ),
                ),
                0,
            ).label("reward_usd"),
        )
        .where(_live())
        .group_by(BountyReport.budget_track)
    )
    if fiscal_year is not None:
        stmt = stmt.where(BountyReport.fiscal_year == fiscal_year)
    rows = {r[0]: r for r in (await db.execute(stmt)).all()}
    # Every track appears even at zero: an empty pot is a fact worth
    # showing, and a missing row reads as "no such budget".
    return {
        "items": [
            {
                "track": track,
                "total": int(rows[track].total or 0) if track in rows else 0,
                "valid": int(rows[track].valid or 0) if track in rows else 0,
                "reward_usd": (
                    _f(rows[track].reward_usd) if track in rows else 0.0
                ),
            }
            for track in const.BUDGET_TRACKS
        ],
    }


async def available_fiscal_years(db: AsyncSession) -> list[int]:
    """Fiscal years that actually carry a reward, newest first, always
    including the current one so the picker is never empty."""
    rows = (
        await db.execute(
            select(BountyReport.fiscal_year)
            .where(_live(), BountyReport.fiscal_year.is_not(None))
            .group_by(BountyReport.fiscal_year),
        )
    ).all()
    years = {int(r[0]) for r in rows if r[0] is not None}
    years.add(await cfg.current_fy(db))
    return sorted(years, reverse=True)


# ════════════════════════════════════════════════════════════════════
# 2. Attribution — "whose code produces findings"
# ════════════════════════════════════════════════════════════════════

GROUP_DEPARTMENT = "department"
GROUP_ASSIGNEE = "assignee"
GROUP_SYSTEM = "system"
GROUP_VULN_TYPE = "vuln_type"
ATTRIBUTION_GROUPS: tuple[str, ...] = (
    GROUP_DEPARTMENT, GROUP_ASSIGNEE, GROUP_SYSTEM, GROUP_VULN_TYPE,
)

_GROUP_COLUMN = {
    GROUP_DEPARTMENT: BountyReport.assignee_department,
    GROUP_ASSIGNEE: BountyReport.assignee_name,
    GROUP_SYSTEM: BountyReport.affected_system,
    GROUP_VULN_TYPE: BountyReport.vuln_type,
}


def _window(stmt: Select[Any], since: date | None, until: date | None) -> Select[Any]:
    if since is not None:
        stmt = stmt.where(
            BountyReport.reported_at >= datetime.combine(since, datetime.min.time()),
        )
    if until is not None:
        stmt = stmt.where(
            BountyReport.reported_at <= datetime.combine(until, datetime.max.time()),
        )
    return stmt


async def attribution_report(
    db: AsyncSession, *, group_by: str = GROUP_DEPARTMENT,
    since: date | None = None, until: date | None = None,
    fiscal_year: int | None = None,
) -> dict[str, Any]:
    """Per-group counts, validity, spend and fix performance.

    ``group_by`` ∈ ``ATTRIBUTION_GROUPS``. Rows with no value for the
    grouping column collapse into one explicit "未归属" bucket rather than
    disappearing — an unassigned backlog is a finding about the process,
    not a row to hide.
    """
    if group_by not in ATTRIBUTION_GROUPS:
        group_by = GROUP_DEPARTMENT
    column = _GROUP_COLUMN[group_by]
    today = date.today()

    stmt = (
        select(
            column.label("bucket"),
            func.count().label("total"),
            func.sum(case((_is_valid(), 1), else_=0)).label("valid"),
            func.sum(
                case(
                    (
                        BountyReport.status.in_((
                            const.STATUS_FALSE_POSITIVE,
                            const.STATUS_OUT_OF_SCOPE,
                            const.STATUS_DUPLICATE,
                        )),
                        1,
                    ),
                    else_=0,
                ),
            ).label("rejected"),
            func.sum(
                case((BountyReport.status.in_(_DECIDED), 1), else_=0),
            ).label("decided"),
            func.coalesce(
                func.sum(
                    case(
                        (
                            BountyReport.payment_status.in_(
                                tuple(
                                    const.PAYMENT_SETTLED_SPENT
                                    | const.PAYMENT_COMMITTED,
                                ),
                            ),
                            BountyReport.reward_amount_usd,
                        ),
                        else_=0,
                    ),
                ),
                0,
            ).label("reward_usd"),
            func.sum(
                case((BountyReport.remediation_status == const.REMEDIATION_FIXED, 1),
                     else_=0),
            ).label("fixed"),
            func.sum(
                case(
                    (
                        BountyReport.fix_due_date.is_not(None)
                        & (BountyReport.fix_due_date < today)
                        & BountyReport.remediation_status.not_in(
                            tuple(const.REMEDIATION_SETTLED),
                        ),
                        1,
                    ),
                    else_=0,
                ),
            ).label("overdue"),
            func.sum(
                case(
                    (
                        BountyReport.remediation_status.not_in(
                            tuple(const.REMEDIATION_SETTLED),
                        ),
                        1,
                    ),
                    else_=0,
                ),
            ).label("open_fix"),
            func.max(BountyReport.reported_at).label("latest"),
        )
        .where(_live())
        .group_by(column)
    )
    stmt = _window(stmt, since, until)
    if fiscal_year is not None:
        stmt = stmt.where(BountyReport.fiscal_year == fiscal_year)

    items: list[dict[str, Any]] = []
    for row in (await db.execute(stmt)).all():
        decided = int(row.decided or 0)
        valid = int(row.valid or 0)
        items.append({
            "bucket": row.bucket or UNATTRIBUTED,
            "total": int(row.total or 0),
            "valid": valid,
            "rejected": int(row.rejected or 0),
            "decided": decided,
            # Over DECIDED reports only: an open submission is not
            # evidence either way, and counting it makes an active area
            # look worse than a dormant one.
            "valid_rate": round(valid / decided, 4) if decided else None,
            "reward_usd": _f(row.reward_usd),
            "fixed": int(row.fixed or 0),
            "overdue": int(row.overdue or 0),
            "open_fix": int(row.open_fix or 0),
            "latest_reported_at": (
                row.latest.isoformat() if row.latest is not None else None
            ),
        })
    items.sort(key=lambda r: (r["valid"], r["total"]), reverse=True)

    return {
        "group_by": group_by,
        "items": items,
        "totals": {
            "buckets": len(items),
            "total": sum(i["total"] for i in items),
            "valid": sum(i["valid"] for i in items),
            "reward_usd": round(sum(i["reward_usd"] for i in items), 2),
            "overdue": sum(i["overdue"] for i in items),
        },
    }


# ════════════════════════════════════════════════════════════════════
# 3. Remediation SLA
# ════════════════════════════════════════════════════════════════════


async def remediation_report(
    db: AsyncSession, *, since: date | None = None, until: date | None = None,
) -> dict[str, Any]:
    """Fix performance per severity: how many are fixed, how many are
    late, and how long a fix actually takes.

    ``on_time`` is measured against the deadline the report was GIVEN
    (``fix_due_date``), not against today's tier configuration — editing
    a tier's SLA must not retroactively rewrite history.
    """
    today = date.today()
    stmt = (
        select(
            BountyReport.severity,
            func.count().label("total"),
            func.sum(
                case((BountyReport.remediation_status == const.REMEDIATION_FIXED, 1),
                     else_=0),
            ).label("fixed"),
            # The on-time DENOMINATOR: fixed **and** it had a deadline. A
            # report fixed without one has no on-time answer, exactly like
            # an unfixed one — counting it reported "0% on time" for
            # backfilled history where the truth is "we never set a
            # deadline for it" (caught 2026-08-24 against real data).
            func.sum(
                case(
                    (
                        (BountyReport.remediation_status == const.REMEDIATION_FIXED)
                        & BountyReport.fix_due_date.is_not(None),
                        1,
                    ),
                    else_=0,
                ),
            ).label("measurable"),
            func.sum(
                case(
                    (
                        (BountyReport.remediation_status == const.REMEDIATION_FIXED)
                        & BountyReport.fix_due_date.is_not(None)
                        & (func.date(BountyReport.fixed_at) <= BountyReport.fix_due_date),
                        1,
                    ),
                    else_=0,
                ),
            ).label("on_time"),
            func.sum(
                case(
                    (
                        BountyReport.fix_due_date.is_not(None)
                        & (BountyReport.fix_due_date < today)
                        & BountyReport.remediation_status.not_in(
                            tuple(const.REMEDIATION_SETTLED),
                        ),
                        1,
                    ),
                    else_=0,
                ),
            ).label("overdue_open"),
            func.avg(
                case(
                    (
                        BountyReport.fixed_at.is_not(None),
                        func.timestampdiff(
                            _DAY, BountyReport.reported_at, BountyReport.fixed_at,
                        ),
                    ),
                    else_=None,
                ),
            ).label("avg_days"),
        )
        .where(_live(), _is_valid())
        .group_by(BountyReport.severity)
    )
    stmt = _window(stmt, since, until)

    tiers = await cfg.tier_map(db)
    items: list[dict[str, Any]] = []
    for row in (await db.execute(stmt)).all():
        key = row.severity
        tier = tiers.get(key or "")
        fixed = int(row.fixed or 0)
        measurable = int(row.measurable or 0)
        on_time = int(row.on_time or 0)
        items.append({
            "severity": key or "ungraded",
            "label_zh": tier.label_zh if tier else "未定级",
            "label_en": tier.label_en if tier else "Ungraded",
            "label_ja": tier.label_ja if tier else "未評定",
            "color": tier.color if tier else None,
            "sla_days": tier.fix_sla_days if tier else None,
            "total": int(row.total or 0),
            "fixed": fixed,
            "on_time": on_time,
            # How many fixed reports we can actually judge punctuality on.
            # Surfaced so the UI can say "3 of 5 fixed have a deadline"
            # rather than implying the rate covers everything.
            "measurable": measurable,
            # Over fixed reports THAT HAD A DEADLINE. An unfixed report has
            # no on-time answer yet; neither does one fixed without a
            # deadline. Including either reports something other than
            # punctuality.
            "on_time_rate": round(on_time / measurable, 4) if measurable else None,
            "overdue_open": int(row.overdue_open or 0),
            "avg_days_to_fix": (
                round(float(row.avg_days), 1) if row.avg_days is not None else None
            ),
            "sort_order": tier.sort_order if tier else 999,
        })
    items.sort(key=lambda r: r["sort_order"])
    return {
        "items": items,
        "totals": {
            "valid": sum(i["total"] for i in items),
            "fixed": sum(i["fixed"] for i in items),
            "measurable": sum(i["measurable"] for i in items),
            "overdue_open": sum(i["overdue_open"] for i in items),
        },
    }


# ════════════════════════════════════════════════════════════════════
# 4. Researchers
# ════════════════════════════════════════════════════════════════════


async def researcher_leaderboard(
    db: AsyncSession, *, limit: int = 50,
) -> dict[str, Any]:
    """Who is worth paying attention to: submissions, valid rate over
    decided reports, and total paid."""
    stmt = (
        select(
            BountyResearcher.id,
            BountyResearcher.email,
            BountyResearcher.display_name,
            BountyResearcher.handle,
            BountyResearcher.country,
            BountyResearcher.is_blocked,
            func.count(BountyReport.id).label("total"),
            func.sum(case((_is_valid(), 1), else_=0)).label("valid"),
            func.sum(
                case((BountyReport.status.in_(_DECIDED), 1), else_=0),
            ).label("decided"),
            func.coalesce(
                func.sum(
                    case(
                        (
                            BountyReport.payment_status == const.PAYMENT_PAID,
                            BountyReport.reward_amount_usd,
                        ),
                        else_=0,
                    ),
                ),
                0,
            ).label("paid_usd"),
            func.min(BountyReport.reported_at).label("first_at"),
            func.max(BountyReport.reported_at).label("last_at"),
        )
        .join(
            BountyReport,
            (BountyReport.researcher_id == BountyResearcher.id)
            & BountyReport.deleted_at.is_(None),
        )
        .group_by(
            BountyResearcher.id, BountyResearcher.email,
            BountyResearcher.display_name, BountyResearcher.handle,
            BountyResearcher.country, BountyResearcher.is_blocked,
        )
    )
    rows = (await db.execute(stmt)).all()
    items = []
    for row in rows:
        decided = int(row.decided or 0)
        valid = int(row.valid or 0)
        items.append({
            "researcher_id": str(row.id),
            "email": row.email,
            "display_name": row.display_name,
            "handle": row.handle,
            "country": row.country,
            "is_blocked": bool(row.is_blocked),
            "report_count": int(row.total or 0),
            "valid_count": valid,
            "decided_count": decided,
            "valid_rate": round(valid / decided, 4) if decided else None,
            "paid_usd": _f(row.paid_usd),
            "first_report_at": (
                row.first_at.isoformat() if row.first_at is not None else None
            ),
            "last_report_at": (
                row.last_at.isoformat() if row.last_at is not None else None
            ),
        })
    # Valid findings first, then money, then volume — a high-volume
    # submitter with nothing valid should not top this list.
    items.sort(
        key=lambda r: (r["valid_count"], r["paid_usd"], r["report_count"]),
        reverse=True,
    )
    truncated = len(items) > limit
    return {
        "items": items[:limit],
        "total_researchers": len(items),
        "truncated": truncated,
    }


# ════════════════════════════════════════════════════════════════════
# 5. Programme health (the report the operator opens first)
# ════════════════════════════════════════════════════════════════════


async def programme_overview(
    db: AsyncSession, *, fiscal_year: int | None = None,
) -> dict[str, Any]:
    """Intake volume, decision mix, responsiveness and the budget, in one
    payload so the reports tab is a single round trip."""
    budget = await budget_report(db, fiscal_year=fiscal_year)
    tracks = await budget_by_track(db, fiscal_year=fiscal_year)

    status_rows = (
        await db.execute(
            select(BountyReport.status, func.count())
            .where(_live())
            .group_by(BountyReport.status),
        )
    ).all()
    by_status = {row[0]: int(row[1]) for row in status_rows}

    type_rows = (
        await db.execute(
            select(BountyReport.vuln_type, func.count())
            .where(_live())
            .group_by(BountyReport.vuln_type),
        )
    ).all()
    by_vuln_type = {(row[0] or "unclassified"): int(row[1]) for row in type_rows}

    # Monthly intake — the programme's activity curve.
    intake_rows = (
        await db.execute(
            select(
                func.date_format(BountyReport.reported_at, "%Y-%m"),
                func.count(),
                func.sum(case((_is_valid(), 1), else_=0)),
            )
            .where(_live())
            .group_by(func.date_format(BountyReport.reported_at, "%Y-%m"))
            .order_by(func.date_format(BountyReport.reported_at, "%Y-%m")),
        )
    ).all()
    monthly_intake = [
        {
            "period": period,
            "reports": int(total or 0),
            "valid": int(valid or 0),
        }
        for period, total, valid in intake_rows if period
    ]

    # Responsiveness: how long until the researcher heard back from us.
    # NULL ``last_replied_at`` means we have not replied at all — those
    # rows leave the average rather than counting as zero delay.
    reply = (
        await db.execute(
            select(
                func.avg(
                    func.timestampdiff(
                        _HOUR,
                        BountyReport.reported_at,
                        BountyReport.last_replied_at,
                    ),
                ),
                func.count(),
            )
            .where(_live(), BountyReport.last_replied_at.is_not(None)),
        )
    ).one()
    unanswered = (
        await db.execute(
            select(func.count())
            .select_from(BountyReport)
            .where(
                _live(),
                BountyReport.last_replied_at.is_(None),
                BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
            ),
        )
    ).scalar_one()

    decided = sum(by_status.get(s, 0) for s in _DECIDED)
    valid_total = (
        await db.execute(
            select(func.count()).select_from(BountyReport).where(_live(), _is_valid()),
        )
    ).scalar_one()

    return {
        "budget": budget,
        "budget_tracks": tracks["tracks"],
        "by_track": (await attribution_by_track(db, fiscal_year=fiscal_year))["items"],
        "fiscal_years": await available_fiscal_years(db),
        "totals": {
            "reports": sum(by_status.values()),
            "open": sum(by_status.get(s, 0) for s in const.OPEN_STATUSES),
            "decided": decided,
            "valid": int(valid_total),
            # Over decided reports — the programme's signal-to-noise.
            "valid_rate": (
                round(int(valid_total) / decided, 4) if decided else None
            ),
            "unanswered_open": int(unanswered),
            "avg_first_reply_hours": (
                round(float(reply[0]), 1) if reply[0] is not None else None
            ),
            "replied_reports": int(reply[1] or 0),
        },
        "by_status": by_status,
        "by_vuln_type": by_vuln_type,
        "monthly_intake": monthly_intake,
    }


# ════════════════════════════════════════════════════════════════════
# 6. Waiting on US — the programme's own responsiveness
# ════════════════════════════════════════════════════════════════════
#
# Added 2026-08-24 after reading the real correspondence with a repeat
# submitter: a report filed 2026-04-01 got its first substantive answer on
# 2026-06-30, after the researcher chased on 05-21 and 06-29. Nothing in
# the console surfaced that — the report sat in ``new`` and looked like
# any other row. Volume and budget reports cannot show it either, because
# the failure is *silence*, not a number going up.
#
# Measured from ``reported_at`` (when the mail actually arrived, not when
# it was keyed in) to ``last_replied_at``. A report we have never answered
# has NULL there, which is the whole point: those are the rows that need
# to be at the top, not averaged away.

#: Days of silence after which an unanswered report is called out. Chosen
#: against the real thread: the researcher's first chase came at 50 days,
#: so a week is early enough to matter and loose enough not to cry wolf.
REPLY_OVERDUE_DAYS = 7


async def awaiting_our_reply(
    db: AsyncSession, *, overdue_days: int = REPLY_OVERDUE_DAYS,
    limit: int = 100,
) -> dict[str, Any]:
    """Open reports the researcher has not heard back on, oldest first.

    ``never_replied`` and ``stale_since_last_reply`` are reported
    separately because they are different failures: one is "we never said
    anything", the other is "we answered once and then went quiet".
    """
    now = datetime.now(UTC).replace(tzinfo=None)
    rows = (
        await db.execute(
            select(BountyReport, BountyResearcher)
            .join(
                BountyResearcher,
                BountyResearcher.id == BountyReport.researcher_id,
            )
            .where(
                _live(),
                BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
            )
            .order_by(BountyReport.reported_at.asc())
            .limit(limit),
        )
    ).all()

    items: list[dict[str, Any]] = []
    never = 0
    stale = 0
    for report, researcher in rows:
        reported = report.reported_at
        if reported is not None and reported.tzinfo is not None:
            reported = reported.replace(tzinfo=None)
        replied = report.last_replied_at
        if replied is not None and replied.tzinfo is not None:
            replied = replied.replace(tzinfo=None)

        since = replied or reported
        waiting_days = (now - since).days if since is not None else None
        if waiting_days is None or waiting_days < overdue_days:
            continue
        if replied is None:
            never += 1
        else:
            stale += 1
        items.append({
            "report_id": str(report.id),
            "code": report.code,
            "title": report.title,
            "status": report.status,
            "severity": report.severity,
            "affected_system": report.affected_system,
            "researcher_email": researcher.email,
            "researcher_name": researcher.display_name or researcher.handle,
            "reported_at": reported.isoformat() if reported else None,
            "last_replied_at": replied.isoformat() if replied else None,
            "reply_count": report.reply_count or 0,
            "waiting_days": waiting_days,
            "never_replied": replied is None,
            # A repeat submitter going unanswered is a bigger problem than
            # a first-timer: they have already invested in us.
            "researcher_report_count": None,
        })
    items.sort(key=lambda r: r["waiting_days"], reverse=True)

    # One grouped query rather than N — the list is capped but the counts
    # should still be right for whoever is on it.
    if items:
        emails = {i["researcher_email"] for i in items}
        count_rows = (
            await db.execute(
                select(BountyResearcher.email, func.count(BountyReport.id))
                .join(
                    BountyReport,
                    (BountyReport.researcher_id == BountyResearcher.id)
                    & BountyReport.deleted_at.is_(None),
                )
                .where(BountyResearcher.email.in_(emails))
                .group_by(BountyResearcher.email),
            )
        ).all()
        counts: dict[str, int] = {
            str(row[0]): int(row[1]) for row in count_rows
        }
        for item in items:
            item["researcher_report_count"] = counts.get(
                str(item["researcher_email"]), 0,
            )

    return {
        "items": items,
        "overdue_days": overdue_days,
        "totals": {
            "waiting": len(items),
            "never_replied": never,
            "stale_since_last_reply": stale,
            "worst_waiting_days": items[0]["waiting_days"] if items else 0,
        },
    }


# ════════════════════════════════════════════════════════════════════
# Export rows
# ════════════════════════════════════════════════════════════════════

LEDGER_COLUMNS: tuple[tuple[str, str], ...] = (
    ("code", "编号"),
    ("title", "标题"),
    ("reported_at", "报告时间"),
    ("researcher_email", "白帽子邮箱"),
    ("researcher_name", "白帽子姓名"),
    ("affected_system", "受影响系统"),
    ("vuln_type", "漏洞类型"),
    ("severity", "级别"),
    ("status", "处置状态"),
    ("remediation_status", "修复状态"),
    ("assignee_name", "业务责任人"),
    ("assignee_department", "部门"),
    ("owner_decision", "业务结论"),
    ("fix_eta", "预计修复"),
    ("fix_due_date", "修复时限"),
    ("fixed_at", "实际修复"),
    ("is_overdue", "是否逾期"),
    ("reward_amount_usd", "赏金(USD)"),
    ("reward_is_override", "金额是否覆盖档位"),
    ("budget_track", "预算块"),
    ("fiscal_year", "归属财年"),
    ("payment_status", "付款状态"),
    ("payment_reference", "付款单号"),
    ("paid_at", "付款时间"),
    ("reply_count", "回复次数"),
    ("last_replied_at", "最近回复"),
    ("closed_at", "结案时间"),
)


async def ledger_rows(
    db: AsyncSession, *, fiscal_year: int | None = None,
    since: date | None = None, until: date | None = None,
    limit: int = 20_000,
) -> list[dict[str, Any]]:
    """One row per report, joined to the researcher, for the xlsx export.

    Money and researcher identity ARE included: the export is gated on
    ``can_view_security`` exactly like the console, and a payout ledger
    without the payee or the amount would not be a ledger.
    """
    today = date.today()
    stmt = (
        select(BountyReport, BountyResearcher)
        .join(BountyResearcher, BountyResearcher.id == BountyReport.researcher_id)
        .where(_live())
        .order_by(BountyReport.reported_at.desc(), BountyReport.code.desc())
        .limit(limit)
    )
    stmt = _window(stmt, since, until)
    if fiscal_year is not None:
        stmt = stmt.where(BountyReport.fiscal_year == fiscal_year)

    rows: list[dict[str, Any]] = []
    for report, researcher in (await db.execute(stmt)).all():
        overdue = bool(
            report.fix_due_date
            and report.fix_due_date < today
            and report.remediation_status not in const.REMEDIATION_SETTLED
        )
        rows.append({
            "code": report.code,
            "title": report.title,
            "reported_at": report.reported_at,
            "researcher_email": researcher.email,
            "researcher_name": researcher.display_name or researcher.handle,
            "affected_system": report.affected_system,
            "vuln_type": report.vuln_type,
            "severity": report.severity,
            "status": report.status,
            "remediation_status": report.remediation_status,
            "assignee_name": report.assignee_name,
            "assignee_department": report.assignee_department,
            "owner_decision": report.owner_decision,
            "fix_eta": report.fix_eta,
            "fix_due_date": report.fix_due_date,
            "fixed_at": report.fixed_at,
            "is_overdue": overdue,
            "reward_amount_usd": report.reward_amount_usd,
            "reward_is_override": report.reward_is_override,
            "budget_track": report.budget_track,
            "fiscal_year": report.fiscal_year,
            "payment_status": report.payment_status,
            "payment_reference": report.payment_reference,
            "paid_at": report.paid_at,
            "reply_count": report.reply_count,
            "last_replied_at": report.last_replied_at,
            "closed_at": report.closed_at,
        })
    return rows

# ════════════════════════════════════════════════════════════════════
# 6. Routable but unrouted — we know who to ask and did not ask
# ════════════════════════════════════════════════════════════════════


async def routable_unassigned(db: AsyncSession) -> dict[str, Any]:
    """Open reports linked to a system whose owner is active and unaware.

    Linking a report to a registry entry records who SHOULD answer; it
    deliberately does not assign, because assigning notifies a colleague
    and starts a fix clock — a human's call, and a bulk auto-assign would
    batch-notify a backlog, which this platform does not do to employees.

    The cost of that decision is this queue, and it is not hypothetical:
    a confirmed TiCard finding with a fix owed sat here while the
    system's owner had never been told. Backfilled history landed in it
    wholesale. So the queue is surfaced rather than left implicit — still
    one human click per row, but no longer invisible.

    The registry side comes through the coordinator, not a join: this
    domain does not import another domain's models.
    """
    from app.core.coordinators import system_registry_coordinator as registry

    owners = await registry.active_owner_by_system(db)
    if not owners:
        return {"items": [], "totals": {"routable": 0, "confirmed": 0}}

    rows = (
        await db.execute(
            select(BountyReport)
            .where(
                _live(),
                BountyReport.assignee_person_id.is_(None),
                BountyReport.system_id.is_not(None),
                BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
            )
            .order_by(
                # Confirmed first: those owe a fix, so silence there
                # costs more than on an untriaged submission.
                case((BountyReport.status == const.STATUS_CONFIRMED, 0), else_=1),
                BountyReport.reported_at.asc(),
            )
        )
    ).scalars().all()

    items: list[dict[str, Any]] = []
    for report in rows:
        owner = owners.get(str(report.system_id))
        if owner is None:
            # Linked to a system that is retired, ownerless, or whose
            # owner has left — a real gap too, but a different one with a
            # different fix (name an owner), already counted by the
            # registry's own coverage read.
            continue
        items.append({
            "report_id": str(report.id),
            "code": report.code,
            "title": report.title,
            "status": report.status,
            "severity": report.severity,
            "remediation_status": report.remediation_status,
            "reported_at": (
                report.reported_at.isoformat() if report.reported_at else None
            ),
            **owner,
        })
    return {
        "items": items,
        "totals": {
            "routable": len(items),
            "confirmed": len(
                [i for i in items if i["status"] == const.STATUS_CONFIRMED],
            ),
        },
    }


# ════════════════════════════════════════════════════════════════════
# 8. Data gaps — what has to be filled in before anything can move
# ════════════════════════════════════════════════════════════════════
#
# Added 2026-08-25. Every mechanism this domain gained recently reads a
# column that is empty on most rows: routing reads the registry owner,
# the reward ladder reads severity, budget attribution reads the linked
# system. Telling an operator "go and backfill it" without showing WHAT
# is missing and WHERE to fix it is how 97 reports sat unroutable while
# the console reported nothing wrong.
#
# This read is deliberately blunt: four numbers, each with the page that
# fixes it. It is not a health score — a score would let 6-of-8 ownerless
# average away against something green.


async def data_gaps(db: AsyncSession) -> dict[str, Any]:
    """What is missing, how much it blocks, and where to go fix it."""
    from app.core.coordinators import system_registry_coordinator as registry

    needing_owner = await registry.systems_needing_an_owner(db)
    by_id = {s["id"]: s for s in needing_owner}

    # How many live reports are pinned to a system nobody answers for.
    blocked = 0
    if by_id:
        rows = (
            await db.execute(
                select(
                    BountyReport.system_id, func.count().label("n"),
                ).where(
                    _live(),
                    BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
                    BountyReport.system_id.in_(tuple(by_id)),
                ).group_by(BountyReport.system_id),
            )
        ).all()
        for system_id, n in rows:
            blocked += int(n)
            entry = by_id.get(str(system_id))
            if entry is not None:
                entry["open_reports"] = int(n)

    # Only a genuinely missing level counts as a gap. A report sitting on
    # the level it ARRIVED with is not work: the whole point of the arrival
    # default (2026-08-26) is that nobody has to touch it, so counting it
    # here recreated the wall of 142 the default was meant to clear.
    #
    # The distinction still matters, just not here: `severity_is_default`
    # keeps the reply composer off the tier's reward letter, so a critical
    # finding nobody graded can never be offered the $0 one.
    ungraded = (
        await db.execute(
            select(func.count()).select_from(BountyReport).where(
                _live(),
                BountyReport.severity.is_(None),
                BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
            ),
        )
    ).scalar_one()

    # "Nobody has looked" only. A report somebody examined and ruled out
    # (open-source finding, CI workflow) is not work, and counting it
    # forever is how a gap number stops being believed.
    unlinked = (
        await db.execute(
            select(func.count()).select_from(BountyReport).where(
                _live(),
                BountyReport.system_id.is_(None),
                BountyReport.system_not_applicable.is_(False),
                BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
            ),
        )
    ).scalar_one()

    # Unlinked but naming a host is a different job from naming nothing at
    # all: the first is one click on a suggestion, the second needs a human
    # to read the report.
    unlinked_with_host = (
        await db.execute(
            select(func.count()).select_from(BountyReport).where(
                _live(),
                BountyReport.system_id.is_(None),
                BountyReport.system_not_applicable.is_(False),
                BountyReport.affected_system.is_not(None),
                BountyReport.affected_system != "",
                BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
            ),
        )
    ).scalar_one()

    ruled_out = (
        await db.execute(
            select(func.count()).select_from(BountyReport).where(
                _live(),
                BountyReport.system_not_applicable.is_(True),
            ),
        )
    ).scalar_one()

    silent = (
        await db.execute(
            select(
                func.count().label("n"),
                func.min(BountyReport.reported_at).label("oldest"),
            ).select_from(BountyReport).where(
                _live(),
                BountyReport.status.in_(tuple(const.OPEN_STATUSES)),
                or_(
                    BountyReport.reply_count == 0,
                    BountyReport.reply_count.is_(None),
                ),
            ),
        )
    ).one()

    oldest_days = None
    if silent.oldest is not None:
        oldest_days = (date.today() - silent.oldest.date()).days

    return {
        "ownerless_systems": {
            "count": len(needing_owner),
            "open_reports_blocked": blocked,
            "items": sorted(
                needing_owner,
                key=lambda s: (-s.get("open_reports", 0), s["name"]),
            ),
        },
        "ungraded": {"count": int(ungraded)},
        "not_applicable": {"count": int(ruled_out)},
        "unlinked": {
            "count": int(unlinked),
            "with_host": int(unlinked_with_host),
            "without_host": int(unlinked) - int(unlinked_with_host),
        },
        "never_replied": {
            "count": int(silent.n or 0),
            "oldest_days": oldest_days,
        },
    }


# ════════════════════════════════════════════════════════════════════
# 9. The matrix — ownership × stage, coloured by how long people waited
# ════════════════════════════════════════════════════════════════════
#
# The console's landing page was a 149-row flat table, and the complaint
# about it was exactly right: it had no hierarchy because the DATA has no
# hierarchy. 114 of 149 rows share one status, 142 share one (empty)
# severity, and 84 of 149 share one hostname. Every column was "sorting"
# on a value that is identical on 90% of rows.
#
# Two axes fix that, and they are the two questions actually asked:
# whose problem is this (rows, banded by budget track) and what is it
# waiting on (columns). The heat is silence duration, which is the only
# dimension here that ranks priority objectively — it runs from 2 days to
# 295.
#
# Aggregated in Python over a narrow column projection rather than six
# GROUP BY round-trips: the whole live corpus is 149 rows today and an
# admin console at 10k would still be one cheap scan, whereas six
# separate aggregates would have to agree with each other about the stage
# definitions, and that is exactly the drift that makes a dashboard stop
# being believed.

#: Stage keys, in pipeline order. These are the six the platform owner
#: signed off on; renaming one is a product decision, not a refactor.
MATRIX_STAGES: tuple[str, ...] = (
    "need_reply",
    "need_grade",
    "need_assign",
    "await_owner",
    "need_fix",
    "money",
)

#: Silence buckets the heat ramp maps onto, in days.
_HEAT_EDGES: tuple[int, ...] = (7, 30, 90, 180)


def _heat(days: int | None) -> int:
    """0…4. Deliberately a bucket rather than a raw day count: the UI needs
    a colour, and letting the client bucket it is how two pages end up
    disagreeing about what 'urgent' means."""
    if days is None:
        return 0
    for index, edge in enumerate(_HEAT_EDGES):
        if days < edge:
            return index
    return len(_HEAT_EDGES)


def _bucket_for(
    row: Any,
    by_system_id: dict[str, dict[str, Any]],
    needing_owner: set[str],
) -> tuple[str, dict[str, Any]]:
    """Which ownership row a report belongs to.

    Linked reports group by their registry system — that is the thing an
    owner is attached to. Unlinked ones group by the hostname they name, so
    an unregistered domain stays visible AS a domain rather than dissolving
    into a catch-all. Reports naming nothing get their own band: they are
    structurally unattributable, and pretending otherwise is what made the
    old flat table unreadable.

    Module level, not a closure, because the lane view needs the identical
    grouping — two copies of this is how a console starts disagreeing with
    itself about which row a report is in.
    """
    if row.system_id and str(row.system_id) in by_system_id:
        system = by_system_id[str(row.system_id)]
        return f"sys:{system['id']}", {
            "key": f"sys:{system['id']}",
            "label": system["name"],
            "sublabel": (system["hostnames"] or [system["slug"]])[0],
            "track": system["budget_track"],
            "system_id": system["id"],
            "registered": True,
            "ownerless": system["id"] in needing_owner,
        }
    host = (row.affected_system or "").strip()
    if host:
        return f"host:{host.lower()}", {
            "key": f"host:{host.lower()}",
            "label": host,
            "sublabel": None,
            "track": row.budget_track or const.DEFAULT_BUDGET_TRACK,
            "system_id": None,
            "registered": False,
            "ownerless": True,
        }
    return "none:unattributed", {
        "key": "none:unattributed",
        "label": None,
        "sublabel": None,
        "track": "unattributed",
        "system_id": None,
        "registered": False,
        "ownerless": True,
    }


def _report_stages(row: Any, *, assignable: bool) -> list[str]:
    """Which stage columns this report counts in.

    A report can be in more than one — "not graded" and "never replied" are
    both true of most of this corpus, and forcing a single bucket would
    understate both. Module level for the same reason as ``_bucket_for``:
    the matrix and the lane view must agree, and six aggregates that each
    re-derive this is the drift that makes a dashboard stop being believed.
    """
    out: list[str] = []
    is_open = row.status in const.OPEN_STATUSES
    if is_open and not (row.reply_count or 0):
        out.append("need_reply")
    # Same rule as the gap panel: the arrival default is not outstanding
    # work, so it does not fill the 待定级 column.
    if is_open and row.severity is None:
        out.append("need_grade")
    # Only when assigning is actually possible. A report on a system with
    # no owner is rendered as an impossible cell, and counting it in the
    # column total anyway makes the header disagree with the cells under it.
    if is_open and row.assignee_person_id is None and assignable:
        out.append("need_assign")
    if is_open and row.assignee_person_id and not row.owner_decision:
        out.append("await_owner")
    if (
        row.status == const.STATUS_CONFIRMED
        and row.remediation_status not in const.REMEDIATION_SETTLED
    ):
        out.append("need_fix")
    if (
        row.reward_amount_usd is not None
        and row.payment_status in ("pending", "requested")
    ):
        out.append("money")
    return out


async def matrix(db: AsyncSession) -> dict[str, Any]:
    """Ownership rows × stage columns, with counts and silence heat."""
    from app.core.coordinators import system_registry_coordinator as registry

    systems = await registry.live_systems_for_matching(db)
    by_system_id = {s["id"]: s for s in systems}
    needing_owner = {
        s["id"] for s in await registry.systems_needing_an_owner(db)
    }

    rows = (
        await db.execute(
            select(
                BountyReport.id,
                BountyReport.code,
                BountyReport.status,
                BountyReport.severity,
                BountyReport.severity_is_default,
                BountyReport.system_id,
                BountyReport.system_not_applicable,
                BountyReport.affected_system,
                BountyReport.budget_track,
                BountyReport.assignee_person_id,
                BountyReport.owner_decision,
                BountyReport.remediation_status,
                BountyReport.reward_amount_usd,
                BountyReport.payment_status,
                BountyReport.reply_count,
                BountyReport.reported_at,
            ).where(_live()),
        )
    ).all()

    today = date.today()
    buckets: dict[str, dict[str, Any]] = {}

    totals = dict.fromkeys(MATRIX_STAGES, 0)

    for row in rows:
        key, meta = _bucket_for(row, by_system_id, needing_owner)
        entry = buckets.setdefault(key, {
            **meta,
            "total": 0,
            "cells": {
                stage: {"count": 0, "oldest_days": None, "na": False}
                for stage in MATRIX_STAGES
            },
        })
        entry["total"] += 1
        age = (
            (today - row.reported_at.date()).days if row.reported_at else None
        )
        assignable = bool(
            row.system_id and str(row.system_id) not in needing_owner,
        )
        for stage in _report_stages(row, assignable=assignable):
            cell = entry["cells"][stage]
            cell["count"] += 1
            if age is not None and (
                cell["oldest_days"] is None or age > cell["oldest_days"]
            ):
                cell["oldest_days"] = age
            totals[stage] += 1

    # A cell that cannot exist is not the same as a cell that is empty.
    # "Assign this to the system owner" is impossible while the system has
    # no owner, and rendering it as 0 invites somebody to conclude the
    # work is done.
    for entry in buckets.values():
        if entry["ownerless"] or not entry["system_id"]:
            entry["cells"]["need_assign"]["na"] = True

    for entry in buckets.values():
        for stage, cell in entry["cells"].items():
            cell["heat"] = _heat(cell["oldest_days"]) if cell["count"] else 0

    band_order = {"product": 0, "it": 1, "unattributed": 2}
    ordered = sorted(
        buckets.values(),
        key=lambda e: (
            band_order.get(e["track"], 9),
            -e["total"],
            e["label"] or "",
        ),
    )
    return {
        "stages": list(MATRIX_STAGES),
        "heat_edges": list(_HEAT_EDGES),
        "rows": ordered,
        "totals": totals,
        "bands": [
            {
                "track": track,
                "reports": sum(
                    e["total"] for e in ordered if e["track"] == track
                ),
            }
            for track in ("product", "it", "unattributed")
        ],
    }


async def lane(
    db: AsyncSession, *, row_key: str, per_stage: int = 6,
) -> dict[str, Any]:
    """One ownership row as a left-to-right pipeline, with real reports.

    The matrix answers "how much and how hot"; a number cannot tell you
    *which* report has been waiting 295 days. This is the drill-down for a
    single row, so the horizontal space that made a full swimlane board
    unworkable at 8+ systems is no longer a problem.

    Shares ``_bucket_for`` and ``_report_stages`` with the matrix by
    construction: a second definition of "which stage is this in" is how
    the two views would start disagreeing.
    """
    from app.core.coordinators import system_registry_coordinator as registry

    systems = await registry.live_systems_for_matching(db)
    by_system_id = {s["id"]: s for s in systems}
    needing_owner = {
        s["id"] for s in await registry.systems_needing_an_owner(db)
    }

    # Narrow in SQL first. Loading every live report to show six cards per
    # column is fine at 149 and wasteful at 10k, and the row key maps onto
    # a cheap predicate. The predicate is a deliberate SUPERSET —
    # ``_bucket_for`` still decides, so a report linked to a *retired*
    # system (which falls through to its hostname bucket) lands in the same
    # place here as it does in the matrix.
    stmt = select(BountyReport).where(_live())
    if row_key.startswith("sys:"):
        try:
            stmt = stmt.where(
                BountyReport.system_id == uuid.UUID(row_key[4:]),
            )
        except ValueError:
            return {
                "row": None,
                "stages": {
                    stage: {"count": 0, "items": [], "heat": 0, "hidden": 0}
                    for stage in MATRIX_STAGES
                },
                "order": list(MATRIX_STAGES),
            }
    elif row_key.startswith("host:"):
        # TRIM mirrors ``_bucket_for``'s ``.strip()``. Without it a stored
        # " www.pingcap.com " counted in the matrix row and produced no
        # cards in its own lane — the value arrives by paste out of a
        # researcher's mail, and nothing used to strip it. ``create_report``
        # / ``update_report`` now do, so the set of legacy padded rows can
        # only shrink; MySQL TRIM covers spaces, which is what pasting
        # produces.
        stmt = stmt.where(
            func.lower(func.trim(BountyReport.affected_system)) == row_key[5:],
        )
    else:
        stmt = stmt.where(
            or_(
                BountyReport.affected_system.is_(None),
                func.trim(BountyReport.affected_system) == "",
            ),
        )

    rows = (
        await db.execute(stmt.order_by(BountyReport.reported_at.asc()))
    ).scalars().all()

    today = date.today()
    meta: dict[str, Any] | None = None
    stages: dict[str, dict[str, Any]] = {
        stage: {"count": 0, "items": []} for stage in MATRIX_STAGES
    }
    for report in rows:
        key, row_meta = _bucket_for(report, by_system_id, needing_owner)
        if key != row_key:
            continue
        meta = meta or row_meta
        assignable = bool(
            report.system_id and str(report.system_id) not in needing_owner,
        )
        age = (
            (today - report.reported_at.date()).days
            if report.reported_at else None
        )
        for stage in _report_stages(report, assignable=assignable):
            bucket = stages[stage]
            bucket["count"] += 1
            # Oldest first, capped: the tail of a 59-report cell is not
            # what the operator needs to see, and the count says how much
            # was left out.
            if len(bucket["items"]) < per_stage:
                bucket["items"].append({
                    "id": str(report.id),
                    "code": report.code,
                    "title": report.title,
                    "severity": report.severity,
                    "days_silent": age,
                    "assignee_name": report.assignee_name,
                    "reward_amount_usd": (
                        float(report.reward_amount_usd)
                        if report.reward_amount_usd is not None else None
                    ),
                    "payment_status": report.payment_status,
                })

    for bucket in stages.values():
        oldest = max(
            (i["days_silent"] or 0 for i in bucket["items"]), default=None,
        )
        bucket["heat"] = _heat(oldest) if bucket["count"] else 0
        bucket["hidden"] = max(0, bucket["count"] - len(bucket["items"]))

    if meta is None:
        # An unknown or now-empty row is not an error: the operator may
        # have just cleared the last report in it.
        return {"row": None, "stages": stages, "order": list(MATRIX_STAGES)}
    if meta["ownerless"] or not meta["system_id"]:
        stages["need_assign"]["na"] = True
    return {"row": meta, "stages": stages, "order": list(MATRIX_STAGES)}