"""The lifecycle break the platform owner found by opening 我负责的系统.

Linking a report to a registry entry records who SHOULD answer, and then
nothing routes it. In production that meant a CONFIRMED TiCard finding —
fix owed, $100 pending — sat unassigned while the system's owner had
never been told, and his lead had no surface that would show it either.

Two reads close it, from the two sides:
  * ``analytics.routable_unassigned`` — the security desk's queue of
    "we knew who to ask and did not ask".
  * ``/api/v1/me/team/systems`` — the lead's view of what their reports
    owe.
"""
from __future__ import annotations

import uuid

import pytest

from app.domains.bounty import analytics
from app.domains.bounty import service as bounty_svc
from app.domains.system_registry import service as svc
from app.identity.person_models import IdentityPerson

pytestmark = pytest.mark.asyncio


async def _person(db, *, manager=None, status="active") -> IdentityPerson:
    p = IdentityPerson(
        source_system="test", hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        employee_no=f"E{uuid.uuid4().hex[:8]}",
        work_email=f"u{uuid.uuid4().hex[:8]}@x.test",
        display_name=f"P{uuid.uuid4().hex[:4]}", employment_status=status,
        manager_person_id=manager.id if manager else None,
    )
    db.add(p)
    await db.flush()
    return p


async def _system(db, owner):
    return await svc.create_system(
        db, name=f"Sys {uuid.uuid4().hex[:5]}",
        slug=f"sys-{uuid.uuid4().hex[:8]}",
        owner_person_id=owner.id if owner else None,
    )


async def _linked_report(db, system, *, status="new"):
    report = await bounty_svc.create_report(
        db, title=f"finding {uuid.uuid4().hex[:6]}",
        researcher_email=f"w{uuid.uuid4().hex[:6]}@example.com",
        system_id=system.id, actor=None,
    )
    if status != "new":
        report.status = status
        await db.flush()
    return report


async def test_linking_a_report_to_an_owned_system_routes_it_immediately(
    db_session,
):
    """「系统都有负责人了,为什么还需要指派一下」 (2026-08-26).

    This queue used to be the point: reports we knew the owner of and had
    not asked. Making a person press 指派 on each one was a click with only
    one possible outcome, and 15 reports sat in 待指派 because of it — so
    linking now routes, and the queue stops filling from this direction.
    """
    owner = await _person(db_session)
    system = await _system(db_session, owner)
    report = await _linked_report(db_session, system)

    assert str(report.assignee_person_id) == str(owner.id), (
        "the register already answers who is accountable; asking again "
        "adds nothing"
    )
    result = await analytics.routable_unassigned(db_session)
    assert report.code not in {i["code"] for i in result["items"]}


async def test_the_queue_still_catches_a_report_left_unassigned(db_session):
    """Auto-routing only fires when a report is created or linked, so the
    queue is still the backlog's home: reports linked before it existed, and
    any report a person deliberately unassigns."""
    owner = await _person(db_session)
    system = await _system(db_session, owner)
    report = await _linked_report(db_session, system)

    await bounty_svc.unassign_report(
        db_session, report, reason=None, actor=None,
    )
    await db_session.flush()

    result = await analytics.routable_unassigned(db_session)
    hit = next(
        (i for i in result["items"] if i["code"] == report.code), None,
    )
    assert hit is not None, "we know who to ask — it must be in the queue"
    assert hit["owner_person_id"] == str(owner.id)
    assert hit["system_name"] == system.name


async def test_assigning_it_removes_it_from_the_queue(db_session):
    owner = await _person(db_session)
    system = await _system(db_session, owner)
    report = await _linked_report(db_session, system)

    loaded = await bounty_svc.load_assignable_person(db_session, owner.id)
    await bounty_svc.assign_report(
        db_session, report, person=loaded, note=None, actor=None,
    )
    result = await analytics.routable_unassigned(db_session)
    assert report.code not in {i["code"] for i in result["items"]}


async def test_a_departed_owner_is_not_routable(db_session):
    """Routing a finding at a leaver produces silence, not a queue — the
    same reason the registry's own suggestion skips them."""
    owner = await _person(db_session)
    system = await _system(db_session, owner)
    report = await _linked_report(db_session, system)
    owner.employment_status = "leaver_pending"
    await db_session.flush()

    result = await analytics.routable_unassigned(db_session)
    assert report.code not in {i["code"] for i in result["items"]}


async def test_confirmed_findings_sort_first(db_session):
    """A confirmed finding owes a fix, so silence there costs more than
    on an untriaged submission."""
    owner = await _person(db_session)
    system = await _system(db_session, owner)
    plain = await _linked_report(db_session, system)
    confirmed = await _linked_report(db_session, system, status="confirmed")
    # Both arrive routed now, so put them back in the queue — this test is
    # about the ORDER of the backlog, not about how it fills.
    for row in (plain, confirmed):
        await bounty_svc.unassign_report(
            db_session, row, reason=None, actor=None,
        )
    await db_session.flush()

    result = await analytics.routable_unassigned(db_session)
    mine = [i for i in result["items"] if i["system_slug"] == system.slug]
    assert mine[0]["code"] == confirmed.code
    assert result["totals"]["confirmed"] >= 1


async def test_a_lead_sees_what_their_reports_owe(
    client, db_session, employee_user, employee_headers,
):
    lead, _token = employee_user
    direct = await _person(db_session, manager=lead)
    skip_level = await _person(db_session, manager=direct)
    stranger = await _person(db_session)

    theirs = await _system(db_session, direct)
    deep = await _system(db_session, skip_level)
    unrelated = await _system(db_session, stranger)
    await _linked_report(db_session, theirs, status="confirmed")
    await db_session.commit()

    data = (
        await client.get("/api/v1/me/team/systems", headers=employee_headers)
    ).json()["data"]
    slugs = {i["slug"] for i in data["items"]}
    assert theirs.slug in slugs
    # The whole subtree: a system owned two levels down is still the
    # lead's problem when it goes unanswered.
    assert deep.slug in slugs
    assert unrelated.slug not in slugs

    # Which of my people owns it is the reason this page exists — an
    # empty owner column would defeat it.
    owed = next(i for i in data["items"] if i["slug"] == theirs.slug)
    assert owed["owner_name"] == (direct.display_name or direct.work_email)
    assert len(owed["findings"]) == 1
    assert owed["findings"][0]["status"] == "confirmed"
    # A lead is shown what their team owes — not who reported it or what
    # it cost.
    assert "researcher_email" not in owed["findings"][0]
    assert "reward_amount_usd" not in owed["findings"][0]


async def test_someone_with_no_reports_sees_an_empty_team_view(
    client, employee_headers,
):
    data = (
        await client.get("/api/v1/me/team/systems", headers=employee_headers)
    ).json()["data"]
    assert data["totals"]["systems"] == 0


# ── Drilling into one finding ───────────────────────────────────────────
# The lead view answered "what does my team owe" and then dead-ended:
# three lines of red text with nothing behind them. These cover the
# read-only detail, and — more importantly — what it must NOT carry.

async def test_a_lead_can_open_one_finding_on_their_team_s_system(
    client, db_session, employee_user, employee_headers,
):
    lead, _token = employee_user
    direct = await _person(db_session, manager=lead)
    system = await _system(db_session, direct)
    report = await _linked_report(db_session, system, status="confirmed")
    await db_session.commit()

    res = await client.get(
        f"/api/v1/me/team/systems/findings/{report.code}",
        headers=employee_headers,
    )
    assert res.status_code == 200
    data = res.json()["data"]
    assert data["code"] == report.code
    assert data["status"] == "confirmed"
    assert data["system_name"] == system.name
    assert data["owner_name"] == (direct.display_name or direct.work_email)
    # The clock is the point: how long has this been sitting there.
    assert data["days_open"] is not None
    assert data["days_open"] >= 0
    # A lifecycle trail, so "nothing has happened since April" is visible.
    assert any(e["event_type"] == "created" for e in data["timeline"])


async def test_the_detail_withholds_the_researcher_the_reward_and_the_summary(
    client, db_session, employee_user, employee_headers,
):
    """The redaction is the feature. ``summary`` is named explicitly
    because in the real ledger it carries payment state in free text
    (「已向研究者索取发票,尚未付款」), which is exactly what a lead must
    not be shown — a blocklist would have let it through."""
    from decimal import Decimal

    from app.domains.bounty import config_service as bounty_cfg

    lead, _token = employee_user
    direct = await _person(db_session, manager=lead)
    system = await _system(db_session, direct)
    report = await _linked_report(db_session, system, status="confirmed")
    # A severity MUST be set here: it is what makes the code look up the
    # tier row, and that row also holds the reward columns. Without it
    # severity_label is None and the assertion below vacuously passes —
    # mutation testing caught exactly that.
    # Seed the tier catalogue the way the app does, so the coordinator's
    # tier lookup actually resolves — the reward columns it must not copy
    # only exist once these rows do.
    await bounty_cfg.list_tiers(db_session)
    # Through the real producer, not by assigning the column: every report
    # arrives carrying the lowest tier with ``severity_is_default`` set, and
    # the coordinator reports no level at all while that flag stands. Only
    # ``set_severity`` clears it, so a hand-assigned ``report.severity``
    # would leave this test asserting the label of a report nobody graded.
    await bounty_svc.set_severity(
        db_session, report, severity="low", note=None,
        apply_default_reward=False, actor=None,
    )
    report.summary = "已向研究者索取发票,尚未付款。"
    report.reward_amount_usd = Decimal("100.00")
    report.payment_status = "pending"
    await db_session.flush()
    # A free-text event of a type the whitelist excludes.
    await bounty_svc.add_event(
        db_session, report=report, event_type="comment", actor=None,
        actor_kind="system", body="邮件答复原文摘录:reward agreed, $100",
        visibility="internal",
    )
    await db_session.commit()

    data = (
        await client.get(
            f"/api/v1/me/team/systems/findings/{report.code}",
            headers=employee_headers,
        )
    ).json()["data"]

    for forbidden in (
        "summary", "reward_amount_usd", "payment_status", "researcher_email",
        "researcher_id", "researcher",
    ):
        assert forbidden not in data, f"{forbidden} leaked to a lead"
    # The severity label is copied off the tier row, and that row also
    # holds default_reward_usd / max_reward_usd. Only the three labels
    # and the colour may come across.
    assert data["severity_label"], "tier lookup produced no label to check"
    assert set(data["severity_label"]) == {"zh", "en", "ja", "color"}
    # Bodies never travel, and the excluded type is absent outright.
    blob = str(data)
    assert "发票" not in blob
    assert "reward agreed" not in blob
    assert all(e["event_type"] != "comment" for e in data["timeline"])
    assert all("body" not in e for e in data["timeline"])


async def test_the_detail_carries_the_technical_substance(
    client, db_session, employee_user, employee_headers,
):
    """The other half of the boundary: what a lead MUST be given.

    The withholding test above is only half a specification, and for a
    while it was the only half — so the payload shrank to a clock and a
    trail, and the people accountable for the system could read that a
    finding existed without ever reading what it was. 「没有详细信息 你让他
    怎么确认 怎么跟进 怎么回复?」
    """
    lead, _token = employee_user
    direct = await _person(db_session, manager=lead)
    system = await _system(db_session, direct)
    report = await _linked_report(db_session, system, status="confirmed")
    report.raw_report = "GET /admin/attendees returns all 7,735 rows"
    report.reproduction = "1. curl the endpoint with no token\n2. read the JSON"
    report.affected_urls = ["https://ticard.pingcap.com/api/admin/attendees"]
    report.vuln_type = "broken_access_control"
    report.ai_assessment = {
        "verdict": "likely_valid",
        "reasoning": "An unauthenticated admin endpoint returning PII.",
        "checklist": ["Call the endpoint without a token"],
    }
    await db_session.flush()
    await db_session.commit()

    data = (
        await client.get(
            f"/api/v1/me/team/systems/findings/{report.code}",
            headers=employee_headers,
        )
    ).json()["data"]

    assert "7,735" in (data["raw_report"] or ""), "the write-up must arrive"
    assert "curl the endpoint" in (data["reproduction"] or "")
    assert data["affected_urls"] == [
        "https://ticard.pingcap.com/api/admin/attendees",
    ]
    assert data["vuln_type"] == "broken_access_control"
    assert data["ai_assessment"]["verdict"] == "likely_valid"
    assert data["has_write_up"] is True


async def test_an_arrival_level_is_not_reported_as_a_level(
    client, db_session, employee_user, employee_headers,
):
    """A tier nobody chose must not reach an owner or a lead as a tier.

    Every report now arrives carrying the lowest enabled tier so that no
    queue shows 未定级. In production that wrote ``informational`` onto 142
    reports — and this endpoint rendered it as 「告知」 on 「Unauthenticated
    TiCard admin API leaks 7,735 attendees' full PII」, to the very person
    being asked to fix it. A placeholder displayed as a judgement is worse
    than no answer, because it reads as a judgement someone made.
    """
    from app.domains.bounty import config_service as bounty_cfg

    lead, _token = employee_user
    direct = await _person(db_session, manager=lead)
    system = await _system(db_session, direct)
    await bounty_cfg.list_tiers(db_session)
    report = await _linked_report(db_session, system, status="confirmed")
    await db_session.commit()

    # Precondition: the report really is carrying an arrival level, so the
    # assertions below are not passing on an empty column.
    assert report.severity is not None, "no arrival default was applied"
    assert report.severity_is_default is True

    data = (
        await client.get(
            f"/api/v1/me/team/systems/findings/{report.code}",
            headers=employee_headers,
        )
    ).json()["data"]

    assert data["severity"] is None, (
        f"reported {data['severity']!r} as this finding's level, "
        "but nobody has graded it"
    )
    assert data["severity_label"] is None
    assert data["severity_is_default"] is True

    # And once a person grades it, the same endpoint says so.
    await bounty_svc.set_severity(
        db_session, report, severity="high", note=None,
        apply_default_reward=False, actor=None,
    )
    await db_session.commit()
    graded = (
        await client.get(
            f"/api/v1/me/team/systems/findings/{report.code}",
            headers=employee_headers,
        )
    ).json()["data"]
    assert graded["severity"] == "high"
    assert graded["severity_is_default"] is False
    assert graded["severity_label"], "a graded level must carry its label"


async def test_a_finding_on_someone_else_s_system_is_not_found(
    client, db_session, employee_user, employee_headers,
):
    """404, not 403: a lead must not be able to probe whether a report
    code exists by reading the status line.

    The lead here owns a visible system of their own, and that is the
    point. An earlier version of this test gave them a report who owned
    nothing, so the allowed set was empty and the endpoint 404'd on the
    empty-set early return — it passed with the scope filter deleted.
    Mutation testing caught it. The lead must have something visible for
    the filter itself to be under test.
    """
    lead, _token = employee_user
    mine = await _person(db_session, manager=lead)
    visible = await _system(db_session, mine)
    await _linked_report(db_session, visible, status="confirmed")

    stranger = await _person(db_session)
    theirs = await _system(db_session, stranger)
    report = await _linked_report(db_session, theirs, status="confirmed")
    await db_session.commit()

    # Sanity: this lead really can read their own side.
    assert (await client.get(
        "/api/v1/me/team/systems", headers=employee_headers,
    )).json()["data"]["totals"]["systems"] == 1

    res = await client.get(
        f"/api/v1/me/team/systems/findings/{report.code}",
        headers=employee_headers,
    )
    assert res.status_code == 404


async def test_a_lead_with_no_reports_cannot_open_any_finding(
    client, db_session, employee_headers,
):
    stranger = await _person(db_session)
    system = await _system(db_session, stranger)
    report = await _linked_report(db_session, system)
    await db_session.commit()

    res = await client.get(
        f"/api/v1/me/team/systems/findings/{report.code}",
        headers=employee_headers,
    )
    assert res.status_code == 404


async def test_one_line_per_thing_that_happened(
    client, db_session, employee_user, employee_headers,
):
    """The FY27 backfill wrote its own ``created`` event on top of the one
    ``create_report`` emits, so backfilled findings rendered "Report
    created" twice in a row — which reads as a broken page.

    The pair is no longer adjacent: auto-routing (2026-08-26) writes an
    ``assigned`` event between them, and an adjacency-based collapse
    silently stopped working the day that shipped. Same-type-same-minute
    events collapse wherever they sit; genuinely distinct ones do not."""
    lead, _token = employee_user
    direct = await _person(db_session, manager=lead)
    system = await _system(db_session, direct)
    report = await _linked_report(db_session, system)
    # Exactly what the backfill did.
    await bounty_svc.add_event(
        db_session, report=report, event_type="created", actor=None,
        actor_kind="system", body="从 security@pingcap.com 邮件补录(FY27)。",
        visibility="internal",
    )
    await db_session.commit()

    data = (
        await client.get(
            f"/api/v1/me/team/systems/findings/{report.code}",
            headers=employee_headers,
        )
    ).json()["data"]
    created = [e for e in data["timeline"] if e["event_type"] == "created"]
    assert len(created) == 1, data["timeline"]

    # A different type in the same minute is a different fact — kept.
    await bounty_svc.add_event(
        db_session, report=report, event_type="status_changed", actor=None,
        actor_kind="system", visibility="internal",
    )
    await db_session.commit()
    data = (
        await client.get(
            f"/api/v1/me/team/systems/findings/{report.code}",
            headers=employee_headers,
        )
    ).json()["data"]
    # ``assigned`` is here because linking the report to an owned system
    # routes it (2026-08-26) — a real thing that happened, and one the lead
    # should see. The subject of this test is that ``created`` collapsed to
    # one line, not the exact cast list.
    assert {e["event_type"] for e in data["timeline"]} == {
        "created", "assigned", "status_changed",
    }


# ── Data gaps ───────────────────────────────────────────────────────────
# Every mechanism above reads a column that is empty on most production
# rows. 97 reports sat unroutable because two systems had no owner and
# nothing in the console said so. These cover the read that names the gap.

async def test_data_gaps_names_the_systems_that_block_routing(
    client, db_session, admin_headers,
):
    from app.domains.bounty import analytics

    nobody = await _system(db_session, None)          # never had an owner
    leaver = await _person(db_session)
    owned_by_leaver = await _system(db_session, leaver)
    active = await _person(db_session)
    fine = await _system(db_session, active)

    # Two open reports on the ownerless system, one on the healthy one.
    await _linked_report(db_session, nobody)
    await _linked_report(db_session, nobody, status="in_discussion")
    await _linked_report(db_session, fine)
    # ``leaver_pending``, not ``left``: the LeaverBlocked listener refuses
    # a direct flip to left, and the predicate under test is
    # "employment_status != active" either way.
    leaver.employment_status = "leaver_pending"
    await db_session.flush()
    await db_session.commit()

    gaps = await analytics.data_gaps(db_session)
    slugs = {s["slug"] for s in gaps["ownerless_systems"]["items"]}
    assert nobody.slug in slugs
    # Assigned-then-left counts too: the metric must not read a non-null
    # owner column as "somebody is answering".
    assert owned_by_leaver.slug in slugs
    assert fine.slug not in slugs

    # The two kinds are distinguished, because the fix differs.
    entry = next(s for s in gaps["ownerless_systems"]["items"]
                 if s["slug"] == nobody.slug)
    departed = next(s for s in gaps["ownerless_systems"]["items"]
                    if s["slug"] == owned_by_leaver.slug)
    assert entry["departed"] is False
    assert departed["departed"] is True
    assert departed["departed_owner_name"]

    # The blocked count is what makes the number worth acting on.
    assert entry["open_reports"] == 2
    assert gaps["ownerless_systems"]["open_reports_blocked"] >= 2
    # Highest-blocking first: the operator fixes the expensive one first.
    counts = [s.get("open_reports", 0)
              for s in gaps["ownerless_systems"]["items"]]
    assert counts == sorted(counts, reverse=True)


async def test_data_gaps_splits_unlinked_by_whether_a_host_is_named(
    client, db_session, admin_headers,
):
    """A report naming a host is one click from being linked; a report
    naming nothing needs a human to read it. One number for both would
    hide which pile you are looking at."""
    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc

    await svc.create_report(
        db_session, title="names a host",
        researcher_email="a@example.com",
        affected_system="tidbcloud.com", actor=None,
    )
    await svc.create_report(
        db_session, title="names nothing",
        researcher_email="b@example.com", actor=None,
    )
    await db_session.commit()

    gaps = await analytics.data_gaps(db_session)
    assert gaps["unlinked"]["with_host"] >= 1
    assert gaps["unlinked"]["without_host"] >= 1
    assert (gaps["unlinked"]["count"]
            == gaps["unlinked"]["with_host"] + gaps["unlinked"]["without_host"])
    # 「未定级」 now means severity IS NULL, and nothing arrives that way
    # since 2026-08-26: every report is born at the lowest tier so the
    # console stops showing a wall nobody has to act on. The gap therefore
    # counts only a level somebody actively removed — which is why this
    # clears one rather than assuming a fresh report is ungraded.
    assert gaps["ungraded"]["count"] == 0, (
        "a report that arrived with the default level is not outstanding work"
    )
    cleared = await bounty_svc.create_report(
        db_session, title="level explicitly cleared",
        researcher_email=f"u{uuid.uuid4().hex[:6]}@example.com", actor=None,
    )
    cleared.severity = None
    await db_session.commit()
    gaps = await analytics.data_gaps(db_session)
    assert gaps["ungraded"]["count"] == 1


async def test_data_gaps_endpoint_is_gated_and_shaped(client, admin_headers):
    res = await client.get(
        "/api/v1/bounty/analytics/data-gaps", headers=admin_headers,
    )
    assert res.status_code == 200
    data = res.json()["data"]
    assert set(data) == {
        "ownerless_systems", "ungraded", "unlinked", "never_replied",
        # Reports somebody examined and ruled out. Reported separately so
        # the operator can see the work already done rather than wondering
        # where the rows went.
        "not_applicable",
    }
    assert "oldest_days" in data["never_replied"]


async def test_data_gaps_is_refused_without_the_security_gate(
    client, employee_headers,
):
    res = await client.get(
        "/api/v1/bounty/analytics/data-gaps", headers=employee_headers,
    )
    assert res.status_code == 403


async def test_ruling_out_a_report_takes_it_out_of_the_unlinked_gap(
    client, db_session, admin_headers,
):
    """44 production reports have no host and no body — they are findings in
    the open-source products. Counting them as "needs linking" forever is
    how a gap number stops being believed, so there is a way to say
    "somebody looked, nothing applies"."""
    from app.domains.bounty import analytics
    from app.domains.bounty import service as svc

    report = await svc.create_report(
        db_session, title="TiDB status port exposes an unauthenticated endpoint",
        researcher_email="oss@example.com", actor=None,
    )
    await db_session.commit()

    before = (await analytics.data_gaps(db_session))["unlinked"]["count"]
    assert before >= 1

    res = await client.post(
        "/api/v1/bounty/backfill/no-system",
        json={"report_ids": [str(report.id)],
              "note": "开源 TiDB 仓库,不属于任何自建系统"},
        headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    assert res.json()["data"]["applied"] == [report.code]

    gaps = await analytics.data_gaps(db_session)
    assert gaps["unlinked"]["count"] == before - 1
    assert gaps["not_applicable"]["count"] >= 1

    # The decision is on the report's own timeline — "who decided this was
    # an open-source finding" is the question asked six months later, and a
    # boolean cannot answer it.
    events = (await client.get(
        f"/api/v1/bounty/reports/{report.id}/events", headers=admin_headers,
    )).json()["data"]["items"]
    marked = [
        e for e in events
        if (e.get("meta") or {}).get("field") == "system_not_applicable"
    ]
    assert marked, events
    assert "开源" in (marked[0].get("body") or "")


async def test_bulk_backfill_never_overwrites_a_human_decision(
    client, db_session, admin_headers,
):
    """A re-run, or a suggestion adopted twice, must not re-point what
    somebody already decided."""
    from app.domains.bounty import service as svc

    graded = await svc.create_report(
        db_session, title="already graded", researcher_email="a@example.com",
        actor=None,
    )
    await svc.set_severity(
        db_session, graded, severity="low", note=None,
        apply_default_reward=True, actor=None,
    )
    linked_system = await _system(db_session, None)
    linked = await svc.create_report(
        db_session, title="already linked", researcher_email="b@example.com",
        system_id=linked_system.id, actor=None,
    )
    await db_session.commit()

    res = await client.post(
        "/api/v1/bounty/backfill/severity",
        json={"items": [{"report_id": str(graded.id), "severity": "critical"}]},
        headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    body = res.json()["data"]
    assert body["applied"] == []
    assert body["skipped"][0]["reason"] == "already_graded"

    other = await _system(db_session, None)
    await db_session.commit()
    res = await client.post(
        "/api/v1/bounty/backfill/link-system",
        json={"items": [{"report_id": str(linked.id),
                         "system_id": str(other.id)}]},
        headers=admin_headers,
    )
    assert res.json()["data"]["skipped"][0]["reason"] == "already_linked"

    # And a linked report cannot be marked "no system applies".
    res = await client.post(
        "/api/v1/bounty/backfill/no-system",
        json={"report_ids": [str(linked.id)]}, headers=admin_headers,
    )
    assert res.json()["data"]["skipped"][0]["reason"] == "already_linked"


async def test_backfill_write_endpoints_need_the_manage_gate(
    client, db_session, employee_headers,
):
    from app.domains.bounty import service as svc

    report = await svc.create_report(
        db_session, title="x", researcher_email="c@example.com", actor=None,
    )
    await db_session.commit()
    for path, payload in (
        ("severity", {"items": [{"report_id": str(report.id),
                                 "severity": "low"}]}),
        ("no-system", {"report_ids": [str(report.id)]}),
    ):
        res = await client.post(
            f"/api/v1/bounty/backfill/{path}", json=payload,
            headers=employee_headers,
        )
        assert res.status_code == 403, (path, res.text)


async def test_backfill_candidates_lists_oldest_first_without_ai(
    client, db_session, admin_headers,
):
    """The AI is optional: with no provider configured the queue still
    works and simply carries no suggestion. Degrading to "no suggestion"
    rather than to a guess is the whole design — a human adopts these."""
    from app.domains.bounty import service as svc

    for title in ("older ungraded", "newer ungraded"):
        await svc.create_report(
            db_session, title=title,
            researcher_email=f"{title.split()[0]}@example.com", actor=None,
        )
    await db_session.commit()

    res = await client.get(
        "/api/v1/bounty/backfill/candidates?kind=severity&suggest=false",
        headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    data = res.json()["data"]
    assert data["kind"] == "severity"
    assert all(item["severity"] is None for item in data["items"])
    assert all(item["suggestion"]["severity"] is None
               for item in data["items"])
    dates = [item["reported_at"] for item in data["items"]]
    assert dates == sorted(dates), "oldest silence must come first"


async def test_one_bad_row_does_not_kill_the_rest_of_a_bulk_write(db_session):
    """A failed flush deactivates the WHOLE transaction, so a bulk writer
    that catches per row and continues *without* a SAVEPOINT turns one bad
    row into every later row failing — while the caller is told those rows
    were merely "skipped".

    The failure here is induced with a **failing flush** (an FK violation
    on ``system_id``), not a failing SELECT. That distinction is the point:
    a test built on a failing SELECT stays green even with the isolation
    removed, and so proves nothing.
    """
    from app.domains.bounty.models import BountyReport

    owner = await _person(db_session)
    target = await _system(db_session, owner)

    first = await bounty_svc.create_report(
        db_session, title="before the bad one",
        researcher_email=f"a{uuid.uuid4().hex[:6]}@example.com", actor=None,
    )
    doomed = await bounty_svc.create_report(
        db_session, title="the bad one",
        researcher_email=f"b{uuid.uuid4().hex[:6]}@example.com", actor=None,
    )
    last = await bounty_svc.create_report(
        db_session, title="after the bad one",
        researcher_email=f"c{uuid.uuid4().hex[:6]}@example.com", actor=None,
    )
    await db_session.flush()
    # Read every attribute NOW. A savepoint rollback expires the whole
    # instance — the primary key included — so touching ``doomed.id`` after
    # the call is a synchronous lazy-load on an async session, and raises
    # MissingGreenlet rather than failing an assertion. (No caller in the
    # app hits this: the routers return only the plain string lists this
    # function builds, and never re-touch the ORM objects.)
    first_id, doomed_id, last_id = first.id, doomed.id, last.id
    first_code, doomed_code, last_code = first.code, doomed.code, last.code
    target_id = target.id

    # dicts keep insertion order, so the bad row really is in the middle.
    result = await bounty_svc.bulk_link_system(
        db_session,
        system_by_id={
            first_id: target_id,
            doomed_id: uuid.uuid4(),  # no such system → FK 1452 on flush
            last_id: target_id,
        },
        actor=None,
    )

    assert doomed_code in {row["code"] for row in result["skipped"]}
    assert first_code in result["applied"]
    assert last_code in result["applied"], (
        "the row AFTER the failure must still be written — without the "
        "savepoint it dies on PendingRollbackError and is silently reported "
        "as skipped"
    )

    # And the session is still usable: this is the statement that raises
    # PendingRollbackError when the failure was swallowed unisolated.
    for report_id in (first_id, last_id):
        fresh = await db_session.get(
            BountyReport, report_id, populate_existing=True,
        )
        assert fresh.system_id == target_id
    doomed_fresh = await db_session.get(
        BountyReport, doomed_id, populate_existing=True,
    )
    assert doomed_fresh.system_id is None


async def test_a_report_that_names_a_registered_host_needs_no_ai(
    client, db_session, admin_headers,
):
    """84 of the unlinked reports state a hostname. For those the match
    against the register is exact and needs no model at all.

    It used to ask the AI for a host and match *that*, so the suggestion
    depended on the LLM echoing back a hostname the report already stated —
    and the column was empty on every row whenever the AI was degraded or
    unconfigured, which is also the state a fresh environment is in. The
    deterministic signal now wins, and the response says which signal
    found it so an exact match is not dressed up as a guess.
    """
    owner = await _person(db_session)
    system = await svc.create_system(
        db_session, name="Clinic", slug=f"clinic-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id, hostnames=["clinic.example.com"],
    )
    stated = await bounty_svc.create_report(
        db_session, title="a finding that names its host",
        researcher_email=f"h{uuid.uuid4().hex[:6]}@example.com",
        affected_system="clinic.example.com", actor=None,
    )
    silent = await bounty_svc.create_report(
        db_session, title="a finding that names nothing",
        researcher_email=f"h{uuid.uuid4().hex[:6]}@example.com",
        actor=None,
    )
    await db_session.commit()

    # suggest=true, but no LLM is configured under test — which is exactly
    # the condition that used to empty this column.
    resp = await client.get(
        "/api/v1/bounty/backfill/candidates?kind=system&suggest=true",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    by_code = {i["code"]: i["suggestion"] for i in resp.json()["data"]["items"]}

    matched = by_code[stated.code]
    assert matched["system"] is not None, (
        "the report states a hostname that is in the register; the match is "
        "exact and must not depend on the AI"
    )
    assert matched["system"]["id"] == str(system.id)
    assert matched["match_source"] == "report"
    assert matched["host"] == "clinic.example.com"

    assert by_code[silent.code]["system"] is None
    assert by_code[silent.code]["match_source"] is None


async def test_a_padded_hostname_still_matches_the_register(
    client, db_session, admin_headers,
):
    """`affected_system` was unnormalised until 2026-08-26 and arrives by
    paste out of a researcher's mail, so legacy rows carry padding. Written
    directly, because the service now strips on the way in and could not
    reproduce the state the database is actually in."""
    from app.domains.bounty.models import BountyReport

    owner = await _person(db_session)
    system = await svc.create_system(
        db_session, name="Padded", slug=f"padded-{uuid.uuid4().hex[:6]}",
        owner_person_id=owner.id, hostnames=["padded.example.com"],
    )
    report = await bounty_svc.create_report(
        db_session, title="legacy padded host",
        researcher_email=f"p{uuid.uuid4().hex[:6]}@example.com",
        affected_system="padded.example.com", actor=None,
    )
    await db_session.execute(
        BountyReport.__table__.update()
        .where(BountyReport.id == report.id)
        .values(affected_system="  Padded.example.com  "),
    )
    await db_session.commit()

    resp = await client.get(
        "/api/v1/bounty/backfill/candidates?kind=system&suggest=true",
        headers=admin_headers,
    )
    suggestion = {
        i["code"]: i["suggestion"] for i in resp.json()["data"]["items"]
    }[report.code]
    assert suggestion["system"] is not None, (
        "padding and case must not stop an otherwise exact match"
    )
    assert suggestion["system"]["id"] == str(system.id)
    assert suggestion["match_source"] == "report"


async def test_the_level_read_from_the_write_up_beats_one_guessed_from_a_title(
    client, db_session, admin_headers,
):
    """131 reports now carry `suggested_severity` from the full-text
    assessment. The backfill drawer was still calling the title-only
    suggester and using its answer — a strictly worse question with a
    better stored answer sitting right there.
    """
    report = await bounty_svc.create_report(
        db_session, title="opaque title",
        researcher_email=f"w{uuid.uuid4().hex[:6]}@example.com", actor=None,
    )
    report.severity = None
    report.severity_is_default = False
    report.suggested_severity = "critical"   # read from the whole write-up
    await db_session.commit()

    resp = await client.get(
        "/api/v1/bounty/backfill/candidates?kind=severity&suggest=true",
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    row = {
        item["code"]: item["suggestion"]
        for item in resp.json()["data"]["items"]
    }[report.code]
    assert row["severity"] == "critical"
    assert row["severity_source"] == "write_up", (
        "and the operator has to be able to see WHICH question produced it"
    )