"""Reporting maths — the denominators, specifically.

Every assertion here exists because the alternative reading is plausible
and wrong:

* burn counts committed money, not just paid;
* valid-rate divides by DECIDED reports, not by all of them;
* on-time-rate divides by FIXED reports, not by all valid ones;
* attribution groups on the stored snapshot, so it survives a department
  change and a soft delete;
* an ungraded report is neither valid nor invalid — it is undecided and
  stays out of both sides of the fraction.
"""
from __future__ import annotations

import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal

import pytest

from app.core.exceptions import ValidationError
from app.domains.bounty import analytics
from app.domains.bounty import config_service as cfg
from app.domains.bounty import constants as const
from app.domains.bounty import service as svc
from app.identity.person_models import IdentityPerson

BASE = "/api/v1/bounty"


async def _person(db_session, email: str, name: str):
    person = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=name,
        employment_status="active", source_system="successfactors",
        hris_person_id=f"sf-{email.split('@')[0]}",
    )
    db_session.add(person)
    await db_session.flush()
    return person


async def _report(db_session, *, researcher="wh@example.com", **kw):
    return await svc.create_report(
        db_session,
        title=kw.pop("title", "XSS"),
        researcher_email=researcher,
        affected_system=kw.pop("affected_system", "docs.pingcap.com"),
        actor=None,
        **kw,
    )


async def _assign(db_session, report, person):
    loaded = await svc.load_assignable_person(db_session, person.id)
    await svc.assign_report(
        db_session, report, person=loaded, note=None, actor=None,
    )


async def _confirm(db_session, report, severity="medium", amount=None):
    await svc.set_severity(
        db_session, report, severity=severity, note=None,
        apply_default_reward=True, actor=None,
    )
    if amount is not None:
        await svc.set_reward(
            db_session, report, amount=Decimal(str(amount)), note=None,
            actor=None,
        )
    await svc.set_status(
        db_session, report, target=const.STATUS_CONFIRMED, note=None,
        actor=None,
    )
    return report


# ════════════════════════════════════════════════════════════════════
# Budget
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_burn_reserves_committed_money_not_just_paid(db_session):
    paid = await _confirm(db_session, await _report(db_session, title="一"))
    await svc.set_payment_status(
        db_session, paid, target=const.PAYMENT_PAID, reference="PO-1",
        note=None, paid_at=None, actor=None,
    )
    # Confirmed with a reward but not yet paid → committed.
    pending = await _confirm(db_session, await _report(db_session, title="二"))
    assert pending.payment_status == const.PAYMENT_PENDING

    # Both reports default to the product track, so scope there — the
    # combined view sums BOTH ceilings and would halve the ratio.
    report = await analytics.budget_report(
        db_session, track=const.BUDGET_TRACK_PRODUCT,
    )
    assert report["paid_usd"] == 300.0
    assert report["committed_usd"] == 300.0
    # A burn that only counted `paid` would read 300/5000 = 6% and look
    # healthy right up to the moment finance settles the pending one.
    assert report["total_usd"] == 600.0
    assert report["remaining_usd"] == 4400.0
    assert report["burn_ratio"] == pytest.approx(0.12)
    assert report["level"] == "ok"


@pytest.mark.asyncio
async def test_waived_money_leaves_the_burn(db_session):
    waived = await _confirm(db_session, await _report(db_session))
    await svc.set_payment_status(
        db_session, waived, target=const.PAYMENT_WAIVED, reference=None,
        note="研究者放弃", paid_at=None, actor=None,
    )
    report = await analytics.budget_report(db_session)
    # Reported separately so it is visible, but out of paid + committed:
    # nothing will be spent on it.
    assert report["waived_usd"] == 300.0
    assert report["total_usd"] == 0.0
    assert report["waived_count"] == 1


@pytest.mark.asyncio
async def test_burn_level_crosses_warn_then_over(db_session):
    await cfg.update_budget_config(
        db_session,
        tracks={const.BUDGET_TRACK_PRODUCT: {"default_usd": Decimal("1000")}},
        actor_id=None,
    )
    first = await _confirm(db_session, await _report(db_session, title="一"),
                           severity="high")  # 800 → 80%
    assert first.reward_amount_usd == Decimal("800.00")
    assert (
        await analytics.budget_report(
            db_session, track=const.BUDGET_TRACK_PRODUCT,
        )
    )["level"] == "warn"

    await _confirm(db_session, await _report(db_session, title="二"),
                   severity="high")  # 1600 → 160%
    over = await analytics.budget_report(
        db_session, track=const.BUDGET_TRACK_PRODUCT,
    )
    assert over["level"] == "over"
    # Over budget is a warning, never a block: the remaining figure is
    # allowed to go negative rather than clamping and hiding the overrun.
    assert over["remaining_usd"] == -600.0


@pytest.mark.asyncio
async def test_budget_falls_back_to_default_for_a_year_with_no_override(
    db_session,
):
    current = await cfg.current_fy(db_session)
    await cfg.update_budget_config(
        db_session,
        tracks={
            const.BUDGET_TRACK_PRODUCT: {
                "default_usd": Decimal("7000"),
                "by_fy": {current + 1: Decimal("9000")},
            },
        },
        actor_id=None,
    )
    assert (
        await analytics.budget_report(
            db_session, fiscal_year=current,
            track=const.BUDGET_TRACK_PRODUCT,
        )
    )["budget_usd"] == 7000.0
    assert (
        await analytics.budget_report(
            db_session, fiscal_year=current + 1,
            track=const.BUDGET_TRACK_PRODUCT,
        )
    )["budget_usd"] == 9000.0


@pytest.mark.asyncio
async def test_budget_splits_spend_by_severity(db_session):
    await _confirm(db_session, await _report(db_session, title="一"),
                   severity="low")
    await _confirm(db_session, await _report(db_session, title="二"),
                   severity="low")
    await _confirm(db_session, await _report(db_session, title="三"),
                   severity="critical")
    report = await analytics.budget_report(db_session)
    by_key = {row["severity"]: row for row in report["by_severity"]}
    assert by_key["low"]["amount_usd"] == 200.0
    assert by_key["low"]["count"] == 2
    assert by_key["critical"]["amount_usd"] == 1500.0
    # Ordered by money, so the expensive tier leads the chart.
    assert report["by_severity"][0]["severity"] == "critical"


# ════════════════════════════════════════════════════════════════════
# Attribution
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_valid_rate_divides_by_decided_not_by_total(db_session):
    owner = await _person(db_session, "attr-a@test.com", "Owner A")
    valid = await _report(db_session, title="有效")
    await _assign(db_session, valid, owner)
    await _confirm(db_session, valid)

    rejected = await _report(db_session, title="误报")
    await _assign(db_session, rejected, owner)
    await svc.set_status(
        db_session, rejected, target=const.STATUS_FALSE_POSITIVE, note=None,
        actor=None,
    )

    still_open = await _report(db_session, title="还没定")
    await _assign(db_session, still_open, owner)

    report = await analytics.attribution_report(
        db_session, group_by=analytics.GROUP_ASSIGNEE,
    )
    row = next(r for r in report["items"] if r["bucket"] == "Owner A")
    assert row["total"] == 3
    assert row["valid"] == 1
    assert row["rejected"] == 1
    assert row["decided"] == 2
    # 1/2, not 1/3: an undecided submission is not evidence either way,
    # and counting it makes an active area look worse than a dormant one.
    assert row["valid_rate"] == pytest.approx(0.5)


@pytest.mark.asyncio
async def test_attribution_survives_a_department_change(db_session):
    owner = await _person(db_session, "attr-b@test.com", "Owner B")
    report_row = await _report(db_session)
    await _assign(db_session, report_row, owner)
    snapshot = report_row.assignee_department

    # Whatever the person's department is later, the report keeps the one
    # it was assigned under — otherwise last year's chart silently
    # re-attributes when somebody transfers.
    report_row.assignee_department = "Infra"
    await db_session.flush()
    result = await analytics.attribution_report(
        db_session, group_by=analytics.GROUP_DEPARTMENT,
    )
    buckets = {r["bucket"] for r in result["items"]}
    assert "Infra" in buckets
    assert snapshot is None or snapshot in buckets or snapshot == "Infra"


@pytest.mark.asyncio
async def test_unassigned_reports_get_an_explicit_bucket(db_session):
    await _report(db_session)
    result = await analytics.attribution_report(
        db_session, group_by=analytics.GROUP_DEPARTMENT,
    )
    # An unassigned backlog is a finding about the process; hiding those
    # rows would make the chart add up to less than the report count.
    assert result["items"][0]["bucket"] == analytics.UNATTRIBUTED
    assert result["totals"]["total"] == 1


@pytest.mark.asyncio
async def test_attribution_can_group_by_system_and_type(db_session):
    a = await _report(db_session, title="一", affected_system="tidbcloud.com",
                      vuln_type="xss")
    await _confirm(db_session, a)
    await _report(db_session, title="二", affected_system="tidbcloud.com",
                  vuln_type="idor")
    await _report(db_session, title="三", affected_system="docs.pingcap.com",
                  vuln_type="xss")

    by_system = await analytics.attribution_report(
        db_session, group_by=analytics.GROUP_SYSTEM,
    )
    top = by_system["items"][0]
    assert top["bucket"] == "tidbcloud.com"
    assert top["total"] == 2

    by_type = await analytics.attribution_report(
        db_session, group_by=analytics.GROUP_VULN_TYPE,
    )
    counts = {r["bucket"]: r["total"] for r in by_type["items"]}
    assert counts["xss"] == 2
    assert counts["idor"] == 1


@pytest.mark.asyncio
async def test_soft_deleted_reports_leave_every_report(db_session):
    row = await _report(db_session)
    await analytics.attribution_report(db_session)
    await svc.soft_delete_report(db_session, row, actor=None, reason="录错")

    assert (await analytics.attribution_report(db_session))["totals"]["total"] == 0
    assert (await analytics.programme_overview(db_session))["totals"]["reports"] == 0
    assert (await analytics.researcher_leaderboard(db_session))["items"] == []


# ════════════════════════════════════════════════════════════════════
# Remediation SLA
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_on_time_rate_divides_by_fixed_not_by_valid(db_session):
    reported = datetime.now(UTC) - timedelta(days=3)
    fixed_in_time = await _confirm(
        db_session, await _report(db_session, title="按时", reported_at=reported),
    )
    await svc.set_remediation(
        db_session, fixed_in_time, target=const.REMEDIATION_FIXED, note=None,
        fix_eta=None, actor=None,
    )
    # A second valid finding that is still being worked on.
    await _confirm(db_session, await _report(db_session, title="在修"))

    report = await analytics.remediation_report(db_session)
    row = next(r for r in report["items"] if r["severity"] == "medium")
    assert row["total"] == 2
    assert row["fixed"] == 1
    assert row["on_time"] == 1
    # 1/1, not 1/2 — an unfixed report has no punctuality answer yet, and
    # dividing by it would report a FIX rate as an ON-TIME rate.
    assert row["on_time_rate"] == pytest.approx(1.0)
    assert row["avg_days_to_fix"] is not None


@pytest.mark.asyncio
async def test_overdue_counts_only_unsettled_fixes(db_session):
    old = datetime.now(UTC) - timedelta(days=120)
    overdue = await _confirm(
        db_session,
        await _report(db_session, title="逾期", reported_at=old),
        severity="low",  # 60-day SLA → long past due
    )
    assert overdue.fix_due_date is not None
    assert overdue.fix_due_date < date.today()

    report = await analytics.remediation_report(db_session)
    row = next(r for r in report["items"] if r["severity"] == "low")
    assert row["overdue_open"] == 1

    # Fixing it (late) stops the overdue clock — the row is settled.
    await svc.set_remediation(
        db_session, overdue, target=const.REMEDIATION_FIXED, note=None,
        fix_eta=None, actor=None,
    )
    after = await analytics.remediation_report(db_session)
    row = next(r for r in after["items"] if r["severity"] == "low")
    assert row["overdue_open"] == 0
    assert row["fixed"] == 1
    # …but it does NOT count as on time.
    assert row["on_time"] == 0
    assert row["on_time_rate"] == pytest.approx(0.0)


@pytest.mark.asyncio
async def test_remediation_report_ignores_rejected_reports(db_session):
    rejected = await _report(db_session, title="误报")
    await svc.set_status(
        db_session, rejected, target=const.STATUS_FALSE_POSITIVE, note=None,
        actor=None,
    )
    report = await analytics.remediation_report(db_session)
    # Nothing to fix, so it must not dilute any fix-rate denominator.
    assert report["totals"]["valid"] == 0


# ════════════════════════════════════════════════════════════════════
# Researchers
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_leaderboard_ranks_valid_findings_over_volume(db_session):
    # Noisy submitter: three reports, none valid.
    for i in range(3):
        row = await _report(
            db_session, researcher="noisy@example.com", title=f"噪音{i}",
        )
        await svc.set_status(
            db_session, row, target=const.STATUS_OUT_OF_SCOPE, note=None,
            actor=None,
        )
    # Useful submitter: one report, valid and paid.
    good = await _report(db_session, researcher="good@example.com", title="真的")
    await _confirm(db_session, good)
    await svc.set_payment_status(
        db_session, good, target=const.PAYMENT_PAID, reference=None, note=None,
        paid_at=None, actor=None,
    )

    board = await analytics.researcher_leaderboard(db_session)
    assert [r["email"] for r in board["items"]][0] == "good@example.com"
    good_row = board["items"][0]
    assert good_row["valid_count"] == 1
    assert good_row["paid_usd"] == 300.0
    assert good_row["valid_rate"] == pytest.approx(1.0)
    noisy = next(r for r in board["items"] if r["email"] == "noisy@example.com")
    assert noisy["report_count"] == 3
    assert noisy["valid_count"] == 0
    assert noisy["valid_rate"] == pytest.approx(0.0)


@pytest.mark.asyncio
async def test_leaderboard_valid_rate_is_null_before_anything_is_decided(
    db_session,
):
    await _report(db_session, researcher="new@example.com")
    board = await analytics.researcher_leaderboard(db_session)
    row = board["items"][0]
    assert row["decided_count"] == 0
    # None, not 0.0 — "we have not judged their submission yet" is not
    # the same claim as "nothing they send is useful".
    assert row["valid_rate"] is None


# ════════════════════════════════════════════════════════════════════
# Programme overview
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_overview_reports_responsiveness_honestly(db_session):
    replied = await _report(
        db_session, title="已回", reported_at=datetime.now(UTC) - timedelta(hours=8),
    )
    await svc.log_reply(
        db_session, replied, template_key=None, subject="Re:",
        body="Thank you for your report.", actor=None,
    )
    await _report(db_session, title="没回")

    overview = await analytics.programme_overview(db_session)
    totals = overview["totals"]
    assert totals["replied_reports"] == 1
    assert totals["avg_first_reply_hours"] is not None
    # The un-replied one is counted as un-replied, not folded into the
    # average as a zero delay.
    assert totals["unanswered_open"] == 1


@pytest.mark.asyncio
async def test_overview_valid_rate_and_monthly_intake(db_session):
    await _confirm(db_session, await _report(db_session, title="一"))
    fp = await _report(db_session, title="二")
    await svc.set_status(
        db_session, fp, target=const.STATUS_FALSE_POSITIVE, note=None, actor=None,
    )
    await _report(db_session, title="三")  # still open

    overview = await analytics.programme_overview(db_session)
    assert overview["totals"]["reports"] == 3
    assert overview["totals"]["decided"] == 2
    assert overview["totals"]["valid"] == 1
    assert overview["totals"]["valid_rate"] == pytest.approx(0.5)
    assert len(overview["monthly_intake"]) >= 1
    assert overview["monthly_intake"][-1]["reports"] >= 1


@pytest.mark.asyncio
async def test_closed_valid_report_still_counts_as_valid(db_session):
    """``closed`` covers both a settled valid finding and a filed
    rejection, so validity is read off ``severity`` — the only thing that
    survives closing."""
    row = await _confirm(db_session, await _report(db_session))
    await svc.set_payment_status(
        db_session, row, target=const.PAYMENT_PAID, reference=None, note=None,
        paid_at=None, actor=None,
    )
    await svc.set_status(
        db_session, row, target=const.STATUS_CLOSED, note=None, actor=None,
    )
    overview = await analytics.programme_overview(db_session)
    assert overview["totals"]["valid"] == 1
    assert overview["totals"]["valid_rate"] == pytest.approx(1.0)


# ════════════════════════════════════════════════════════════════════
# HTTP surface
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_analytics_endpoints_are_read_gated(client, admin_headers,
                                                  employee_headers):
    for path in (
        "/analytics/overview", "/analytics/budget",
        "/analytics/attribution", "/analytics/remediation",
        "/analytics/researchers",
    ):
        assert (
            await client.get(f"{BASE}{path}", headers=admin_headers)
        ).status_code == 200, path
        assert (
            await client.get(f"{BASE}{path}", headers=employee_headers)
        ).status_code == 403, path


@pytest.mark.asyncio
async def test_attribution_rejects_an_unknown_group_by(client, admin_headers):
    resp = await client.get(
        f"{BASE}/analytics/attribution?group_by=astrology", headers=admin_headers,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_ledger_export_streams_xlsx_with_the_money_columns(
    client, admin_headers,
):
    created = await client.post(
        f"{BASE}/reports",
        json={
            "title": "导出用", "researcher_email": "exp@example.com",
            "affected_system": "docs.pingcap.com",
        },
        headers=admin_headers,
    )
    assert created.status_code == 200
    rid = created.json()["data"]["id"]
    await client.post(
        f"{BASE}/reports/{rid}/severity",
        json={"severity": "low", "apply_default_reward": True},
        headers=admin_headers,
    )

    resp = await client.get(
        f"{BASE}/analytics/ledger/export.xlsx", headers=admin_headers,
    )
    assert resp.status_code == 200
    assert "spreadsheet" in resp.headers["content-type"]
    assert resp.headers["X-Export-Total-Matched"] == "1"
    # xlsx is a zip — a truncated/HTML error page would not start PK.
    assert resp.content[:2] == b"PK"


@pytest.mark.asyncio
async def test_analytics_reads_do_not_hide_the_route_collision(
    client, admin_headers,
):
    """``/analytics/...`` must not be shadowed by ``/reports/{report_id}``.

    If the analytics routes had been mounted under ``/reports/...`` the
    UUID path converter on the sibling would swallow them and every call
    would 422 on 'analytics' not being a UUID.
    """
    resp = await client.get(f"{BASE}/analytics/overview", headers=admin_headers)
    assert resp.status_code == 200
    assert "budget" in resp.json()["data"]

# ════════════════════════════════════════════════════════════════════
# Waiting on us
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_awaiting_reply_surfaces_the_silence(db_session):
    """Modelled on the real 2026 correspondence this report exists for: a
    report filed on 04-01 got its first substantive answer on 06-30, after
    the researcher chased twice. Nothing in the console showed that."""
    old = datetime.now(UTC) - timedelta(days=90)
    never = await _report(
        db_session, title="从没回过", reported_at=old,
        researcher="repeat@example.com",
    )
    # A second report from the SAME researcher, answered once and then
    # left alone — a different failure from never answering.
    stale = await _report(
        db_session, title="回过一次就没动了",
        reported_at=datetime.now(UTC) - timedelta(days=40),
        researcher="repeat@example.com",
    )
    await svc.log_reply(
        db_session, stale, template_key=None, subject="Re:",
        body="Thank you, we are evaluating.", actor=None,
    )
    stale.last_replied_at = datetime.now(UTC) - timedelta(days=30)
    await db_session.flush()

    # Answered yesterday — must NOT appear.
    fresh = await _report(
        db_session, title="刚回过",
        reported_at=datetime.now(UTC) - timedelta(days=20),
    )
    await svc.log_reply(
        db_session, fresh, template_key=None, subject="Re:", body="ok",
        actor=None,
    )

    result = await analytics.awaiting_our_reply(db_session)
    codes = [i["code"] for i in result["items"]]
    assert never.code in codes
    assert stale.code in codes
    assert fresh.code not in codes

    totals = result["totals"]
    assert totals["never_replied"] == 1
    assert totals["stale_since_last_reply"] == 1
    # Longest silence first — that is the row that will send the next
    # "do you have any updates?" mail.
    assert result["items"][0]["code"] == never.code
    assert result["items"][0]["waiting_days"] >= 89
    assert totals["worst_waiting_days"] >= 89

    # A repeat submitter going unanswered is worse than a first-timer, so
    # the row carries how much they have sent us.
    row = next(i for i in result["items"] if i["code"] == never.code)
    assert row["researcher_report_count"] == 2
    assert row["never_replied"] is True


@pytest.mark.asyncio
async def test_awaiting_reply_ignores_settled_reports(db_session):
    row = await _report(
        db_session, title="已结案",
        reported_at=datetime.now(UTC) - timedelta(days=200),
    )
    await svc.set_status(
        db_session, row, target=const.STATUS_OUT_OF_SCOPE, note=None,
        actor=None,
    )
    # Nobody is waiting on us for a report we already ruled on and told
    # them about; leaving it in would make the queue permanently red.
    assert (await analytics.awaiting_our_reply(db_session))["items"] == []


@pytest.mark.asyncio
async def test_awaiting_reply_threshold_is_tunable(db_session):
    await _report(
        db_session, title="三天前",
        reported_at=datetime.now(UTC) - timedelta(days=3),
    )
    assert (await analytics.awaiting_our_reply(db_session))["items"] == []
    loosened = await analytics.awaiting_our_reply(db_session, overdue_days=2)
    assert len(loosened["items"]) == 1


@pytest.mark.asyncio
async def test_awaiting_reply_endpoint(client, admin_headers, employee_headers):
    assert (
        await client.get(
            f"{BASE}/analytics/awaiting-reply", headers=admin_headers,
        )
    ).status_code == 200
    assert (
        await client.get(
            f"{BASE}/analytics/awaiting-reply", headers=employee_headers,
        )
    ).status_code == 403

@pytest.mark.asyncio
async def test_on_time_rate_excludes_fixes_that_never_had_a_deadline(db_session):
    """A report fixed without a deadline has no punctuality answer.

    Found 2026-08-24 by running the report over backfilled historical
    correspondence: those rows carry ``fixed_at`` but no ``fix_due_date``
    (nobody set an SLA at the time), and dividing by ``fixed`` reported
    "0% on time" for work that was never given a deadline to miss.
    """
    row = await _report(db_session, title="历史补录")
    # Severity set directly, as a backfill does — no SLA stamped.
    row.severity = "medium"
    row.status = const.STATUS_CONFIRMED
    row.remediation_status = const.REMEDIATION_FIXED
    row.fixed_at = datetime.now(UTC)
    row.fix_due_date = None
    await db_session.flush()

    report = await analytics.remediation_report(db_session)
    item = next(r for r in report["items"] if r["severity"] == "medium")
    assert item["fixed"] == 1
    assert item["measurable"] == 0
    # None, not 0.0 — we cannot say they were late, only that nobody said
    # when it was due.
    assert item["on_time_rate"] is None

    # A second one WITH a deadline, missed, makes the rate meaningful.
    late = await _confirm(
        db_session,
        await _report(
            db_session, title="有时限但迟了",
            reported_at=datetime.now(UTC) - timedelta(days=90),
        ),
    )
    await svc.set_remediation(
        db_session, late, target=const.REMEDIATION_FIXED, note=None,
        fix_eta=None, actor=None,
    )
    report = await analytics.remediation_report(db_session)
    item = next(r for r in report["items"] if r["severity"] == "medium")
    assert item["fixed"] == 2
    assert item["measurable"] == 1
    assert item["on_time_rate"] == pytest.approx(0.0)

# ════════════════════════════════════════════════════════════════════
# Budget tracks (产研 / IT)
# ════════════════════════════════════════════════════════════════════


@pytest.mark.asyncio
async def test_each_track_has_its_own_ceiling_and_burn(db_session):
    """The pots are approved by different owners, so "how much is left"
    is only a real answer per track. A combined bar would average an
    exhausted product budget with an untouched IT one."""
    product = await _confirm(
        db_session, await _report(db_session, title="产研的"), severity="high",
    )
    it = await _report(db_session, title="IT 的")
    it.budget_track = const.BUDGET_TRACK_IT
    await db_session.flush()
    await _confirm(db_session, it, severity="low")

    prod_report = await analytics.budget_report(
        db_session, track=const.BUDGET_TRACK_PRODUCT,
    )
    it_report = await analytics.budget_report(
        db_session, track=const.BUDGET_TRACK_IT,
    )
    combined = await analytics.budget_report(db_session)

    assert prod_report["budget_usd"] == 5000.0
    assert prod_report["committed_usd"] == 800.0
    assert it_report["budget_usd"] == 5000.0
    assert it_report["committed_usd"] == 100.0
    # Combined sums BOTH ceilings, not just one — otherwise the
    # leadership view shows 900/5000 against a 10,000 programme.
    assert combined["budget_usd"] == 10000.0
    assert combined["committed_usd"] == 900.0
    assert product.budget_track == const.BUDGET_TRACK_PRODUCT


@pytest.mark.asyncio
async def test_track_ceilings_are_edited_independently(db_session):
    await cfg.update_budget_config(
        db_session,
        tracks={const.BUDGET_TRACK_IT: {"default_usd": Decimal("2000")}},
        actor_id=None,
    )
    config = await cfg.get_budget_config(db_session)
    # Editing IT must not blank product, which a full-object PUT would.
    assert config["tracks"][const.BUDGET_TRACK_PRODUCT]["default_usd"] == Decimal("5000")
    assert config["tracks"][const.BUDGET_TRACK_IT]["default_usd"] == Decimal("2000")
    current = await cfg.current_fy(db_session)
    assert await cfg.budget_for_fy(
        db_session, current, track=const.BUDGET_TRACK_IT,
    ) == Decimal("2000")
    assert await cfg.budget_for_fy(db_session, current) == Decimal("7000")


@pytest.mark.asyncio
async def test_legacy_flat_budget_setting_reads_as_the_product_track(db_session):
    """An environment whose migration has not run yet must keep its
    configured ceiling rather than silently reverting to the default."""
    from app.integrations.models import PlatformSetting

    row = await db_session.get(PlatformSetting, const.SETTING_BUDGET)
    if row is None:
        row = PlatformSetting(key=const.SETTING_BUDGET, value={})
        db_session.add(row)
    row.value = {"default_usd": 12345, "by_fy": {"2026": 999}, "currency": "USD"}
    await db_session.flush()

    config = await cfg.get_budget_config(db_session)
    assert config["tracks"][const.BUDGET_TRACK_PRODUCT]["default_usd"] == Decimal("12345")
    assert config["tracks"][const.BUDGET_TRACK_PRODUCT]["by_fy"][2026] == Decimal("999")
    # The pot that did not exist in the old shape falls back to the ship
    # default rather than to zero.
    assert config["tracks"][const.BUDGET_TRACK_IT]["default_usd"] == Decimal(
        const.DEFAULT_BUDGET_USD,
    )


@pytest.mark.asyncio
async def test_moving_a_report_between_tracks_is_logged(db_session):
    row = await _confirm(db_session, await _report(db_session))
    await svc.set_payment_status(
        db_session, row, target=const.PAYMENT_PAID, reference=None, note=None,
        paid_at=None, actor=None,
    )
    await svc.set_budget_track(
        db_session, row, track=const.BUDGET_TRACK_IT,
        note="财务更正:这套系统归 IT", actor=None,
    )
    assert row.budget_track == const.BUDGET_TRACK_IT

    events = await svc.list_events(db_session, row.id, internal=True)
    move = [e for e in events if (e.meta or {}).get("field") == "budget_track"]
    assert len(move) == 1
    # Allowed after payment — which budget an already-settled reward was
    # charged to is a bookkeeping fact finance sometimes corrects — but the
    # event says so out loud.
    assert move[0].meta["already_paid"] is True
    assert move[0].meta["to"] == const.BUDGET_TRACK_IT


@pytest.mark.asyncio
async def test_by_track_rollup_lists_an_empty_pot(db_session):
    await _confirm(db_session, await _report(db_session))
    rollup = await analytics.attribution_by_track(db_session)
    tracks = {row["track"]: row for row in rollup["items"]}
    assert set(tracks) == set(const.BUDGET_TRACKS)
    assert tracks[const.BUDGET_TRACK_PRODUCT]["total"] == 1
    # A pot with nothing in it is a fact worth showing; a missing row
    # would read as "no such budget".
    assert tracks[const.BUDGET_TRACK_IT]["total"] == 0
    assert tracks[const.BUDGET_TRACK_IT]["reward_usd"] == 0.0


@pytest.mark.asyncio
async def test_unknown_track_is_rejected_everywhere(db_session):
    with pytest.raises(ValidationError):
        await _report(db_session, budget_track="marketing")
    row = await _report(db_session)
    with pytest.raises(ValidationError):
        await svc.set_budget_track(
            db_session, row, track="marketing", note=None, actor=None,
        )
    with pytest.raises(ValidationError):
        await cfg.update_budget_config(
            db_session, tracks={"marketing": {"default_usd": Decimal("1")}},
            actor_id=None,
        )


@pytest.mark.asyncio
async def test_fy_label_reads_end_year(db_session):
    """The company says FY27 for 2026-04 ~ 2027-03; the report must not
    say FY2026 next to a number people call FY27."""
    report = await analytics.budget_report(db_session, fiscal_year=2026)
    assert report["fiscal_year"] == 2026
    assert report["fiscal_year_label"].startswith("FY27 ")
    assert report["fiscal_year_short"] == "FY27"


@pytest.mark.asyncio
async def test_track_endpoints(client, admin_headers, employee_headers):
    assert (
        await client.get(f"{BASE}/analytics/budget-tracks", headers=admin_headers)
    ).status_code == 200
    assert (
        await client.get(
            f"{BASE}/analytics/budget-tracks", headers=employee_headers,
        )
    ).status_code == 403
    assert (
        await client.get(
            f"{BASE}/analytics/budget?track=it", headers=admin_headers,
        )
    ).status_code == 200
    assert (
        await client.get(
            f"{BASE}/analytics/budget?track=marketing", headers=admin_headers,
        )
    ).status_code == 422


# ── The matrix ──────────────────────────────────────────────────────────
# Landing page rebuild, 2026-08-25. The flat table had no hierarchy because
# the data has none: 114 of 149 rows share one status, 142 one empty
# severity, 84 one hostname. Rows = whose problem, columns = what it waits
# on, heat = how long somebody has been waiting.

@pytest.mark.asyncio
async def test_matrix_bands_rows_by_track_and_counts_stages(db_session):
    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc
    from app.domains.system_registry import service as reg_svc

    owned = await reg_svc.create_system(
        db_session, name="Owned Thing", slug="owned-thing",
        owner_person_id=None, budget_track="it",
    )
    linked = await svc.create_report(
        db_session, title="linked and silent",
        researcher_email="a@example.com", system_id=owned.id, actor=None,
    )
    await svc.create_report(
        db_session, title="names a host only",
        researcher_email="b@example.com",
        affected_system="unregistered.example.com", actor=None,
    )
    await svc.create_report(
        db_session, title="names nothing",
        researcher_email="c@example.com", actor=None,
    )
    await db_session.commit()

    out = await analytics.matrix(db_session)
    assert out["stages"][0] == "need_reply"
    by_key = {r["key"]: r for r in out["rows"]}

    # Linked → grouped under its registry system, in that system's band.
    system_row = by_key[f"sys:{owned.id}"]
    assert system_row["track"] == "it"
    assert system_row["registered"] is True
    assert system_row["cells"]["need_reply"]["count"] == 1
    # 0, not 1: a report arrives at the lowest tier (2026-08-26), and the
    # arrival default is not outstanding work — counting it here recreated
    # the wall of 142 the default exists to clear. The cell is still
    # possible, it is simply empty.
    assert system_row["cells"]["need_grade"]["count"] == 0

    # A hostname nobody registered stays visible AS a hostname rather than
    # dissolving into a catch-all.
    host_row = by_key["host:unregistered.example.com"]
    assert host_row["registered"] is False

    # Naming nothing gets its own band: structurally unattributable, and
    # pretending otherwise is what made the old table unreadable.
    assert "none:unattributed" in by_key
    assert by_key["none:unattributed"]["track"] == "unattributed"

    # Bands are ordered product → it → unattributed.
    assert [b["track"] for b in out["bands"]] == [
        "product", "it", "unattributed",
    ]
    assert out["totals"]["need_reply"] >= 3
    assert linked.code


@pytest.mark.asyncio
async def test_assign_column_is_impossible_not_zero_without_an_owner(
    db_session,
):
    """A cell that cannot exist is not the same as a cell that is empty.
    Rendering "assign to the system owner" as 0 on an ownerless system
    invites the reader to conclude the work is done."""
    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc
    from app.domains.system_registry import service as reg_svc
    from app.identity.person_models import IdentityPerson

    person = IdentityPerson(
        source_system="test", hris_person_id="MX1", employee_no="MX1",
        work_email="mx1@x.test", display_name="Owner One",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    ownerless = await reg_svc.create_system(
        db_session, name="No Owner", slug="mx-no-owner", owner_person_id=None,
    )
    owned = await reg_svc.create_system(
        db_session, name="Has Owner", slug="mx-has-owner",
        owner_person_id=person.id,
    )
    await svc.create_report(
        db_session, title="on the ownerless one",
        researcher_email="d@example.com", system_id=ownerless.id, actor=None,
    )
    await svc.create_report(
        db_session, title="on the owned one",
        researcher_email="e@example.com", system_id=owned.id, actor=None,
    )
    await db_session.commit()

    out = await analytics.matrix(db_session)
    by_key = {r["key"]: r for r in out["rows"]}
    assert by_key[f"sys:{ownerless.id}"]["cells"]["need_assign"]["na"] is True
    assert by_key[f"sys:{owned.id}"]["cells"]["need_assign"]["na"] is False
    # A real 0, and the distinction the test is about: the owned system's
    # cell CAN hold work and currently holds none, because a report on an
    # owned system routes itself the moment it is linked (2026-08-26). The
    # ownerless one is hatched because that work cannot exist at all.
    assert by_key[f"sys:{owned.id}"]["cells"]["need_assign"]["count"] == 0


@pytest.mark.asyncio
async def test_matrix_heat_buckets_the_oldest_wait_in_each_cell(db_session):
    """The heat is the OLDEST item in the cell, not the average: one report
    silent for 200 days inside a cell of twenty is the fact worth seeing,
    and an average would hide it."""
    from datetime import UTC, datetime, timedelta

    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc
    from app.domains.system_registry import service as reg_svc

    system = await reg_svc.create_system(
        db_session, name="Heat Test", slug="heat-test", owner_person_id=None,
    )
    now = datetime.now(UTC)
    for days in (2, 200):
        await svc.create_report(
            db_session, title=f"silent {days}",
            researcher_email=f"h{days}@example.com", system_id=system.id,
            reported_at=now - timedelta(days=days), actor=None,
        )
    await db_session.commit()

    out = await analytics.matrix(db_session)
    row = next(r for r in out["rows"] if r["key"] == f"sys:{system.id}")
    cell = row["cells"]["need_reply"]
    assert cell["count"] == 2
    assert cell["oldest_days"] >= 200
    # 200 days lands in the top bucket (edges 7/30/90/180).
    assert cell["heat"] == 4
    assert out["heat_edges"] == [7, 30, 90, 180]


@pytest.mark.asyncio
async def test_the_assign_column_total_agrees_with_its_cells(db_session):
    """A column header that disagrees with the cells beneath it is how a
    dashboard stops being believed. An impossible cell must not contribute
    to its column total: 84 reports on an ownerless system are rendered as
    "cannot assign", so counting them in "needs assigning" would show 84
    over a column of hatching."""
    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc
    from app.domains.system_registry import service as reg_svc
    from app.identity.person_models import IdentityPerson

    person = IdentityPerson(
        source_system="test", hris_person_id="AG1", employee_no="AG1",
        work_email="ag1@x.test", display_name="Active Owner",
        employment_status="active",
    )
    db_session.add(person)
    await db_session.flush()

    ownerless = await reg_svc.create_system(
        db_session, name="Agree No Owner", slug="agree-no-owner",
        owner_person_id=None,
    )
    owned = await reg_svc.create_system(
        db_session, name="Agree Owned", slug="agree-owned",
        owner_person_id=person.id,
    )
    for _ in range(3):
        await svc.create_report(
            db_session, title="on the ownerless system",
            researcher_email=f"n{_}@example.com", system_id=ownerless.id,
            actor=None,
        )
    routed = await svc.create_report(
        db_session, title="on the owned system",
        researcher_email="o1@example.com", system_id=owned.id, actor=None,
    )
    # It routes itself on link (2026-08-26). Put it back in the column so
    # the invariant below is checked against a real non-zero cell —
    # "total == sum of cells" is vacuous when every cell is 0.
    await svc.unassign_report(db_session, routed, reason=None, actor=None)
    await db_session.commit()

    out = await analytics.matrix(db_session)
    visible = sum(
        row["cells"]["need_assign"]["count"]
        for row in out["rows"]
        if not row["cells"]["need_assign"]["na"]
    )
    assert out["totals"]["need_assign"] == visible, (
        "the column total must equal the sum of the cells that exist"
    )
    by_key = {r["key"]: r for r in out["rows"]}
    assert by_key[f"sys:{ownerless.id}"]["cells"]["need_assign"]["count"] == 0
    assert by_key[f"sys:{owned.id}"]["cells"]["need_assign"]["count"] == 1


# ── The lane view (matrix drill-down) ───────────────────────────────────

@pytest.mark.asyncio
async def test_lane_shares_the_matrix_stage_definitions_exactly(db_session):
    """Two definitions of "which stage is this report in" is how the two
    views start disagreeing. They share the module-level predicate, and
    this asserts the observable consequence: for one row, every stage
    count matches the matrix cell."""
    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc
    from app.domains.system_registry import service as reg_svc

    system = await reg_svc.create_system(
        db_session, name="Lane Parity", slug="lane-parity",
        owner_person_id=None,
    )
    for i in range(4):
        await svc.create_report(
            db_session, title=f"lane report {i}",
            researcher_email=f"lane{i}@example.com", system_id=system.id,
            actor=None,
        )
    await db_session.commit()

    key = f"sys:{system.id}"
    grid = await analytics.matrix(db_session)
    row = next(r for r in grid["rows"] if r["key"] == key)
    out = await analytics.lane(db_session, row_key=key)

    assert out["row"]["key"] == key
    for stage in grid["stages"]:
        assert out["stages"][stage]["count"] == row["cells"][stage]["count"], (
            f"lane and matrix disagree on {stage}"
        )
    # The impossible cell is impossible in both views.
    assert out["stages"]["need_assign"].get("na") is True
    assert row["cells"]["need_assign"]["na"] is True


@pytest.mark.asyncio
async def test_lane_shows_oldest_first_and_says_what_it_hid(db_session):
    """A capped list that does not say it is capped reads as "this is
    everything" — the same defect the matrix drill-down had."""
    from datetime import UTC, datetime, timedelta

    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc
    from app.domains.system_registry import service as reg_svc

    system = await reg_svc.create_system(
        db_session, name="Lane Cap", slug="lane-cap", owner_person_id=None,
    )
    now = datetime.now(UTC)
    for days in (5, 100, 250, 12):
        await svc.create_report(
            db_session, title=f"aged {days}",
            researcher_email=f"cap{days}@example.com", system_id=system.id,
            reported_at=now - timedelta(days=days), actor=None,
        )
    await db_session.commit()

    out = await analytics.lane(
        db_session, row_key=f"sys:{system.id}", per_stage=2,
    )
    bucket = out["stages"]["need_reply"]
    assert bucket["count"] == 4
    assert len(bucket["items"]) == 2
    assert bucket["hidden"] == 2, "a cap must say how much it hid"
    ages = [i["days_silent"] for i in bucket["items"]]
    assert ages == sorted(ages, reverse=True), "oldest silence first"
    assert ages[0] >= 250


@pytest.mark.asyncio
async def test_lane_for_an_empty_row_is_not_an_error(db_session):
    """The operator may have just cleared the last report in a row; that is
    a success, not a 500."""
    from app.domains.bounty import analytics

    out = await analytics.lane(db_session, row_key="sys:does-not-exist")
    assert out["row"] is None
    assert all(b["count"] == 0 for b in out["stages"].values())


@pytest.mark.asyncio
async def test_lane_matches_the_matrix_for_every_row_shape(db_session):
    """Parity has to hold for all three key shapes, not just ``sys:``.

    The lane narrows in SQL for speed; a predicate that is subtly too
    narrow silently drops reports from one column. The first version of
    this test only covered a registry-linked row, and a deliberately
    broken ``host:`` predicate passed it — so it now walks every row the
    matrix produced.
    """
    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc
    from app.domains.system_registry import service as reg_svc

    system = await reg_svc.create_system(
        db_session, name="Shape Parity", slug="shape-parity",
        owner_person_id=None,
    )
    await svc.create_report(
        db_session, title="linked", researcher_email="s1@example.com",
        system_id=system.id, actor=None,
    )
    # A hostname nobody registered.
    for i in range(2):
        await svc.create_report(
            db_session, title=f"host only {i}",
            researcher_email=f"h{i}@example.com",
            affected_system="Shape.Example.COM", actor=None,
        )
    # And one naming nothing at all.
    await svc.create_report(
        db_session, title="nothing named", researcher_email="n1@example.com",
        actor=None,
    )
    await db_session.commit()

    grid = await analytics.matrix(db_session)
    assert len(grid["rows"]) >= 3

    for row in grid["rows"]:
        out = await analytics.lane(db_session, row_key=row["key"])
        assert out["row"] is not None, row["key"]
        assert out["row"]["key"] == row["key"]
        for stage in grid["stages"]:
            assert (
                out["stages"][stage]["count"] == row["cells"][stage]["count"]
            ), f"{row['key']} disagrees on {stage}"

    # The hostname row is keyed lower-case by the matrix while the stored
    # value here is mixed case. Note for future readers: this particular
    # assertion is NOT mutation-provable on MySQL — the default collation
    # is already case-insensitive, so dropping the explicit ``lower()``
    # still passes. The ``lower()`` stays for portability and because the
    # key really is lower-cased; the load-bearing part of this test is the
    # per-shape parity loop above.
    host_row = next(
        r for r in grid["rows"] if r["key"] == "host:shape.example.com"
    )
    assert host_row["total"] == 2
    lane_out = await analytics.lane(db_session, row_key=host_row["key"])
    assert lane_out["stages"]["need_reply"]["count"] == 2


@pytest.mark.asyncio
async def test_lane_rejects_a_malformed_row_key_without_scanning(db_session):
    """``sys:not-a-uuid`` reaches this from a hand-edited URL. It must be an
    empty answer, not a 500 and not a full-table scan."""
    from app.domains.bounty import analytics

    out = await analytics.lane(db_session, row_key="sys:not-a-uuid")
    assert out["row"] is None
    assert all(b["count"] == 0 for b in out["stages"].values())


@pytest.mark.asyncio
async def test_no_live_report_can_vanish_from_the_matrix(db_session):
    """Every live report lands in exactly one row, whatever state its link
    is in.

    The matrix is the console's default view and the number it prints is
    the number the operator trusts. A report that belongs to no row is
    invisible work — worse than a wrong total, because nothing on screen
    says anything is missing.

    The case that would produce one is a report linked to a system that has
    since been RETIRED. Retiring is status-only (the row stays readable so
    a historical finding still resolves to a name), so the link survives
    while the system drops out of the live set. `_bucket_for` therefore has
    to fall through to the hostname — the obvious edit, indexing
    `by_system_id` on `system_id` unconditionally, either raises or drops
    the report.
    """
    from app.domains.system_registry import service as reg_svc

    doomed = await reg_svc.create_system(
        db_session, name="Soon Retired", slug="soon-retired",
        owner_person_id=None, budget_track="it",
        hostnames=["retired.example.com"],
    )
    linked_then_orphaned = await svc.create_report(
        db_session, title="linked to a system that gets retired",
        researcher_email="v1@example.com", system_id=doomed.id,
        affected_system="retired.example.com", actor=None,
    )
    # Same, but naming no host at all — the harder half: there is no
    # hostname to fall back to, so it has to reach the unattributed band.
    linked_no_host = await svc.create_report(
        db_session, title="linked, retired, and names nothing",
        researcher_email="v2@example.com", system_id=doomed.id, actor=None,
    )
    plain_host = await svc.create_report(
        db_session, title="names a host only",
        researcher_email="v3@example.com",
        affected_system="unregistered.example.com", actor=None,
    )
    nothing = await svc.create_report(
        db_session, title="names nothing",
        researcher_email="v4@example.com", actor=None,
    )
    await db_session.flush()

    live_codes = {
        linked_then_orphaned.code, linked_no_host.code,
        plain_host.code, nothing.code,
    }

    before = await analytics.matrix(db_session)
    assert sum(row["total"] for row in before["rows"]) >= len(live_codes)

    await reg_svc.retire_system(db_session, doomed.id, actor=None)
    await db_session.flush()

    after = await analytics.matrix(db_session)
    assert not any(
        row["system_id"] == str(doomed.id) for row in after["rows"]
    ), "a retired system must not keep an ownership row of its own"

    # Every one of the four is still somewhere, and each in exactly one row.
    seen: dict[str, list[str]] = {}
    for row in after["rows"]:
        lane = await analytics.lane(db_session, row_key=row["key"])
        for stage in lane["order"]:
            for item in lane["stages"][stage].get("items") or []:
                seen.setdefault(item["code"], [])
                if row["key"] not in seen[item["code"]]:
                    seen[item["code"]].append(row["key"])

    for code in live_codes:
        assert code in seen, f"{code} appears in no matrix row at all"
        assert len(seen[code]) == 1, (
            f"{code} is in {seen[code]} — a report counted under two rows "
            "double-counts the whole view"
        )

    # And the row it fell into is the honest one.
    assert seen[linked_then_orphaned.code][0] == "host:retired.example.com"
    assert seen[linked_no_host.code][0] == "none:unattributed"

    # The grand total did not move: retiring a system reassigns rows, it
    # does not make findings disappear.
    assert (
        sum(row["total"] for row in after["rows"])
        == sum(row["total"] for row in before["rows"])
    )


@pytest.mark.asyncio
async def test_a_padded_hostname_does_not_split_or_vanish(db_session):
    """`affected_system` is the matrix's grouping key AND arrives by paste
    out of a researcher's mail, and nothing used to strip it.

    Two failures came out of that. `_bucket_for` strips, so
    " www.pingcap.com " grouped under the same ownership row as the clean
    value — but `lane()`'s SQL narrowing (added in this same review) did
    not, so the row showed a count the drill-down could not produce a
    single card for. Normalising on write fixes it going forward; TRIM in
    the lane predicate covers rows written before that.
    """
    clean = await svc.create_report(
        db_session, title="clean host", researcher_email="c1@example.com",
        affected_system="padded.example.com", actor=None,
    )
    padded = await svc.create_report(
        db_session, title="padded host", researcher_email="c2@example.com",
        affected_system="  Padded.example.com  ", actor=None,
    )
    await db_session.flush()

    # Normalised on the way in — one grouping key, not two.
    assert padded.affected_system == "Padded.example.com"

    rows = (await analytics.matrix(db_session))["rows"]
    host_rows = [
        r for r in rows if (r["key"] or "").startswith("host:padded.example")
    ]
    assert len(host_rows) == 1, (
        f"a padded value must not open a second ownership row: {host_rows}"
    )
    assert host_rows[0]["total"] == 2

    lane = await analytics.lane(db_session, row_key=host_rows[0]["key"])
    seen = {
        item["code"]
        for stage in lane["order"]
        for item in (lane["stages"][stage].get("items") or [])
    }
    assert {clean.code, padded.code} <= seen, (
        "the lane must produce a card for every report its matrix row "
        f"counted; missing {({clean.code, padded.code}) - seen}"
    )


@pytest.mark.asyncio
async def test_a_legacy_padded_row_still_reaches_its_lane(db_session):
    """The rows written before the strip existed. Set the column directly,
    bypassing the service, because that is the state the database is
    actually in — a test that writes through the now-normalising service
    could not reproduce it."""
    from app.domains.bounty.models import BountyReport

    report = await svc.create_report(
        db_session, title="legacy padded", researcher_email="c3@example.com",
        affected_system="legacy.example.com", actor=None,
    )
    await db_session.flush()
    await db_session.execute(
        BountyReport.__table__.update()
        .where(BountyReport.id == report.id)
        .values(affected_system="  legacy.example.com  "),
    )
    await db_session.flush()
    fresh = await db_session.get(
        BountyReport, report.id, populate_existing=True,
    )
    assert fresh.affected_system == "  legacy.example.com  "

    rows = (await analytics.matrix(db_session))["rows"]
    row = next(
        r for r in rows if (r["key"] or "") == "host:legacy.example.com"
    )
    lane = await analytics.lane(db_session, row_key=row["key"])
    seen = {
        item["code"]
        for stage in lane["order"]
        for item in (lane["stages"][stage].get("items") or [])
    }
    assert report.code in seen, (
        "a legacy padded row is counted by the matrix, so its lane has to "
        "be able to show it"
    )