"""security@ ingest: a queue a person confirms, never an auto-created report.

Every report is pasted in by hand today, which is the mechanical reason 118
of 149 never got an answer and the oldest waited 295 days. These cover the
three decisions that keep the automation from making things worse.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest

from app.domains.bounty import mail_intake
from app.domains.bounty import service as svc
from app.domains.bounty.models import BountyMailIntake

pytestmark = pytest.mark.asyncio

BASE = "/api/v1/bounty"


async def _queued(db, **overrides):
    row = BountyMailIntake(
        message_id=overrides.get("message_id", f"<{uuid.uuid4()}@example.com>"),
        thread_id=overrides.get("thread_id"),
        from_email=overrides.get("from_email", "researcher@example.com"),
        from_name=overrides.get("from_name", "A Researcher"),
        subject=overrides.get("subject", "SQL injection on example.com"),
        snippet=overrides.get("snippet", "I found an unauthenticated endpoint"),
        received_at=overrides.get(
            "received_at", datetime.now(UTC) - timedelta(days=3),
        ),
        status=overrides.get("status", "pending"),
        parsed=overrides.get("parsed"),
    )
    db.add(row)
    await db.flush()
    return row


async def _desk_operator(db, email="officer@test.com"):
    """One active 安全员 — otherwise every desk-facing push fans out to
    nobody and a card-content assertion passes on an empty list."""
    from sqlalchemy import select

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

    person = IdentityPerson(
        id=uuid.uuid4(),
        work_email=email,
        display_name="Security Officer",
        employment_status="active",
        source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:8]}",
    )
    db.add(person)
    await db.flush()
    role_id = (
        await db.execute(
            select(AuthRole.id).where(AuthRole.name == "security_officer"),
        )
    ).scalar_one_or_none()
    if role_id is None:
        role_id = uuid.uuid4()
        db.add(AuthRole(id=role_id, name="security_officer", is_system=True))
        await db.flush()
    db.add(AuthUserRole(user_id=person.id, role_id=role_id))
    await db.flush()
    return person


async def test_accepting_a_queued_mail_creates_the_report(
    client, db_session, admin_headers,
):
    row = await _queued(
        db_session, parsed={"severity": "high", "confidence": "medium"},
    )
    await db_session.commit()

    res = await client.post(
        f"{BASE}/mail-intake/{row.id}/accept", json={},
        headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    report = res.json()["data"]
    assert report["title"] == "SQL injection on example.com"
    assert report["researcher"]["email"] == "researcher@example.com"
    # Reported-at is the mail's date, not now: the silence clock has to
    # start when the researcher wrote, or every ingested report looks fresh.
    assert report["days_silent"] >= 3

    # The AI's severity is carried as a SUGGESTION, never applied — grading
    # pulls a reward tier and a fix deadline, a decision with money on it.
    assert report["severity"] == "informational", (
        "reports arrive at the lowest tier (2026-08-26), so nothing is "
        "blocked waiting for a grade"
    )
    assert report["severity_is_default"] is True, (
        "and that arrival level is not a grade — the AI reading it as "
        "High must not be silently applied, and the console must still "
        "count this in 待定级"
    )
    assert report["suggested_severity"] == "high"


async def test_a_decided_mail_cannot_be_decided_twice(
    client, db_session, admin_headers,
):
    row = await _queued(db_session)
    await db_session.commit()

    assert (await client.post(
        f"{BASE}/mail-intake/{row.id}/accept", json={}, headers=admin_headers,
    )).status_code == 200
    again = await client.post(
        f"{BASE}/mail-intake/{row.id}/accept", json={}, headers=admin_headers,
    )
    assert again.status_code == 409
    assert again.json()["errors"][0]["code"] == "BOUNTY_MAIL_ALREADY_DECIDED"

    ignored = await client.post(
        f"{BASE}/mail-intake/{row.id}/ignore", json={}, headers=admin_headers,
    )
    assert ignored.status_code == 409


async def test_ignoring_keeps_the_row_so_it_is_never_re_offered(
    client, db_session, admin_headers,
):
    row = await _queued(db_session, subject="Buy our pentest services")
    await db_session.commit()

    res = await client.post(
        f"{BASE}/mail-intake/{row.id}/ignore",
        json={"note": "厂商推销,不是漏洞报告"}, headers=admin_headers,
    )
    assert res.status_code == 200
    assert res.json()["data"]["status"] == "ignored"

    listed = (await client.get(
        f"{BASE}/mail-intake", headers=admin_headers,
    )).json()["data"]["items"]
    assert str(row.id) not in {item["id"] for item in listed}

    # The row survives: deleting it would let the next poll offer the same
    # vendor pitch again, for ever.
    await db_session.refresh(row)
    assert row.status == "ignored"
    assert row.note


async def test_the_queue_is_oldest_first(client, db_session, admin_headers):
    now = datetime.now(UTC)
    await _queued(db_session, subject="newer", received_at=now - timedelta(days=1))
    await _queued(db_session, subject="older", received_at=now - timedelta(days=90))
    await db_session.commit()

    items = (await client.get(
        f"{BASE}/mail-intake", headers=admin_headers,
    )).json()["data"]["items"]
    subjects = [i["subject"] for i in items]
    assert subjects.index("older") < subjects.index("newer")


async def test_the_queue_needs_the_security_gate(
    client, db_session, employee_headers,
):
    row = await _queued(db_session)
    await db_session.commit()

    assert (await client.get(
        f"{BASE}/mail-intake", headers=employee_headers,
    )).status_code == 403
    assert (await client.post(
        f"{BASE}/mail-intake/{row.id}/accept", json={},
        headers=employee_headers,
    )).status_code == 403


async def test_the_serialised_row_never_carries_the_body(
    client, db_session, admin_headers,
):
    """Researchers send tax forms and bank details unprompted. Only a
    snippet is stored, and the payload must not grow a body field later."""
    await _queued(db_session)
    await db_session.commit()

    item = (await client.get(
        f"{BASE}/mail-intake", headers=admin_headers,
    )).json()["data"]["items"][0]
    assert "snippet" in item
    for forbidden in ("body", "raw", "payload", "attachments"):
        assert forbidden not in item, f"{forbidden} leaked into the queue"


async def test_a_followup_on_a_known_thread_attaches_instead_of_duplicating(
    db_session,
):
    """Researchers chase their own threads, and 118 unanswered reports means
    a lot of chasing. A second mail on a thread that already produced a
    report must land ON that report as inbound correspondence — otherwise
    every chase becomes a duplicate finding and the conversation splits
    across two timelines."""
    report = await svc.create_report(
        db_session, title="original finding",
        researcher_email="chaser@example.com", actor=None,
    )
    await db_session.flush()

    followup = await _queued(
        db_session, thread_id="THREAD-1", from_email="chaser@example.com",
        subject="Re: original finding — any update?",
        snippet="Any update on this? It has been three months.",
    )
    await db_session.commit()

    attached = await mail_intake.attach_followup(
        db_session, followup, report.id,
    )
    assert attached is True
    await db_session.commit()

    assert followup.status == "attached"
    assert followup.report_id == report.id
    assert followup.decided_at is not None

    events = await svc.list_events(db_session, report.id, internal=True)
    inbound = [e for e in events if e.event_type == "inbound_logged"]
    assert len(inbound) == 1, "the chase must land on the existing report"
    assert "three months" in (inbound[0].body or "")
    assert (inbound[0].meta or {}).get("via") == "mail_intake"

    # And it is out of the confirmation queue: nobody should be asked to
    # triage a follow-up as if it were a new finding.
    assert followup.id not in {
        row.id for row in await mail_intake.pending(db_session)
    }


async def test_attaching_to_a_missing_report_is_not_silently_successful(
    db_session,
):
    """A report can be soft-deleted between polls. Returning True there
    would mark the mail handled while nothing recorded it."""
    row = await _queued(db_session, thread_id="THREAD-2")
    await db_session.commit()

    assert await mail_intake.attach_followup(
        db_session, row, uuid.uuid4(),
    ) is False
    assert row.status == "pending"


async def test_two_operators_accepting_the_same_mail_create_one_report(
    db_engine,
):
    """The queue is worked by hand, so the same mail gets double-clicked and
    two people work it at once. Without a locking read both callers see
    ``pending`` and both create a report.

    A duplicate here is not cosmetic: every count, the budget attribution
    and the reward owed are computed off these rows, so a duplicated
    finding can be graded twice and paid twice.

    The overlap is forced, not hoped for. Two sessions run concurrently
    under ``asyncio.gather`` in scheduling order and finish before each
    other's window opens — that version of this test passed with the lock
    removed. So A claims the row and *holds* its transaction open while B
    tries. With ``FOR UPDATE`` B blocks, and by the time it reads, the row
    says ``accepted``; without it B reads its own snapshot's ``pending``
    and inserts a second report.
    """
    import asyncio

    from sqlalchemy import func, select, text
    from sqlalchemy.ext.asyncio import async_sessionmaker

    from app.domains.bounty.models import BountyReport, BountyResearcher

    session_factory = async_sessionmaker(db_engine, expire_on_commit=False)
    email = f"racer{uuid.uuid4().hex[:8]}@example.com"

    async with session_factory() as seed:
        row = await _queued(seed, from_email=email)
        await seed.commit()
        intake_id = row.id

    started = asyncio.Event()

    async def _second_operator():
        async with session_factory() as db:
            # Bound the wait so a genuinely-blocking lock fails the test
            # loudly instead of hanging the suite.
            await db.execute(text("SET innodb_lock_wait_timeout = 5"))
            started.set()
            try:
                report = await mail_intake.accept(db, intake_id, actor=None)
                await db.commit()
                return ("ok", report.code)
            except Exception as exc:  # noqa: BLE001 — the losing racer
                await db.rollback()
                return ("err", getattr(exc, "code", type(exc).__name__))

    async with session_factory() as first:
        winner = await mail_intake.accept(first, intake_id, actor=None)
        # A holds its transaction open here: the row is claimed but not
        # committed. This is the window the lock has to cover.
        second = asyncio.create_task(_second_operator())
        await started.wait()
        await asyncio.sleep(0.4)  # let B reach (and block on) the row
        assert not second.done(), (
            "the second operator read the row without waiting for the "
            "first — the claim is not a locking read"
        )
        await first.commit()

    kind, detail = await second
    assert kind == "err", f"the second accept must be refused, got {detail}"
    assert detail == "BOUNTY_MAIL_ALREADY_DECIDED", detail

    async with session_factory() as db:
        made = int(await db.scalar(
            select(func.count()).select_from(BountyReport)
            .join(BountyResearcher,
                  BountyResearcher.id == BountyReport.researcher_id)
            .where(BountyResearcher.email == email,
                   BountyReport.deleted_at.is_(None)),
        ) or 0)
    assert made == 1, f"one mail must yield one report, found {made}"
    assert winner.code


async def test_the_poller_refuses_a_mailbox_outside_the_allowlist(db_session):
    """`mailbox` arrives from the scheduled job's config, which is editable
    in the console, and it goes straight into `with_subject()` — domain-wide
    delegation impersonates whatever Workspace user it is handed.

    Unconstrained that turns "can edit a cron config" into "can read any
    employee's mail metadata into a table the security roles read", which
    is an escalation across a privacy boundary, not an operational setting.
    The refusal has to happen BEFORE any credential is built, or a
    misconfigured job reaches Google with someone else's address.
    """
    from app.core.exceptions import AppException

    with pytest.raises(AppException) as caught:
        await mail_intake.poll(
            db_session, config={"mailbox": "ceo@pingcap.com"},
        )
    assert caught.value.code == "BOUNTY_MAIL_MAILBOX_NOT_ALLOWED"
    assert "ceo@pingcap.com" in caught.value.message

    # Case and whitespace must not slip past the check.
    with pytest.raises(AppException) as caught:
        await mail_intake.poll(
            db_session, config={"mailbox": "  CEO@PingCAP.com "},
        )
    assert caught.value.code == "BOUNTY_MAIL_MAILBOX_NOT_ALLOWED"

    # And the allowlisted ones get past the check — they fail later, on the
    # missing credential, which is a different error entirely.
    for allowed in sorted(mail_intake.ALLOWED_POLL_MAILBOXES):
        with pytest.raises(AppException) as caught:
            await mail_intake.poll(db_session, config={"mailbox": allowed})
        assert caught.value.code != "BOUNTY_MAIL_MAILBOX_NOT_ALLOWED", allowed


def test_only_the_researchers_first_message_is_imported():
    """The write-up is the first message. Everything after it is the
    conversation — chasing, negotiation, and the bank details that follow a
    reward being agreed. Taking only the first is what keeps this import to
    technical content BY CONSTRUCTION, before redaction has to catch
    anything."""
    messages = [
        {"sender": "researcher@example.com", "date": "2026-07-01",
         "plaintextBody": "Title: IDOR on /api/x\nSteps: call it as tenant B"},
        {"sender": "security@pingcap.com", "date": "2026-07-02",
         "plaintextBody": "Thanks, we are looking at it."},
        {"sender": "researcher@example.com", "date": "2026-07-20",
         "plaintextBody": "My IBAN is GB29NWBK60161331926819, please pay."},
    ]
    body = mail_intake.researcher_first_message(
        messages, "researcher@example.com",
    )
    assert "IDOR on /api/x" in body
    assert "IBAN" not in body, (
        "the payment message is a later message and must never be the one "
        "imported"
    )


def test_a_forwarded_thread_falls_back_to_the_first_outside_sender():
    """Some threads arrived by forward, so the address on file is not in
    any header. Falling back to the first non-internal sender beats
    importing our own forwarding note as the vulnerability."""
    messages = [
        {"sender": "shuai.feng@pingcap.com", "date": "2026-07-01",
         "plaintextBody": "FYI, forwarding this one."},
        {"sender": "someone@elsewhere.test", "date": "2026-07-02",
         "plaintextBody": "Title: open redirect on /go"},
    ]
    body = mail_intake.researcher_first_message(messages, "nobody@nowhere")
    assert "open redirect" in body


def test_banking_details_are_stripped_on_the_way_in():
    """The stated reason the FY27 import stored no bodies at all: a ledger
    that never held an IBAN cannot leak one. Importing the write-up is
    worth doing; importing a stranger's account number with it is not.

    Deliberately over-eager — a false positive costs a redacted string in
    a description that is still readable in the original mail, one click
    away on every report. A false negative is permanent.
    """
    raw = (
        "Steps to reproduce: GET /api/v1/users/2 as user 1.\n"
        "For the reward: IBAN GB29NWBK60161331926819\n"
        "SWIFT: DEUTDEFF\n"
        "Account Number: 1234 5678 9012\n"
        "Card 4111 1111 1111 1111\n"
    )
    out = mail_intake.redact_payment_details(raw)
    assert "GET /api/v1/users/2" in out, "the finding itself must survive"
    assert "GB29NWBK60161331926819" not in out
    assert "DEUTDEFF" not in out
    assert "1234 5678 9012" not in out
    assert "4111 1111 1111 1111" not in out


async def test_a_report_that_already_has_a_body_is_never_overwritten(
    db_session,
):
    """Re-running the backfill must be free. Somebody may have pasted the
    write-up in by hand, and an import that clobbers it loses work nobody
    can get back."""
    report = await svc.create_report(
        db_session, title="already has one",
        researcher_email="keep@example.com", actor=None,
    )
    report.raw_report = "pasted in by a person"
    await db_session.flush()

    assert await mail_intake.import_body_for_report(db_session, report) is False
    assert report.raw_report == "pasted in by a person"


async def test_the_group_address_can_never_be_polled(db_session):
    """`security@pingcap.com` is a GROUP. Domain-wide delegation only
    impersonates real USERS, so a poller pointed at it fails with
    `unauthorized_client` for every scope — verified against production
    2026-08-26, including `gmail.send`, which Helix uses successfully all
    day. Leaving it as the default made a job that could never work look
    configured."""
    from app.core.exceptions import AppException

    assert "security@pingcap.com" not in mail_intake.ALLOWED_POLL_MAILBOXES
    assert "security@pingcap.com" not in mail_intake.ALLOWED_FETCH_MAILBOXES

    with pytest.raises(AppException) as caught:
        await mail_intake.poll(
            db_session, config={"mailbox": "security@pingcap.com"},
        )
    assert caught.value.code == "BOUNTY_MAIL_MAILBOX_NOT_ALLOWED"
    assert "邮件组" in caught.value.message, (
        "and it has to say WHY, or the next person re-adds it"
    )


async def test_listing_and_fetching_are_gated_separately(db_session):
    """Listing an inbox walks whatever the query matches; fetching one
    thread by an id Helix already recorded is bounded to correspondence the
    ledger already knows about. The archive mailbox is a person's, so it may
    be fetched from and never listed."""
    assert mail_intake.ARCHIVE_MAILBOX in mail_intake.ALLOWED_FETCH_MAILBOXES
    assert mail_intake.ARCHIVE_MAILBOX not in mail_intake.ALLOWED_POLL_MAILBOXES

    from app.core.exceptions import AppException

    with pytest.raises(AppException) as caught:
        await mail_intake.poll(
            db_session, config={"mailbox": mail_intake.ARCHIVE_MAILBOX},
        )
    assert caught.value.code == "BOUNTY_MAIL_MAILBOX_NOT_ALLOWED"


async def test_fetching_from_an_unlisted_mailbox_is_refused(db_session):
    from app.core.exceptions import AppException

    report = await svc.create_report(
        db_session, title="x", researcher_email="f@example.com", actor=None,
    )
    await db_session.flush()
    with pytest.raises(AppException) as caught:
        await mail_intake.import_body_for_report(
            db_session, report, mailbox="ceo@pingcap.com",
        )
    assert caught.value.code == "BOUNTY_MAIL_MAILBOX_NOT_ALLOWED"


def test_messages_are_ordered_by_real_time_not_by_weekday_name():
    """RFC2822 Date headers sorted as strings order by WEEKDAY —
    Fri < Mon < Sat < Thu < Tue < Wed. So "the researcher's first message"
    came out effectively random, and the first production import batch put
    a follow-up in one report and OUR OWN REPLY in another, both presented
    as the finding.
    """
    messages = [
        # Deliberately chosen so string-sorting picks the wrong one:
        # "Fri" sorts before "Tue", but Tue 28 Jul is genuinely earlier.
        {"sender": "r@example.com", "date": "Fri, 01 Aug 2026 09:00:00 +0000",
         "plaintextBody": "Ping — any update?"},
        {"sender": "r@example.com", "date": "Tue, 28 Jul 2026 01:17:00 +0000",
         "plaintextBody": "Title: the actual vulnerability report"},
        {"sender": "r@example.com", "date": "Mon, 18 Aug 2026 15:50:00 +0000",
         "plaintextBody": "Ping again."},
    ]
    body = mail_intake.researcher_first_message(messages, "r@example.com")
    assert body == "Title: the actual vulnerability report"


def test_an_unparseable_date_sorts_last_rather_than_first():
    """A malformed header must not win the "earliest" slot by accident."""
    messages = [
        {"sender": "r@example.com", "date": "not a date",
         "plaintextBody": "junk"},
        {"sender": "r@example.com", "date": "Tue, 28 Jul 2026 01:17:00 +0000",
         "plaintextBody": "the real report"},
    ]
    assert mail_intake.researcher_first_message(
        messages, "r@example.com",
    ) == "the real report"


def test_our_own_reply_is_never_imported_as_the_finding():
    """The bug this exists to stop, caught in production on a real thread.

    The address on file was unavailable, so the "match the researcher"
    guard short-circuited on an empty value and the loop returned the
    earliest message of ANY sender — which was a reply written by the
    security team. It then rendered on the report page as the researcher's
    vulnerability report. That is worse than the empty page it replaced.

    Our own messages are excluded in BOTH passes, whatever the address on
    file says.
    """
    messages = [
        # Ours, and genuinely earliest in absolute time (+0800 vs +0700).
        {"sender": "dong.zhang@pingcap.com",
         "date": "Tue, 18 Aug 2026 16:36:14 +0800",
         "plaintextBody": "Dear Submitter, Thank you for your report."},
        {"sender": "rhyfk0z@gmail.com",
         "date": "Tue, 18 Aug 2026 16:14:55 +0700",
         "plaintextBody": "Dear Triager, here is the region mismatch detail."},
    ]
    got = mail_intake.researcher_first_message(messages, "rhyfk0z@gmail.com")
    assert got.startswith("Dear Triager")
    # The case that actually failed: no address on file at all.
    got = mail_intake.researcher_first_message(messages, "")
    assert got.startswith("Dear Triager"), (
        "an empty address must not turn the filter off"
    )
    got = mail_intake.researcher_first_message(messages, "someone@else.test")
    assert got.startswith("Dear Triager")


def test_a_thread_with_only_our_messages_imports_nothing():
    """Better no body than our own words presented as the researcher's."""
    messages = [
        {"sender": "security@pingcap.com",
         "date": "Tue, 18 Aug 2026 16:36:14 +0800",
         "plaintextBody": "Thanks, we are looking at it."},
    ]
    assert mail_intake.researcher_first_message(messages, "") is None


async def test_the_fallback_search_is_pinned_to_one_sender(db_session):
    """~half the recorded thread ids point at OUR reply, because Gmail
    threads an outbound message separately from the inbound report in the
    sender's own mailbox. The write-up is still there under another thread,
    so the fallback finds it by sender.

    What it must never become is a subject or keyword search: that would
    walk mail unrelated to any report. The query is one external address the
    ledger already records as a submitter.
    """
    seen: list[str] = []

    class _Svc:
        def users(self):
            return self

        def messages(self):
            return self

        def list(self, **kw):
            seen.append(kw.get("q", ""))
            return self

        def get(self, **kw):
            return self

        def execute(self):
            return {"messages": []}

    out = await mail_intake._search_researcher_message(
        _Svc(), "Researcher@Example.COM",
    )
    assert out is None
    assert seen == ["from:researcher@example.com"], (
        "lower-cased, and scoped to from: — nothing else"
    )


async def test_the_fallback_refuses_a_malformed_address(db_session):
    """A crafted researcher address must not be able to inject extra Gmail
    query operators — the whole point of pinning the query.

    Uses a RECORDER, not an exception, as the tripwire: the function wraps
    the API call in `except Exception`, so a raising stub returns None
    whether the guard fired or not. The first version of this test passed
    with the guard deleted.
    """
    reached: list[str] = []

    class _Svc:
        def users(self):
            reached.append("users")
            return self

        def messages(self):
            return self

        def list(self, **kw):
            reached.append(kw.get("q", ""))
            return self

        def get(self, **kw):
            return self

        def execute(self):
            return {"messages": []}

    for bad in ("", "not-an-email", 'a@b.com" OR subject:salary',
                "a@b.com\nfrom:ceo@pingcap.com", "  "):
        assert await mail_intake._search_researcher_message(_Svc(), bad) is None
        assert reached == [], (
            f"{bad!r} reached the Gmail API — the guard did not fire"
        )


async def test_a_report_with_no_recorded_thread_still_tries_the_sender(
    db_session, monkeypatch,
):
    """The FY27 import left a `sav-` placeholder on some reports instead of
    a real thread id. Returning early on that left reports empty whose
    researcher had five messages sitting in the mailbox — measured in
    production on BB-2026-146.
    """
    searched: list[str] = []

    async def _fake_search(_service, email):
        searched.append(email)
        return "Title: found by sender, not by thread"

    async def _fake_service(_db, _mailbox):
        return object()

    monkeypatch.setattr(mail_intake, "_search_researcher_message", _fake_search)
    monkeypatch.setattr(mail_intake, "_service", _fake_service)

    report = await svc.create_report(
        db_session, title="no thread recorded",
        researcher_email="findme@example.com", actor=None,
    )
    # A placeholder marker, exactly as the import wrote them.
    await svc.add_event(
        db_session, report=report, event_type="created", actor=None,
        actor_kind="system", meta={"backfill": f"gmail:sav-{report.code}"},
        visibility="internal",
    )
    await db_session.flush()

    assert await mail_intake.import_body_for_report(db_session, report) is True
    assert searched == ["findme@example.com"]
    assert "found by sender" in report.raw_report


async def test_a_report_whose_mail_is_absent_is_not_retried_forever(
    db_session, monkeypatch,
):
    """The batch selects newest-first among reports lacking a write-up. Any
    report whose mail simply is not in the archive mailbox therefore sat at
    the head of the queue re-costing two API round trips every run, and
    throughput decayed to zero before reaching importable reports further
    down — measured in production, 4 of 8 and falling.

    So a failed attempt is recorded and excluded next time. The job has to
    converge on its own.
    """
    calls: list[str] = []

    async def _nothing(_service, email):
        calls.append(email)
        return None

    async def _fake_service(_db, _mailbox):
        return object()

    monkeypatch.setattr(mail_intake, "_search_researcher_message", _nothing)
    monkeypatch.setattr(mail_intake, "_service", _fake_service)

    report = await svc.create_report(
        db_session, title="mail is not in this mailbox",
        researcher_email="absent@example.com", actor=None,
    )
    await db_session.flush()

    assert await mail_intake.import_body_for_report(db_session, report) is False
    assert calls == ["absent@example.com"]

    events = await svc.list_events(db_session, report.id, internal=True)
    markers = [
        e for e in events
        if (e.meta or {}).get("import") == mail_intake.IMPORT_NOT_FOUND
    ]
    assert len(markers) == 1, "the attempt has to be recorded"

    result = await mail_intake.backfill_bodies_and_assess(db_session, limit=50)
    assert report.code not in str(result), result
    assert calls == ["absent@example.com"], (
        "it must not be re-attempted while the marker stands"
    )


async def test_the_default_desk_is_the_security_officers_alone(db_session):
    """`security_officer` (安全员) IS the operational desk — the role exists
    to triage inbound vulnerability reports. The resolver originally targeted
    `global_admin` only, so the people doing the work were the only ones not
    told; then it targeted both, which in production meant eight recipients
    for work four people do.

    Now the DEFAULT is the desk alone, and anyone else — including a global
    admin who does want these — is named in 配置. See
    `test_alert_recipients.py` for the configurable side.
    """
    from app.domains.bounty import notify as bnotify

    assert bnotify._DESK_ROLES == ("security_officer",)
    assert "auditor" not in bnotify._DESK_ROLES, (
        "still a read-heavy governance seat, not an operational hand-off"
    )


async def test_a_quiet_poll_notifies_nobody(db_session):
    """Fifteen-minute polling means a push on every run — or one per mail —
    would train the desk to mute the channel. One push per poll, and only
    when something actually arrived."""
    from app.domains.bounty import notify as bnotify

    assert await bnotify.notify_security_of_new_mail(db_session, rows=[]) == 0


async def test_the_card_names_the_sender_the_subject_and_the_ai_read(
    db_session, monkeypatch,
):
    """「你觉得谁能看的懂这个推送? 你不分析 不给说出重要信息 就是一个提醒?」

    The first version of this card carried a count and a link. Sender,
    subject and the AI's first read all sat on the row already, and they
    are what decide whether somebody stops what they are doing — a card
    that omits them is indistinguishable from a vendor pitch arriving.
    """
    from app.domains.bounty import notify as bnotify

    captured: list = []

    async def _capture(self, env, **_kw):
        captured.append(env)

    monkeypatch.setattr(
        "app.notifications.service.NotificationService.push_inbox_and_feishu",
        _capture,
    )

    await _desk_operator(db_session)
    row = await _queued(
        db_session,
        from_name="Yogesh Y",
        from_email="yogesh@example.com",
        subject="[Vulnerability 4] AllowCleartextPasswords in Dumpling",
        snippet="Hello team It is been pending for a while, the flag allows",
        parsed={"severity": "high", "host": "dumpling", "confidence": "medium"},
    )

    sent = await bnotify.notify_security_of_new_mail(db_session, rows=[row])
    assert sent > 0 and captured, "the desk has to be told at all"

    env = captured[0]
    body = env.context["items_block"].zh
    assert "Yogesh Y" in body and "yogesh@example.com" in body
    assert "AllowCleartextPasswords" in body
    assert "high" in body and "dumpling" in body, "the AI read is the triage"
    assert "Hello team It is been pending" in body, "what they actually wrote"
    # The Feishu chat list shows the title only, so a single-mail poll has
    # to be readable there without opening the card.
    assert "Yogesh Y" in env.title and "AllowCleartextPasswords" in env.title
    # 「去确认」 has to arrive AT the queue. /bounty alone leaves it behind a
    # badged button in the page header — one hop short of the thing the card
    # asked for, which is how a worklist stays at zero rows.
    assert env.url == "/bounty?intake=1"
    for lang in ("zh", "en", "ja"):
        assert getattr(env.context["items_block"], lang), (
            f"{lang} card would render an empty list block"
        )


async def test_the_card_flags_a_thread_that_has_written_before(
    db_session, monkeypatch,
):
    """A thread whose earlier mail is still undecided in the queue is the
    middle case: ``poll`` only auto-attaches when the thread already
    produced a REPORT. Unflagged, the desk opens two reports for one
    conversation — the specific way this ledger gets corrupted."""
    from app.domains.bounty import notify as bnotify

    captured: list = []

    async def _capture(self, env, **_kw):
        captured.append(env)

    monkeypatch.setattr(
        "app.notifications.service.NotificationService.push_inbox_and_feishu",
        _capture,
    )

    await _desk_operator(db_session)
    await _queued(db_session, thread_id="t-repeat", subject="first mail")
    second = await _queued(
        db_session, thread_id="t-repeat", subject="any update on this?",
    )

    await bnotify.notify_security_of_new_mail(db_session, rows=[second])
    assert captured, "the desk has to be told at all"
    assert "thread" in captured[0].context["items_block"].zh, (
        "the repeat-thread warning is missing"
    )


async def test_the_chase_card_carries_what_the_researcher_wrote(
    db_session, monkeypatch,
):
    """Same defect as the queue card, other template: a chase that names
    only the report code makes the reader open the console to find out
    whether the researcher answered a question, sent a PoC, or asked about
    the reward."""
    captured: list = []

    async def _capture(self, env, **_kw):
        captured.append(env)

    monkeypatch.setattr(
        "app.notifications.service.NotificationService.push_inbox_and_feishu",
        _capture,
    )

    await _desk_operator(db_session)
    report = await svc.create_report(
        db_session, title="chased", researcher_email="chase@example.com",
        actor=None,
    )
    row = await _queued(
        db_session, thread_id="t-chase",
        snippet="Any update? I sent the PoC video three weeks ago.",
    )
    await db_session.flush()

    assert await mail_intake.attach_followup(db_session, row, report.id) is True
    assert captured, "a chase is worth interrupting somebody for"
    block = captured[0].context["excerpt_block"]
    assert "PoC video" in block.zh
    for lang in ("zh", "en", "ja"):
        assert "PoC video" in getattr(block, lang)


async def test_a_failed_push_does_not_lose_the_attachment(
    db_session, monkeypatch,
):
    """The follow-up notification is a side effect. If Feishu is down the
    researcher's message must still be attached to the report — losing the
    record to save a push is the wrong trade."""
    from app.domains.bounty import notify as bnotify

    async def _boom(*_a, **_k):
        raise RuntimeError("feishu is down")

    monkeypatch.setattr(
        bnotify, "notify_security_of_researcher_followup", _boom,
    )

    report = await svc.create_report(
        db_session, title="chased", researcher_email="chase@example.com",
        actor=None,
    )
    row = await _queued(db_session, thread_id="t-chase")
    await db_session.flush()

    assert await mail_intake.attach_followup(db_session, row, report.id) is True
    assert row.status == "attached", "the attachment survives a failed push"


async def test_the_ai_read_actually_lands_on_the_queued_row(
    db_session, monkeypatch,
):
    """`parsed` was NULL on every mail the live poller ever queued.

    `suggest_from_titles` narrows the code the model echoes back with
    `_clean_str(..., limit=32)`; a UUID renders as 36 characters. So the
    returned dict was keyed by a 32-char PREFIX while `_annotate_pending`
    looked up the full `str(row.id)` — every `.get()` missed. Silent by
    construction: the AI call succeeded, the model answered correctly, and
    nothing logged. Found on the first real mail the live poller queued
    (2026-08-28), by noticing the card carried no AI line.

    The stub echoes the code it was GIVEN, the way a model does, so the
    truncation happens inside the real `suggest_from_titles` — stubbing
    that function instead would assert its own fixture.
    """
    from app.domains.bounty import ai as bounty_ai

    seen_codes: list[str] = []

    class _Client:
        async def chat_json(self, **kwargs):
            import json
            import re

            # _backfill_prompt renders one - code: <key> line per item.
            user = kwargs.get("user") or ""
            codes = re.findall(r"^\s*- code: (\S+)$", user, re.M)
            seen_codes.extend(codes)
            return json.loads(json.dumps({"items": [
                {"code": c, "severity": "high", "host": "example.com",
                 "internal": False, "confidence": "medium"}
                for c in codes
            ]}))

    async def _build(_db):
        return _Client()

    monkeypatch.setattr(bounty_ai, "_build_client", _build)

    # Built WITHOUT `parsed`, the way `poll` builds it. The shared `_queued`
    # helper passes `parsed=None` explicitly, and `JSON.none_as_null` is
    # False by default — a Python None becomes the JSON literal `null`, which
    # `parsed IS NULL` does not match. The row would then be invisible to
    # `_annotate_pending` and this test would pass for the wrong reason.
    row = BountyMailIntake(
        message_id=f"<{uuid.uuid4()}@example.com>",
        from_email="researcher@example.com",
        from_name="A Researcher",
        subject="XSS on example.com",
        snippet="reflected in the search parameter",
        received_at=datetime.now(UTC) - timedelta(days=1),
        status="pending",
    )
    db_session.add(row)
    await db_session.flush()

    await mail_intake._annotate_pending(db_session)

    assert seen_codes, "the prompt has to carry a code the model can echo"
    assert all(len(c) <= 32 for c in seen_codes), (
        "a key longer than _clean_str's 32-char limit comes back truncated "
        "and can never match — that is the whole bug"
    )
    assert row.parsed is not None, (
        "the AI's reading never reached the row: keys did not match"
    )
    assert row.parsed.get("severity") == "high"


async def test_a_chase_on_an_imported_report_can_be_folded_in(
    client, db_session, admin_headers,
):
    """The queue's third option, and why it had to exist.

    ``poll`` auto-attaches a follow-up only when the thread id it read
    already produced a report — and **Gmail thread ids are per account**.
    The 149 FY27 reports carry ids from the platform owner's mailbox while
    the live poller reads ``bounty-intake@``, so a chase on any of them
    arrives looking like a brand-new finding. The first mail the live
    poller ever queued (2026-08-28) was exactly that: ``Re: [Bug Bounty]
    IDOR - Cross-Tenant Metrics Exposure…``, a researcher asking for an
    update after ten days on a report that already existed.

    Before this the operator could only accept it (a duplicate report for
    one conversation — every count, the budget attribution and the reward
    owed are computed off these rows) or ignore it (drop the chase and
    leave the researcher waiting).
    """
    report = await svc.create_report(
        db_session,
        title="[Bug Bounty] IDOR - Cross-Tenant Metrics Exposure",
        researcher_email="rhyfk0z@example.com", actor=None,
    )
    # A different thread id on purpose: this is the per-account mismatch
    # that makes the automatic path unreachable.
    row = await _queued(
        db_session, thread_id="thread-from-the-other-mailbox",
        subject="Re: [Bug Bounty] IDOR - Cross-Tenant Metrics Exposure",
        snippet="Any update on this report? Still reproducible.",
    )
    await db_session.commit()

    # The queue offers the target rather than making the operator search.
    listed = await client.get(f"{BASE}/mail-intake", headers=admin_headers)
    assert listed.status_code == 200, listed.text
    item = next(
        x for x in listed.json()["data"]["items"] if x["id"] == str(row.id)
    )
    assert item["suggested_report"] is not None, (
        "the subject matches an existing report after stripping Re:"
    )
    assert item["suggested_report"]["code"] == report.code

    res = await client.post(
        f"{BASE}/mail-intake/{row.id}/attach",
        json={"report_id": str(report.id)}, headers=admin_headers,
    )
    assert res.status_code == 200, res.text
    assert res.json()["data"]["code"] == report.code

    await db_session.refresh(row)
    assert row.status == "attached"
    assert row.report_id == report.id

    # The researcher's words land on the report's own timeline, which is
    # the entire point — a chase nobody can read is a chase nobody answers.
    events = await svc.list_events(db_session, report.id, internal=True)
    assert any(
        "Any update on this report" in (e.body or "") for e in events
    ), "the follow-up has to be readable on the report"

    # And it is not offered twice.
    again = await client.post(
        f"{BASE}/mail-intake/{row.id}/attach",
        json={"report_id": str(report.id)}, headers=admin_headers,
    )
    assert again.status_code == 409, again.text


async def test_a_first_report_is_never_offered_as_a_followup(
    client, db_session, admin_headers,
):
    """A new mail whose title resembles an old report is a NEW finding.

    Offering to fold it into the old one invites exactly the mistake the
    suggestion exists to prevent, in reverse — and a merged-away report
    means a researcher never gets paid. So the suggestion requires the
    subject to actually look like a reply.
    """
    await svc.create_report(
        db_session, title="SQL injection on example.com",
        researcher_email="first@example.com", actor=None,
    )
    row = await _queued(
        db_session, subject="SQL injection on example.com",
        from_email="second@example.com",
    )
    await db_session.commit()

    listed = await client.get(f"{BASE}/mail-intake", headers=admin_headers)
    item = next(
        x for x in listed.json()["data"]["items"] if x["id"] == str(row.id)
    )
    assert item["suggested_report"] is None, (
        "no Re:/Fwd: prefix — this is a first report, not a chase"
    )


async def test_a_bare_re_subject_is_not_enough_to_suggest_a_report(
    db_session,
):
    """`Re: update?` matches half the ledger. A suggestion that fires on a
    12-character subject is worse than none, because the operator stops
    reading it."""
    report = await svc.create_report(
        db_session, title="hi", researcher_email="x@example.com", actor=None,
    )
    assert report is not None
    row = await _queued(db_session, subject="Re: hi")
    assert await mail_intake.suggest_report_for(db_session, row) is None


def test_the_gmail_snippet_is_unescaped_but_a_plain_part_is_not():
    """Gmail returns ``snippet`` HTML-escaped; a base64 text/plain part is not.

    Unescaped, the first real chase the live poller ingested reached the
    desk's Feishu card and BB-2026-144's own timeline as
    ``it&#39;s still reproducible``. Cosmetic in the database, not cosmetic
    in a card somebody is meant to read and act on.

    The plain-text branch must NOT be unescaped: a researcher who literally
    types ``&amp;`` in a plaintext mail means those five characters, and
    "fixing" that would corrupt the one case where the text is already
    what it says.
    """
    import base64

    assert mail_intake._decode_snippet(
        {"snippet": "it&#39;s still reproducible &amp; unpatched"},
    ) == "it's still reproducible & unpatched"

    encoded = base64.urlsafe_b64encode(
        b"literal &amp; typed by the researcher",
    ).decode()
    assert mail_intake._decode_snippet({
        "payload": {"mimeType": "text/plain", "body": {"data": encoded}},
    }) == "literal &amp; typed by the researcher"