"""HTTP surface: envelope shape, RBAC, filters, config CRUD, and the
reply preview/log round trip."""
from __future__ import annotations

from decimal import Decimal

import pytest

BASE = "/api/v1/bounty"


async def _create(client, headers, **overrides):
    payload = {
        "title": "反射型 XSS on docs site",
        "researcher_email": "whitehat@example.com",
        "researcher_name": "Alex Chen",
        "affected_system": "docs.pingcap.com",
        "raw_report": "GET /search?q=<script>alert(1)</script>",
        "vuln_type": "xss",
    }
    payload.update(overrides)
    resp = await client.post(f"{BASE}/reports", json=payload, headers=headers)
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


@pytest.mark.asyncio
async def test_meta_returns_vocabularies_and_live_ladder(client, admin_headers):
    resp = await client.get(f"{BASE}/meta", headers=admin_headers)
    assert resp.status_code == 200
    body = resp.json()
    assert body["success"] is True
    data = body["data"]
    assert "new" in data["statuses"]
    assert {t["key"] for t in data["severity_tiers"]} >= {
        "critical", "high", "medium", "low", "informational",
    }
    assert {t["key"] for t in data["templates"]} >= {
        "thanks_generic", "under_evaluation", "out_of_scope_general",
        "out_of_scope_tls", "duplicate_known",
        # One reward letter per tier (2026-08-25): the letter states the
        # amount, and the tier is what decides it.
        "reward_critical", "reward_high", "reward_medium", "reward_low",
        "reward_informational", "payment_sent",
    }
    assert "reward_confirmed" not in {t["key"] for t in data["templates"]}
    # Template bodies are NOT in the meta payload — the picker only needs
    # names, and the bodies are several KB each.
    assert all("body" not in t for t in data["templates"])
    assert data["can_manage"] is True


@pytest.mark.asyncio
async def test_employee_cannot_read_or_write_bounty_data(
    client, employee_headers,
):
    assert (await client.get(f"{BASE}/meta", headers=employee_headers)
            ).status_code == 403
    assert (await client.get(f"{BASE}/reports", headers=employee_headers)
            ).status_code == 403
    assert (
        await client.post(
            f"{BASE}/reports",
            json={"title": "x", "researcher_email": "a@b.com"},
            headers=employee_headers,
        )
    ).status_code == 403


@pytest.mark.asyncio
async def test_auditor_can_manage(client, auditor_headers):
    report = await _create(client, auditor_headers)
    assert report["code"].startswith("BB-")


@pytest.mark.asyncio
async def test_management_reads_but_cannot_mutate(client, db_session):
    """Leadership sees bounty spend (that is the point of the budget
    report) but changes nothing."""
    import uuid

    from sqlalchemy import select

    from app.auth.models import AuthRole, AuthUserRole
    from app.core.security import create_access_token
    from app.identity.person_models import IdentityPerson

    person_id = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=person_id, work_email="mgmt-bounty@test.com", display_name="Mgmt",
        employment_status="active", source_system="successfactors",
        hris_person_id="sf-mgmt-bounty",
    ))
    await db_session.flush()
    role_id = (
        await db_session.execute(
            select(AuthRole.id).where(AuthRole.name == "management"),
        )
    ).scalar_one()
    db_session.add(AuthUserRole(user_id=person_id, role_id=role_id))
    await db_session.commit()
    headers = {
        "Authorization": f"Bearer {create_access_token(str(person_id), ['management'])}",
    }

    assert (await client.get(f"{BASE}/overview", headers=headers)).status_code == 200
    resp = await client.post(
        f"{BASE}/reports",
        json={"title": "x", "researcher_email": "a@b.com"},
        headers=headers,
    )
    assert resp.status_code == 403


@pytest.mark.asyncio
async def test_report_detail_carries_the_event_timeline(client, admin_headers):
    created = await _create(client, admin_headers)
    resp = await client.get(f"{BASE}/reports/{created['id']}", headers=admin_headers)
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["researcher"]["email"] == "whitehat@example.com"
    assert [e["event_type"] for e in data["events"]] == ["created"]


@pytest.mark.asyncio
async def test_list_filters_by_status_and_search(client, admin_headers):
    await _create(client, admin_headers, title="第一个")
    await _create(client, admin_headers, title="第二个",
                  affected_system="tidbcloud.com")

    resp = await client.get(
        f"{BASE}/reports?status=new&search=tidbcloud", headers=admin_headers,
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["meta"]["total"] == 1
    assert body["data"][0]["affected_system"] == "tidbcloud.com"

    # A status nobody is in returns an empty page, not everything.
    resp = await client.get(f"{BASE}/reports?status=closed", headers=admin_headers)
    assert resp.json()["meta"]["total"] == 0


@pytest.mark.asyncio
async def test_search_finds_reports_by_the_researcher(client, admin_headers):
    """「这个白帽子给我们报过什么」 used to need the researcher tab and a
    copied UUID. The address is already on every serialised row, so
    searching it exposes nothing new."""
    await _create(client, admin_headers, title="别人报的",
                  researcher_email="someone.else@example.com")
    await _create(client, admin_headers, title="他报的第一条",
                  researcher_email="savitar0x01@example.com",
                  researcher_name="Savitar")
    await _create(client, admin_headers, title="他报的第二条",
                  researcher_email="savitar0x01@example.com")

    # By full address.
    body = (await client.get(
        f"{BASE}/reports?search=savitar0x01@example.com", headers=admin_headers,
    )).json()
    assert body["meta"]["total"] == 2
    assert {r["title"] for r in body["data"]} == {"他报的第一条", "他报的第二条"}

    # By fragment, which is how anyone actually types it.
    body = (await client.get(
        f"{BASE}/reports?search=savitar", headers=admin_headers,
    )).json()
    assert body["meta"]["total"] == 2

    # And by display name, since the same box already searches names.
    body = (await client.get(
        f"{BASE}/reports?search=Savitar", headers=admin_headers,
    )).json()
    assert body["meta"]["total"] >= 1

    # Page total agrees with the page: cheap guard against a filter that
    # multiplies rows. (It would not catch a plain join here —
    # researcher_id is many-to-one — so it is a sanity check, not a proof.)
    assert body["meta"]["total"] == len(body["data"])

    # Someone else's report is not swept in.
    body = (await client.get(
        f"{BASE}/reports?search=someone.else", headers=admin_headers,
    )).json()
    assert body["meta"]["total"] == 1
    assert body["data"][0]["title"] == "别人报的"


@pytest.mark.asyncio
async def test_reports_carry_the_fy_label_not_just_the_start_year(
    client, admin_headers,
):
    """The stored integer is the START year (2026 = 2026-04 ~ 2027-03) and
    the company says "FY27". The platform centralised that formatter on
    2026-08-24 after a budget page showed "FY2026" next to a figure
    everyone called FY27 — the report payload was still handing the raw
    integer to the drawer, beside a budget bar labelled FY27."""
    created = await _create(client, admin_headers)

    # Grade + reward so the report gets a fiscal year at all.
    await client.post(
        f"{BASE}/reports/{created['id']}/severity",
        json={"severity": "low"}, headers=admin_headers,
    )
    await client.post(
        f"{BASE}/reports/{created['id']}/reward",
        json={"amount_usd": 100}, headers=admin_headers,
    )

    detail = (await client.get(
        f"{BASE}/reports/{created['id']}", headers=admin_headers,
    )).json()["data"]
    assert detail["fiscal_year"] is not None
    label = detail["fiscal_year_label"]
    assert label, "detail payload carries no FY label"
    # Named after the year it ENDS in, two digits.
    assert label.startswith("FY")
    assert str(detail["fiscal_year"] + 1)[-2:] in label
    assert str(detail["fiscal_year"]) not in label.split("(")[0]

    # And on the list, which is what the console renders in bulk.
    row = next(
        r for r in (await client.get(f"{BASE}/reports", headers=admin_headers))
        .json()["data"] if r["id"] == created["id"]
    )
    assert row["fiscal_year_label"] == label


@pytest.mark.asyncio
async def test_list_omits_heavy_content_fields(client, admin_headers):
    await _create(client, admin_headers)
    resp = await client.get(f"{BASE}/reports", headers=admin_headers)
    row = resp.json()["data"][0]
    # The list must not ship every pasted mail body — that is what the
    # detail endpoint is for.
    assert row.get("raw_report") is None
    assert "summary" not in row


@pytest.mark.asyncio
async def test_assign_then_owner_response_flow(client, admin_headers, db_session):
    import uuid

    from app.identity.person_models import IdentityPerson

    owner_id = uuid.uuid4()
    db_session.add(IdentityPerson(
        id=owner_id, work_email="bizowner@test.com", display_name="Biz Owner",
        employment_status="active", source_system="successfactors",
        hris_person_id="sf-bizowner",
    ))
    await db_session.commit()

    report = await _create(client, admin_headers)
    resp = await client.post(
        f"{BASE}/reports/{report['id']}/assign",
        json={"person_id": str(owner_id), "note": "你的服务"},
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["status"] == "awaiting_owner"
    assert data["assignee_name"] == "Biz Owner"
    # The assignment pushes the owner-confirmation card, and the response
    # reports whether a channel actually DELIVERED — not merely that an
    # outbox row was written. No integration is configured in tests, so
    # this is False here; the dedicated test in
    # ``test_owner_self_service.py`` pins that semantics.
    assert data["notified"] is False

    resp = await client.post(
        f"{BASE}/reports/{report['id']}/owner-response",
        json={"decision": "valid", "note": "确实存在", "suggested_severity": "low"},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["status"] == "in_discussion"
    assert data["owner_decision"] == "valid"
    assert data["remediation_status"] == "pending"


@pytest.mark.asyncio
async def test_full_money_path_updates_the_budget_widget(client, admin_headers):
    report = await _create(client, admin_headers)
    rid = report["id"]

    resp = await client.post(
        f"{BASE}/reports/{rid}/severity",
        json={"severity": "low", "apply_default_reward": True},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["reward_amount_usd"] == 100.0

    resp = await client.post(
        f"{BASE}/reports/{rid}/status", json={"status": "confirmed"},
        headers=admin_headers,
    )
    assert resp.status_code == 200

    resp = await client.post(
        f"{BASE}/reports/{rid}/payment",
        json={"payment_status": "paid", "reference": "PO-42"},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["payment_status"] == "paid"
    assert data["payment_reference"] == "PO-42"
    assert data["budget"]["paid_usd"] == 100.0
    # Two funding pots of 5000 each — the console strip shows the
    # programme total; the reports tab breaks it down per track.
    assert data["budget"]["budget_usd"] == 10000.0
    assert data["budget"]["level"] == "ok"


@pytest.mark.asyncio
async def test_overview_counts_and_budget(client, admin_headers):
    await _create(client, admin_headers)
    resp = await client.get(f"{BASE}/overview", headers=admin_headers)
    data = resp.json()["data"]
    assert data["total_reports"] == 1
    assert data["by_status"]["new"] == 1
    assert data["unassigned_open"] == 1
    assert data["budget"]["remaining_usd"] == 10000.0


@pytest.mark.asyncio
async def test_reply_preview_is_side_effect_free_then_log_records_it(
    client, admin_headers,
):
    report = await _create(client, admin_headers)
    rid = report["id"]
    await client.post(
        f"{BASE}/reports/{rid}/severity",
        json={"severity": "low", "apply_default_reward": True},
        headers=admin_headers,
    )

    # The shipped signature has no signer name (we cannot know the
    # operator's), so a reward letter warns until it is configured —
    # that prompt is the intended behaviour, not a bug.
    resp = await client.post(
        f"{BASE}/reports/{rid}/reply-preview",
        json={"template_key": "reward_low"}, headers=admin_headers,
    )
    assert resp.json()["data"]["missing_variables"] == ["signer_name"]

    await client.put(
        f"{BASE}/config/signature",
        json={"signer_name": "Shuai Feng"}, headers=admin_headers,
    )
    resp = await client.post(
        f"{BASE}/reports/{rid}/reply-preview",
        json={"template_key": "reward_low"}, headers=admin_headers,
    )
    assert resp.status_code == 200
    preview = resp.json()["data"]
    assert "USD 100" in preview["body"]
    assert report["code"] in preview["subject"]
    assert "Shuai Feng" in preview["body"]
    assert preview["missing_variables"] == []
    assert preview["template"]["use_count"] == 0

    resp = await client.post(
        f"{BASE}/reports/{rid}/replies",
        json={
            "template_key": "reward_low",
            "subject": preview["subject"],
            "body": preview["body"],
        },
        headers=admin_headers,
    )
    assert resp.status_code == 200
    data = resp.json()["data"]
    assert data["reply_count"] == 1
    assert data["last_reply_template_key"] == "reward_low"
    # The verbatim text we told the researcher is stored, not just a
    # template pointer.
    assert "USD 100" in data["event"]["body"]

    resp = await client.get(f"{BASE}/config/templates", headers=admin_headers)
    tpl = next(
        t for t in resp.json()["data"]["items"] if t["key"] == "reward_low"
    )
    assert tpl["use_count"] == 1


@pytest.mark.asyncio
async def test_reply_preview_refuses_a_reward_letter_with_no_amount(
    client, admin_headers,
):
    """This test used to assert that the missing amount was merely
    *reported* — its own comment said the letter "would promise USD ." and
    accepted that. Reporting it was not enough: the operator copies the
    preview into their mail client, so the advisory became the shipped
    symptom. It is now refused."""
    report = await _create(client, admin_headers)
    resp = await client.post(
        f"{BASE}/reports/{report['id']}/reply-preview",
        json={"template_key": "reward_low"}, headers=admin_headers,
    )
    assert resp.status_code == 422, resp.text
    assert resp.json()["errors"][0]["code"] == "BOUNTY_REPLY_AMOUNT_REQUIRED"


@pytest.mark.asyncio
async def test_inbound_paste_lands_in_the_thread(client, admin_headers):
    report = await _create(client, admin_headers)
    resp = await client.post(
        f"{BASE}/reports/{report['id']}/inbound",
        json={"body": "Here is my bank info ..."}, headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["meta"]["direction"] == "inbound"


@pytest.mark.asyncio
async def test_researcher_profile_aggregates_history(client, admin_headers):
    first = await _create(client, admin_headers, title="一")
    await _create(client, admin_headers, title="二")
    await client.post(
        f"{BASE}/reports/{first['id']}/severity",
        json={"severity": "low", "apply_default_reward": True},
        headers=admin_headers,
    )
    await client.post(
        f"{BASE}/reports/{first['id']}/status", json={"status": "confirmed"},
        headers=admin_headers,
    )
    await client.post(
        f"{BASE}/reports/{first['id']}/payment",
        json={"payment_status": "paid"}, headers=admin_headers,
    )

    resp = await client.get(f"{BASE}/researchers", headers=admin_headers)
    assert resp.status_code == 200
    row = resp.json()["data"][0]
    assert row["email"] == "whitehat@example.com"
    assert row["stats"]["report_count"] == 2
    assert row["stats"]["valid_count"] == 1
    assert row["stats"]["paid_usd"] == 100.0

    resp = await client.get(
        f"{BASE}/researchers/{row['id']}", headers=admin_headers,
    )
    assert len(resp.json()["data"]["reports"]) == 2


@pytest.mark.asyncio
async def test_tier_config_crud_and_guardrails(client, admin_headers):
    # A builtin tier can be re-priced.
    resp = await client.patch(
        f"{BASE}/config/tiers/low", json={"default_reward_usd": 150},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["default_reward_usd"] == 150.0

    # …but not deleted, because historical reports reference its key.
    resp = await client.delete(f"{BASE}/config/tiers/low", headers=admin_headers)
    assert resp.status_code == 409
    assert resp.json()["errors"][0]["code"] == "BOUNTY_TIER_BUILTIN"

    # max < default is rejected — it would make every override illegal.
    resp = await client.patch(
        f"{BASE}/config/tiers/low",
        json={"default_reward_usd": 500, "max_reward_usd": 100},
        headers=admin_headers,
    )
    assert resp.status_code == 422

    # Restore puts the shipped ladder back.
    resp = await client.post(
        f"{BASE}/config/tiers/restore-defaults", headers=admin_headers,
    )
    assert resp.status_code == 200
    low = next(t for t in resp.json()["data"]["items"] if t["key"] == "low")
    assert low["default_reward_usd"] == 100.0


@pytest.mark.asyncio
async def test_template_crud_keeps_variables_hint_honest(client, admin_headers):
    resp = await client.post(
        f"{BASE}/config/templates",
        json={
            "key": "custom_ping",
            "name_zh": "自定义催办", "name_en": "Custom", "name_ja": "カスタム",
            "category": "other",
            "subject": "Re: [{report_code}]",
            "body": "Hello, about {affected_system}. — {signer_name}",
        },
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert set(resp.json()["data"]["variables"]) == {
        "report_code", "affected_system", "signer_name",
    }

    # Editing the body re-derives the hint instead of leaving it stale.
    resp = await client.patch(
        f"{BASE}/config/templates/custom_ping",
        json={"body": "Hello — {signer_name}"}, headers=admin_headers,
    )
    assert set(resp.json()["data"]["variables"]) == {
        "report_code", "signer_name",
    }

    assert (
        await client.delete(
            f"{BASE}/config/templates/custom_ping", headers=admin_headers,
        )
    ).status_code == 200
    # Builtins are protected.
    resp = await client.delete(
        f"{BASE}/config/templates/thanks_generic", headers=admin_headers,
    )
    assert resp.status_code == 409


@pytest.mark.asyncio
async def test_builtin_template_restore_discards_edits(client, admin_headers):
    await client.patch(
        f"{BASE}/config/templates/thanks_generic",
        json={"body": "oops I broke it"}, headers=admin_headers,
    )
    resp = await client.post(
        f"{BASE}/config/templates/thanks_generic/restore", headers=admin_headers,
    )
    assert resp.status_code == 200
    assert "sincerest appreciation" in resp.json()["data"]["body"]


@pytest.mark.asyncio
async def test_budget_config_roundtrip_with_fy_override(client, admin_headers):
    resp = await client.get(f"{BASE}/config/budget", headers=admin_headers)
    data = resp.json()["data"]
    tracks = {row["track"]: row for row in data["tracks"]}
    assert set(tracks) == {"product", "it"}
    assert tracks["product"]["default_usd"] == 5000.0
    assert tracks["it"]["default_usd"] == 5000.0

    fy = data["current_fiscal_year"]
    resp = await client.put(
        f"{BASE}/config/budget",
        json={
            "product": {"default_usd": 6000, "by_fy": {str(fy + 1): 8000}},
        },
        headers=admin_headers,
    )
    assert resp.status_code == 200
    tracks = {row["track"]: row for row in resp.json()["data"]["tracks"]}
    assert tracks["product"]["default_usd"] == 6000.0
    assert tracks["product"]["by_fy"][str(fy + 1)] == 8000.0
    # The current FY has no override, so it inherits the default rather
    # than falling to zero.
    assert tracks["product"]["current_budget_usd"] == 6000.0
    # …and the pot we did NOT patch keeps what it had.
    assert tracks["it"]["default_usd"] == 5000.0

    # An unknown pot is refused rather than silently created.
    resp = await client.put(
        f"{BASE}/config/budget",
        json={"marketing": {"default_usd": 1}}, headers=admin_headers,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_signature_config_rejects_unknown_fields(client, admin_headers):
    resp = await client.put(
        f"{BASE}/config/signature",
        json={"signer_name": "Shuai Feng", "signer_title": "Security Team"},
        headers=admin_headers,
    )
    assert resp.status_code == 200
    assert resp.json()["data"]["signer_name"] == "Shuai Feng"

    resp = await client.put(
        f"{BASE}/config/signature", json={"bogus": "x"}, headers=admin_headers,
    )
    assert resp.status_code == 422


@pytest.mark.asyncio
async def test_soft_deleted_report_disappears_from_list_and_detail(
    client, admin_headers,
):
    report = await _create(client, admin_headers)
    assert (
        await client.request(
            "DELETE", f"{BASE}/reports/{report['id']}",
            json={"reason": "录错了"}, headers=admin_headers,
        )
    ).status_code == 200
    assert (
        await client.get(f"{BASE}/reports/{report['id']}", headers=admin_headers)
    ).status_code == 404
    assert (
        await client.get(f"{BASE}/reports", headers=admin_headers)
    ).json()["meta"]["total"] == 0


@pytest.mark.asyncio
async def test_unknown_vuln_type_and_severity_are_rejected(client, admin_headers):
    resp = await client.post(
        f"{BASE}/reports",
        json={
            "title": "x", "researcher_email": "a@b.com", "vuln_type": "telepathy",
        },
        headers=admin_headers,
    )
    assert resp.status_code == 422

    report = await _create(client, admin_headers)
    resp = await client.post(
        f"{BASE}/reports/{report['id']}/severity",
        json={"severity": "apocalyptic"}, headers=admin_headers,
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_a_reward_letter_is_refused_until_the_amount_exists(
    client, admin_headers,
):
    """The operator copies the preview into their mail client, so an empty
    {amount_usd} becomes a letter reading "your reward ... is USD ." —
    the reported symptom. Refuse to render it instead."""
    created = await _create(client, admin_headers)

    # Ungraded: no tier, so no amount. This is the state 142 production
    # reports are in.
    res = await client.post(
        f"{BASE}/reports/{created['id']}/reply-preview",
        json={"template_key": "reward_high"}, headers=admin_headers,
    )
    assert res.status_code == 422, res.text
    assert res.json()["errors"][0]["code"] == "BOUNTY_REPLY_AMOUNT_REQUIRED"

    # An explicit override is enough — the operator may quote a figure
    # that is not (yet) stored on the report.
    res = await client.post(
        f"{BASE}/reports/{created['id']}/reply-preview",
        json={"template_key": "reward_high", "amount_usd": 2000},
        headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    assert "USD 2,000" in res.json()["data"]["body"]

    # And grading is enough on its own: it pulls the tier's default
    # reward onto the report, so the letter has a figure without anyone
    # typing one.
    await client.post(
        f"{BASE}/reports/{created['id']}/severity",
        json={"severity": "high"}, headers=admin_headers,
    )
    res = await client.post(
        f"{BASE}/reports/{created['id']}/reply-preview",
        json={"template_key": "reward_high"}, headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    assert "USD 800" in res.json()["data"]["body"]


@pytest.mark.asyncio
async def test_letters_without_money_still_preview_with_no_amount(
    client, admin_headers,
):
    """The guard is scoped to letters that reference the amount. An
    acknowledgement must not become un-sendable because no reward has been
    decided — that is the normal state of a new report."""
    created = await _create(client, admin_headers)
    res = await client.post(
        f"{BASE}/reports/{created['id']}/reply-preview",
        json={"template_key": "under_evaluation"}, headers=admin_headers,
    )
    assert res.status_code == 200, res.text


@pytest.mark.asyncio
async def test_meta_exposes_the_per_severity_letter_map(client, admin_headers):
    data = (await client.get(f"{BASE}/meta", headers=admin_headers)).json()["data"]
    by_sev = data["suggested_template_by_severity"]
    assert by_sev["critical"] == "reward_critical"
    assert by_sev["low"] == "reward_low"
    # The retired letter must not be suggested by anything any more.
    assert "reward_confirmed" not in by_sev.values()
    assert "reward_confirmed" not in data["suggested_template_by_status"].values()


@pytest.mark.asyncio
async def test_never_replied_filter_drives_the_reply_queue(
    client, admin_headers,
):
    """118 open reports have never had an answer, the oldest for 295 days.
    Bulk replying was ruled out, so the queue that works through them one
    at a time has to be able to ask for exactly this set, oldest first."""
    silent = await _create(client, admin_headers, title="never answered")
    answered = await _create(client, admin_headers, title="answered once")

    await client.post(
        f"{BASE}/reports/{answered['id']}/replies",
        json={
            "template_key": "thanks_generic",
            "subject": "Re: your report",
            "body": "Thank you for the report.",
        },
        headers=admin_headers,
    )

    body = (await client.get(
        f"{BASE}/reports?never_replied=true&sort=reported_at&order=asc",
        headers=admin_headers,
    )).json()
    codes = {row["code"] for row in body["data"]}
    assert silent["code"] in codes
    assert answered["code"] not in codes, (
        "a report that has been answered must leave the queue"
    )

    # The complement is a real filter too, not a client-side inversion.
    body = (await client.get(
        f"{BASE}/reports?never_replied=false", headers=admin_headers,
    )).json()
    codes = {row["code"] for row in body["data"]}
    assert answered["code"] in codes
    assert silent["code"] not in codes

    # Oldest silence first is the queue's whole ordering claim.
    body = (await client.get(
        f"{BASE}/reports?never_replied=true&sort=reported_at&order=asc",
        headers=admin_headers,
    )).json()
    dates = [row["reported_at"] for row in body["data"]]
    assert dates == sorted(dates)


@pytest.mark.asyncio
async def test_two_addresses_can_be_recorded_as_one_researcher(
    client, admin_headers,
):
    """Confirmed in production: suhaspandya55@ and suhaspandya124@ are one
    person and Helix counted them as two, so every per-researcher figure
    was wrong for them."""
    first = await _create(
        client, admin_headers, title="from the first address",
        researcher_email="dual55@example.com",
    )
    second = await _create(
        client, admin_headers, title="from the second address",
        researcher_email="dual124@example.com",
    )
    alias_id = second["researcher"]["id"]
    canonical_id = first["researcher"]["id"]

    res = await client.post(
        f"{BASE}/researchers/{alias_id}/merge",
        json={"into_id": canonical_id}, headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    assert res.json()["data"]["merged_into_id"] == canonical_id

    # The person's totals are now right from EITHER id — a caller holding
    # the alias must not be told they have one report.
    for rid in (canonical_id, alias_id):
        detail = (await client.get(
            f"{BASE}/researchers/{rid}", headers=admin_headers,
        )).json()["data"]
        assert detail["stats"]["report_count"] == 2, rid

    # Reports are NOT re-pointed: a reply has to go back to the address the
    # mail arrived on, and "which address did this come in on" is a real
    # question in a dispute.
    detail = (await client.get(
        f"{BASE}/reports/{second['id']}", headers=admin_headers,
    )).json()["data"]
    assert detail["researcher"]["email"] == "dual124@example.com"

    # Searching EITHER address finds everything the person sent.
    for needle in ("dual55", "dual124"):
        body = (await client.get(
            f"{BASE}/reports?search={needle}", headers=admin_headers,
        )).json()
        assert body["meta"]["total"] == 2, needle


@pytest.mark.asyncio
async def test_merges_may_not_form_chains_or_cycles(client, admin_headers):
    """Chains would force alias resolution to recurse, and a recursive
    resolve over operator-entered data is how a page ends up hanging."""
    a = (await _create(client, admin_headers, title="a",
                       researcher_email="chain-a@example.com"))["researcher"]["id"]
    b = (await _create(client, admin_headers, title="b",
                       researcher_email="chain-b@example.com"))["researcher"]["id"]
    c = (await _create(client, admin_headers, title="c",
                       researcher_email="chain-c@example.com"))["researcher"]["id"]

    assert (await client.post(
        f"{BASE}/researchers/{a}/merge", json={"into_id": b},
        headers=admin_headers,
    )).status_code == 200

    # b now has an alias, so b may not itself become an alias.
    res = await client.post(
        f"{BASE}/researchers/{b}/merge", json={"into_id": c},
        headers=admin_headers,
    )
    assert res.status_code == 409
    assert res.json()["errors"][0]["code"] == "BOUNTY_MERGE_SOURCE_HAS_ALIASES"

    # And nothing may be merged into an alias.
    res = await client.post(
        f"{BASE}/researchers/{c}/merge", json={"into_id": a},
        headers=admin_headers,
    )
    assert res.status_code == 409
    assert res.json()["errors"][0]["code"] == "BOUNTY_MERGE_TARGET_IS_ALIAS"

    # Self-merge is nonsense.
    res = await client.post(
        f"{BASE}/researchers/{c}/merge", json={"into_id": c},
        headers=admin_headers,
    )
    assert res.status_code == 422


@pytest.mark.asyncio
async def test_a_merge_can_be_undone(client, admin_headers):
    """An identity claim typed by a person is sometimes wrong, and a
    one-way door on a guess is a trap."""
    a = (await _create(client, admin_headers, title="undo a",
                       researcher_email="undo-a@example.com"))["researcher"]["id"]
    b = (await _create(client, admin_headers, title="undo b",
                       researcher_email="undo-b@example.com"))["researcher"]["id"]

    await client.post(f"{BASE}/researchers/{a}/merge", json={"into_id": b},
                      headers=admin_headers)
    res = await client.post(f"{BASE}/researchers/{a}/unmerge",
                            headers=admin_headers)
    assert res.status_code == 200
    assert res.json()["data"]["merged_into_id"] is None

    detail = (await client.get(f"{BASE}/researchers/{a}",
                                headers=admin_headers)).json()["data"]
    assert detail["stats"]["report_count"] == 1


@pytest.mark.asyncio
async def test_merging_two_addresses_makes_the_person_s_totals_right(
    client, admin_headers,
):
    """The point of the merge, and the thing nothing asserted.

    Confirmed in production: `suhaspandya55@` and `suhaspandya124@` are one
    person, counted as two. So "how many has he sent us" and "how much have
    we paid him" — the two figures the reply templates and the repeat-
    submitter signal lean on — were both wrong for anyone who ever used a
    second address.

    Both directions matter. Reports keep pointing at the address that sent
    them (a reply has to go back to the address the mail came from, and
    "which address did this arrive on" is a real question in a dispute), so
    the totals have to resolve through the pointer — and they have to
    resolve for a caller holding EITHER id, or opening the alias's page
    would say the person has one report and no money.
    """
    first = await _create(client, admin_headers, title="finding one",
                          researcher_email="totals-a@example.com")
    second = await _create(client, admin_headers, title="finding two",
                           researcher_email="totals-b@example.com")
    third = await _create(client, admin_headers, title="finding three",
                          researcher_email="totals-b@example.com")
    canonical = first["researcher"]["id"]
    alias = second["researcher"]["id"]
    assert third["researcher"]["id"] == alias

    async def _stats(researcher_id: str) -> dict:
        resp = await client.get(
            f"{BASE}/researchers/{researcher_id}", headers=admin_headers,
        )
        assert resp.status_code == 200, resp.text
        return resp.json()["data"]["stats"]

    # Pay one of the alias's findings, so the money total has something to
    # be wrong about.
    for step, payload in (
        ("severity", {"severity": "medium"}),
        ("status", {"status": "confirmed"}),
        ("reward", {"amount_usd": "500"}),
        ("payment", {"payment_status": "paid"}),
    ):
        resp = await client.post(
            f"{BASE}/reports/{second['id']}/{step}", json=payload,
            headers=admin_headers,
        )
        assert resp.status_code == 200, (step, resp.text)

    before = await _stats(canonical)
    assert before["report_count"] == 1
    assert Decimal(str(before["paid_usd"] or 0)) == Decimal("0")

    resp = await client.post(
        f"{BASE}/researchers/{alias}/merge", json={"into_id": canonical},
        headers=admin_headers,
    )
    assert resp.status_code == 200, resp.text

    after = await _stats(canonical)
    assert after["report_count"] == 3, (
        "one person, three reports across two addresses"
    )
    assert Decimal(str(after["paid_usd"])) == Decimal("500")

    # The alias's own page must show the same person's numbers, not zero.
    via_alias = await _stats(alias)
    assert via_alias["report_count"] == 3
    assert Decimal(str(via_alias["paid_usd"])) == Decimal("500")

    # And the reports still point at the address that actually sent them.
    body = (await client.get(
        f"{BASE}/reports?researcher_id={alias}", headers=admin_headers,
    )).json()
    assert body["meta"]["total"] == 2, (
        "rewriting researcher_id on the reports would have destroyed both "
        "'reply to the address it came from' and 'which address was this'"
    )


@pytest.mark.asyncio
async def test_the_report_points_at_the_mail_it_came_from(
    client, admin_headers,
):
    """For 149 of 149 production reports the technical detail exists ONLY
    in the security@ thread — the FY27 import kept correspondence in the
    mailbox on purpose. A console that cannot point at it is asking the
    operator to go and find the thread by searching their inbox for the
    title, which is what "我看不到任何关于漏洞的信息" actually meant.
    """
    created = await _create(client, admin_headers, title="from a real thread",
                            researcher_email="thread@example.com")

    resp = await client.post(
        f"{BASE}/reports/{created['id']}/comments",
        json={"body": "x", "visibility": "internal"}, headers=admin_headers,
    )
    assert resp.status_code == 200

    # No thread stamped yet — and a dead link is worse than no link.
    detail = (await client.get(
        f"{BASE}/reports/{created['id']}", headers=admin_headers,
    )).json()["data"]
    assert detail["mail_thread_url"] is None

    # Stamp one exactly as the FY27 import and the mail-intake queue do.
    from app.domains.bounty import service as svc
    from app.domains.bounty.serializers import mail_thread_url_from_events

    assert mail_thread_url_from_events([
        type("E", (), {"event_type": "created",
                       "meta": {"backfill": "gmail:1a015915dfb6880f"}})(),
    ]) == "https://mail.google.com/mail/u/0/#all/1a015915dfb6880f"

    # The import's placeholder ids resolve to nothing, deliberately.
    assert mail_thread_url_from_events([
        type("E", (), {"event_type": "created",
                       "meta": {"backfill": "gmail:sav-BB-2026-002"}})(),
    ]) is None
    assert svc is not None