"""Defects the post-delivery review found, and the guards that keep them
fixed.

Each test here names a real failure the feature shipped with — every one
of them silent, which is why they survived the per-PR test suites.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest

from app.domains.approval_registry import confirmation as conf
from app.domains.approval_registry import constants as const
from app.domains.approval_registry import notify
from app.domains.approval_registry import requests as reqs
from app.domains.approval_registry import service as svc
from app.identity.person_models import IdentityPerson

BASE = "/api/v1/approval-flows"
SELF = "/api/v1/me/approval-flows"


async def _person(db_session, email, *, name="P", status="active"):
    person = IdentityPerson(
        id=uuid.uuid4(), work_email=email, display_name=name,
        employment_status=status, source_system="successfactors",
        hris_person_id=f"sf-{uuid.uuid4().hex[:10]}",
    )
    db_session.add(person)
    await db_session.flush()
    await db_session.commit()
    return person


@pytest.mark.asyncio
async def test_the_flow_is_ready_notification_actually_sends(
    db_session, employee_user,
):
    """The payoff message of the whole feature.

    ``ApprovalFlow.owner`` is ``lazy="raise"``. A flow that was just
    CREATED (rather than adopted from the discovery sweep) is persistent
    but has never loaded that relationship, so reading it to name the
    owner raises — and the route wraps the push in a try/except, so the
    requester simply never hears that their flow is ready. The common
    path was the broken one.
    """
    person, _ = employee_user
    row = await reqs.submit_request(
        db_session, requester=person,
        proposed_name="Request to join the GitHub organization",
        purpose="新同事需要访问公司的 GitHub 组织。",
    )
    await reqs.decide_request(db_session, row, approve=True, actor=None)
    await db_session.commit()
    row = await reqs.get_request(db_session, row.id)

    flow = await reqs.register_built_flow(
        db_session, row, approval_code=str(uuid.uuid4()).upper(), actor=None,
    )
    await db_session.commit()

    # ``flow`` here is the freshly created object, exactly what the route
    # hands to the notifier — that is the shape that used to raise.
    sent = await notify.notify_flow_request_registered(db_session, row, flow)
    assert sent == 1


@pytest.mark.asyncio
async def test_nobody_is_chased_about_a_card_that_never_arrived(
    db_session,
):
    """Escalation copies the owner's MANAGER. Doing that over a card the
    platform failed to deliver blames somebody for silence they were
    never given the chance to break.
    """
    owner = await _person(db_session, f"n-{uuid.uuid4().hex[:6]}@t.com")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="流程",
        owner_person_id=owner.id,
    )
    await db_session.commit()
    round_row, _ = await conf.open_round(db_session, label="2026-REVIEW-A")
    await db_session.commit()

    item = (await conf.round_items(db_session, round_row.id))[0]
    # Old enough to chase, but the card never went out.
    item.created_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=30)
    item.notified_at = None
    await db_session.commit()

    assert await conf.overdue_items(db_session, round_row) == []

    # Once the ask actually reached them, the clock is real.
    item.notified_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=30)
    await db_session.commit()
    assert len(await conf.overdue_items(db_session, round_row)) == 1
    assert flow.owner_person_id == owner.id


@pytest.mark.asyncio
async def test_the_card_landing_page_can_style_itself(client):
    """The platform's default ``Content-Security-Policy: default-src
    'none'`` makes the browser drop the page's inline <style>, so the
    one-click page the half-year card links to renders as bare unstyled
    HTML. Both earlier one-click pages hit this and set their own
    header; this one did not.
    """
    resp = await client.get(f"{BASE}/confirm-by-token", params={"token": "nope"})
    assert resp.status_code == 200
    csp = resp.headers.get("Content-Security-Policy", "")
    assert "style-src 'unsafe-inline'" in csp
    # The page's only interactive element is a same-origin POST form;
    # ``form-action`` does not fall back to ``default-src``, so it has to
    # be stated or a future default could break the button.
    assert "form-action 'self'" in csp


@pytest.mark.asyncio
async def test_two_admins_opening_a_round_at_once_do_not_500(db_session):
    """``label`` is UNIQUE. Two operators pressing 「立即开盘」 together
    both see "no round yet" and both insert — the loser gets an
    IntegrityError, which is a 500 for doing something reasonable.
    """
    owner = await _person(db_session, f"r-{uuid.uuid4().hex[:6]}@t.com")
    await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="流程",
        owner_person_id=owner.id,
    )
    await db_session.commit()

    first, _ = await conf.open_round(db_session, label="2026-REVIEW-B")
    await db_session.commit()

    # The racing caller: the row already exists, but this call's
    # pre-check still says it does not — precisely what two concurrent
    # operators see. Blind for the PRE-CHECK only; the conflict-recovery
    # read must find the winner on its own, as it would in production.
    #
    # The recovery read has to be a LOCKING one: under REPEATABLE READ a
    # plain consistent read keeps this transaction's snapshot and misses
    # a winner committed after it, so the recovery would re-raise the
    # 500 it just caught. Assert the lock is actually asked for, because
    # a single-session test cannot reproduce the snapshot itself.
    original = conf.get_round_by_label
    calls: list[bool] = []

    async def _blind(db, label, *, lock: bool = False):
        calls.append(lock)
        if len(calls) == 1:
            return None
        return await original(db, label, lock=lock)

    conf.get_round_by_label = _blind  # type: ignore[assignment]
    try:
        second, created_second = await conf.open_round(
            db_session, label="2026-REVIEW-B",
        )
    finally:
        conf.get_round_by_label = original  # type: ignore[assignment]
    await db_session.commit()
    assert created_second is False
    assert second.id == first.id
    assert calls == [False, True], (
        "the conflict-recovery read must lock, or it keeps a snapshot "
        "taken before the winner committed and returns None"
    )


@pytest.mark.asyncio
async def test_registering_a_request_nobody_decided_is_refused(
    db_session, employee_user,
):
    """The error message said "only an approved request can be
    completed" while the code also accepted ``pending`` — an accidental
    path that registers a flow and marks the request decided without
    anyone deciding it.
    """
    person, _ = employee_user
    row = await reqs.submit_request(
        db_session, requester=person, proposed_name="未经批准的流程",
        purpose="这是一个足够长的用途说明。",
    )
    await db_session.commit()

    from app.core.exceptions import ConflictError

    with pytest.raises(ConflictError):
        await reqs.register_built_flow(
            db_session, row, approval_code=str(uuid.uuid4()).upper(), actor=None,
        )


@pytest.mark.asyncio
async def test_bulk_confirm_cannot_take_back_a_flow_reassigned_mid_round(
    db_session, employee_user,
):
    """The card's 「全部确认」 answers every item still marked pending.

    If IT re-pointed one of those flows to somebody else after the round
    opened, the item is still pending against the ORIGINAL owner — and
    confirming it wrote that owner back onto the flow, silently taking
    it from the person who now owns it. The round asked "is this still
    yours"; the honest answer when it is not is that the question no
    longer applies, not a change of ownership.
    """
    person, _ = employee_user
    successor = await _person(db_session, f"su-{uuid.uuid4().hex[:6]}@t.com")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="流程",
        owner_person_id=person.id,
    )
    await db_session.commit()

    round_row, _ = await conf.open_round(db_session, label="2026-REVIEW-C")
    await db_session.commit()

    # IT re-points it mid-round.
    await svc.set_owner(
        db_session, flow, owner_person_id=successor.id,
        owner_source=const.OWNER_SOURCE_MANUAL,
    )
    await db_session.commit()

    answered = await conf.answer_all_for_owner(
        db_session, round_row, actor=person, via="token",
    )
    await db_session.commit()

    fresh = await svc.get_flow(db_session, flow.id)
    assert fresh.owner_person_id == successor.id, (
        "confirming an item whose flow moved on must not re-assign it"
    )
    assert answered == 0
    item = (await conf.round_items(db_session, round_row.id))[0]
    assert item.state == const.ITEM_DROPPED


@pytest.mark.asyncio
async def test_a_card_token_can_only_answer_its_own_holders_items(
    client, admin_headers, db_session, employee_user,
):
    """The card link is unauthenticated by design — the token's
    signature IS the authorisation. So the one thing that must hold is
    that it answers for exactly the person it was minted for, even
    though the round contains other people's items.
    """
    from app.core.security import create_action_token

    holder, _ = employee_user
    other = await _person(db_session, f"o-{uuid.uuid4().hex[:6]}@t.com")
    mine = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="我的流程",
        owner_person_id=holder.id,
    )
    theirs = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="别人的流程",
        owner_person_id=other.id,
    )
    await db_session.commit()

    opened = (
        await client.post(
            f"{BASE}/rounds/open-now", json={}, headers=admin_headers,
        )
    ).json()["data"]
    token = create_action_token(
        purpose=const.ACTION_TOKEN_PURPOSE,
        subject_id=str(holder.id), resource_id=opened["round"]["id"],
        ttl_days=const.ACTION_TOKEN_TTL_DAYS,
    )

    page = await client.get(f"{BASE}/confirm-by-token", params={"token": token})
    assert "我的流程" in page.text
    # The page must not disclose a colleague's flows to whoever holds
    # this link — the round has both, the token has one.
    assert "别人的流程" not in page.text

    await client.post(f"{BASE}/confirm-by-token", data={"token": token})

    round_row = await conf.get_round(db_session, uuid.UUID(opened["round"]["id"]))
    states = {i.flow_id: i.state for i in await conf.round_items(db_session, round_row.id)}
    assert states[mine.id] == const.ITEM_CONFIRMED
    assert states[theirs.id] == const.ITEM_PENDING


@pytest.mark.asyncio
async def test_a_stale_confirmation_is_something_the_owner_is_asked_to_redo(
    client, employee_headers, employee_user, db_session,
):
    """Two surfaces, one fact.

    The admin's ``unconfirmed`` filter counts a confirmation older than
    the validity horizon, and the row renders as 「确认已过期」 — but the
    employee's own page computed ``needs_confirmation`` from a different
    rule and told them nothing needed doing. The person being asked and
    the person chasing them must see the same list.
    """
    person, _ = employee_user
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="旧确认",
        owner_person_id=person.id,
    )
    await svc.set_owner(
        db_session, flow, owner_person_id=person.id,
        owner_source=const.OWNER_SOURCE_CONFIRMED, confirmed=True,
    )
    flow.owner_confirmed_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(
        days=const.CONFIRMATION_VALID_DAYS + 30,
    )
    await db_session.commit()

    mine = (await client.get(SELF, headers=employee_headers)).json()["data"]
    assert mine["flows"][0]["owner_state"] == "stale"
    assert mine["needs_confirmation"] == 1


@pytest.mark.asyncio
async def test_the_assistant_does_not_call_a_stale_confirmation_a_guess(
    db_session,
):
    """「平台推断，尚未本人确认」 is false for a flow the owner DID
    confirm, just long ago. Telling a colleague their名字 was guessed
    when they personally stood behind it is a different claim."""
    from app.assistant import flow_grounding as flows

    owner = await _person(db_session, f"st-{uuid.uuid4().hex[:6]}@t.com", name="王五")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="报销申请",
        owner_person_id=owner.id,
    )
    await svc.set_owner(
        db_session, flow, owner_person_id=owner.id,
        owner_source=const.OWNER_SOURCE_CONFIRMED, confirmed=True,
    )
    flow.owner_confirmed_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(
        days=const.CONFIRMATION_VALID_DAYS + 30,
    )
    await db_session.commit()

    grounding = await flows.build_flow_grounding(db_session, "报销怎么申请")
    assert "王五" in grounding.system_context
    assert "平台推断" not in grounding.system_context


@pytest.mark.asyncio
async def test_a_round_whose_cards_never_sent_is_retried_not_lost(
    db_session,
):
    """The opening fan-out happens once, twice a year.

    If Feishu is unreachable in that minute every card is lost — and
    escalation deliberately will not chase an ask nobody received, so
    without a retry the whole round sits unasked until it closes and
    lapses in full. The daily escalator has to pick these up.
    """
    owner = await _person(db_session, f"rs-{uuid.uuid4().hex[:6]}@t.com")
    await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="流程",
        owner_person_id=owner.id,
    )
    await db_session.commit()
    round_row, _ = await conf.open_round(db_session, label="2026-REVIEW-D")
    await db_session.commit()

    # The open ran, the fan-out failed: pending, never notified.
    pending = await conf.unnotified_items(db_session, round_row)
    assert len(pending) == 1

    # Nobody is chased for it, however old it gets.
    item = pending[0]
    item.created_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=60)
    await db_session.commit()
    assert await conf.overdue_items(db_session, round_row) == []

    # Once the retry delivers it, it re-enters the normal flow.
    item.notified_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(days=20)
    await db_session.commit()
    assert await conf.unnotified_items(db_session, round_row) == []
    assert len(await conf.overdue_items(db_session, round_row)) == 1


# ── Round 4: the handover nobody announced ─────────────────────────


def _capture_pushes(monkeypatch):
    """Record every envelope the notifier hands to the push layer."""
    sent: list = []

    async def _fake(self, env, *, feishu=True):  # noqa: ANN001
        sent.append(env)

    monkeypatch.setattr(
        "app.notifications.service.NotificationService.push_inbox_and_feishu",
        _fake,
    )
    return sent


def _owner_assigned(sent):
    return [
        e for e in sent
        if e.template_key == "approval_registry.owner_assigned"
    ]


@pytest.mark.asyncio
async def test_the_person_a_flow_is_transferred_to_is_told(
    db_session, monkeypatch,
):
    """A flow could change hands in total silence.

    Alice hands her flow to Bob in the half-year round. Bob was never
    notified by anything: he became accountable for a flow he had never
    heard of, and would first learn of it either from a confirmation
    card six months later or from a colleague's question he could not
    answer. Handing somebody a responsibility is itself an event.
    """
    alice = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com",
                          name="Alice")
    bob = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@t.com",
                        name="Bob")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(),
        name="Request to join the GitHub organization",
        owner_person_id=alice.id,
    )
    await db_session.commit()
    round_row, _ = await conf.open_round(db_session, label="2026-REVIEW-H")
    await db_session.commit()
    item = next(
        i for i in await conf.round_items(db_session, round_row.id)
        if i.flow_id == flow.id
    )

    sent = _capture_pushes(monkeypatch)
    await conf.answer_item(
        db_session, item, actor=alice, keep=False,
        transfer_to_person_id=bob.id,
    )
    await db_session.commit()

    handovers = _owner_assigned(sent)
    assert len(handovers) == 1
    env = handovers[0]
    assert env.recipient.id == bob.id
    # It says who handed it over — "you own this now" without saying who
    # it came from is a message the recipient cannot act on.
    assert "Alice" in env.context["handover_line"].zh


@pytest.mark.asyncio
async def test_the_english_card_carries_no_chinese(db_session, monkeypatch):
    """The producer interpolates the sentence, the catalogue is
    localised — so a per-language VALUE has to be a ``LocalizedText``,
    not a ``{key}_en`` the English body never references. Getting that
    wrong ships an English card with a Chinese sentence in the middle,
    which is precisely what a production render of the outbox once
    found.

    Scope note: this pins the CHROME — the sentences Helix writes — not
    the data. A flow genuinely named 「VPN 账号申请」 stays Chinese in
    every language's card, which is correct; that is why the fixture
    uses an English-named flow. Do not "fix" a future red here by
    accepting Chinese chrome.
    """
    from app.notifications.service import _localized_context
    from app.notifications.templates.approval_registry import TEMPLATES

    alice = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com",
                          name="Alice")
    bob = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@t.com",
                        name="Bob")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(),
        name="Access request", owner_person_id=alice.id,
        description="Ask for access to the shared drive.",
    )
    await db_session.commit()

    sent = _capture_pushes(monkeypatch)
    assert await notify.notify_owner_assigned(
        db_session, [flow], new_owner_id=bob.id, previous_owner_id=alice.id,
        reason=notify.HANDOVER_TRANSFER, actor=alice,
    ) == 1

    ctx = _localized_context(sent[0].context, "en-US")
    body = TEMPLATES["approval_registry.owner_assigned"]["body_text_en"]
    rendered = body.format(**ctx)
    cjk = [c for c in rendered if "一" <= c <= "鿿"]
    assert not cjk, f"English card contains Chinese: {''.join(cjk)}"
    assert "Alice" in rendered


@pytest.mark.asyncio
async def test_the_nightly_inference_does_not_notify_anybody(
    db_session, monkeypatch,
):
    """The storm guard, and the reason this push is wired per-path
    rather than inside ``set_owner``.

    Inference assigns owners in BULK. Hooking the announcement into the
    shared setter would fire one card per inferred flow at every person
    on the first night after deploy — the 2026-06-12 failure exactly.
    Only a human handover announces itself.
    """
    carol = await _person(db_session, f"c-{uuid.uuid4().hex[:6]}@t.com",
                          name="Carol")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="推断流程",
    )
    await db_session.commit()

    # Load the person the way the sweep does. ``apply_inferred_owner``
    # reads ``person.department``, which walks the ``org_memberships``
    # relationship — handing it a bare object built by this fixture
    # raises ``MissingGreenlet`` on a lazy load and tests a shape the
    # producer never passes.
    from sqlalchemy import select
    from sqlalchemy.orm import selectinload

    loaded = (
        await db_session.execute(
            select(IdentityPerson)
            .where(IdentityPerson.id == carol.id)
            .options(selectinload(IdentityPerson.org_memberships)),
        )
    ).scalar_one()

    sent = _capture_pushes(monkeypatch)
    await svc.apply_inferred_owner(
        db_session, flow, person=loaded,
        evidence={"share": 0.9, "tasks": 12},
    )
    await db_session.commit()

    assert flow.owner_person_id == carol.id
    assert _owner_assigned(sent) == []


@pytest.mark.asyncio
async def test_confirming_yourself_does_not_announce_a_handover(
    db_session, monkeypatch,
):
    """「还是我」 is not a transfer. Telling somebody they now own the
    flow they just confirmed they already owned reads as a mistake and
    trains people to ignore the card."""
    alice = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com",
                          name="Alice")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="自己的流程",
        owner_person_id=alice.id,
    )
    await db_session.commit()
    round_row, _ = await conf.open_round(db_session, label="2026-REVIEW-K")
    await db_session.commit()
    item = next(
        i for i in await conf.round_items(db_session, round_row.id)
        if i.flow_id == flow.id
    )

    sent = _capture_pushes(monkeypatch)
    await conf.answer_item(db_session, item, actor=alice, keep=True)
    await db_session.commit()
    assert _owner_assigned(sent) == []


@pytest.mark.asyncio
async def test_the_offboarding_successor_hears_about_it(
    db_session, monkeypatch,
):
    """Offboarding tells the LEAVER what is leaving their hands. Nothing
    told the person receiving it."""
    from app.core.coordinators import approval_registry_coordinator as coord

    leaver = await _person(db_session, f"l-{uuid.uuid4().hex[:6]}@t.com",
                           name="Leaver")
    lead = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@t.com",
                         name="Lead")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="交接流程",
        owner_person_id=leaver.id,
    )
    await db_session.commit()

    sent = _capture_pushes(monkeypatch)
    moved = await coord.handover_flows(
        db_session, from_person_id=leaver.id, to_person_id=lead.id,
    )
    await db_session.commit()

    assert moved == ["交接流程"]
    handovers = _owner_assigned(sent)
    assert len(handovers) == 1
    assert handovers[0].recipient.id == lead.id
    assert "离职" in handovers[0].context["handover_line"].zh
    assert flow.owner_person_id == lead.id

    # Idempotent: a second scan finds nothing to move and says nothing.
    sent.clear()
    assert await coord.handover_flows(
        db_session, from_person_id=leaver.id, to_person_id=lead.id,
    ) == []
    assert _owner_assigned(sent) == []


def test_every_template_is_complete_in_three_languages():
    """Catches the shape this domain's newest template got wrong on the
    first attempt: a body whose ``_en``/``_ja`` variants are missing, or
    whose placeholders drift apart, renders a KeyError or the wrong
    language at a real person."""
    import re

    from app.notifications.templates.approval_registry import TEMPLATES

    for key, fields in TEMPLATES.items():
        base = {f for f in fields if not f.endswith(("_en", "_ja"))}
        assert base, key
        for field in base:
            for variant in (f"{field}_en", f"{field}_ja"):
                assert variant in fields, f"{key}.{variant} missing"
            placeholders = {
                frozenset(re.findall(r"{(\w+)}", fields[f]))
                for f in (field, f"{field}_en", f"{field}_ja")
            }
            assert len(placeholders) == 1, f"{key}.{field}: {placeholders}"


@pytest.mark.asyncio
async def test_one_handover_sends_one_card_not_one_per_flow(
    db_session, monkeypatch,
):
    """This domain's stated rule is 「一人一张卡,不是一流程一张卡」 and my
    own first cut of the handover push broke it: a leaver who owned five
    flows handed their lead five separate 「你现在负责…」 cards for one
    action. That is the shape people learn to ignore.
    """
    from app.core.coordinators import approval_registry_coordinator as coord

    leaver = await _person(db_session, f"l-{uuid.uuid4().hex[:6]}@t.com",
                           name="Leaver")
    lead = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@t.com",
                         name="Lead")
    names = [f"流程-{i}" for i in range(5)]
    for name in names:
        await svc.create_flow(
            db_session, approval_code=str(uuid.uuid4()).upper(), name=name,
            owner_person_id=leaver.id,
        )
    await db_session.commit()

    sent = _capture_pushes(monkeypatch)
    moved = await coord.handover_flows(
        db_session, from_person_id=leaver.id, to_person_id=lead.id,
    )
    await db_session.commit()

    assert sorted(moved) == sorted(names)
    cards = _owner_assigned(sent)
    assert len(cards) == 1, f"{len(cards)} cards for one handover"
    # And the one card names every flow, so it is a summary rather than
    # a truncation the reader cannot detect.
    listed = cards[0].context["flow_list"].zh
    for name in names:
        assert name in listed
    assert "5" in cards[0].context["flow_summary"].zh


@pytest.mark.asyncio
async def test_a_long_handover_says_how_many_it_left_out(
    db_session, monkeypatch,
):
    """Bounded lists state their overflow. A card that silently shows 12
    of 20 reads as a complete inventory of what somebody just took on."""
    from app.core.coordinators import approval_registry_coordinator as coord
    from app.domains.approval_registry import constants as c

    leaver = await _person(db_session, f"l-{uuid.uuid4().hex[:6]}@t.com")
    lead = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@t.com")
    extra = 3
    for i in range(c.MAX_LISTED_FLOWS + extra):
        await svc.create_flow(
            db_session, approval_code=str(uuid.uuid4()).upper(),
            name=f"多流程-{i}", owner_person_id=leaver.id,
        )
    await db_session.commit()

    sent = _capture_pushes(monkeypatch)
    await coord.handover_flows(
        db_session, from_person_id=leaver.id, to_person_id=lead.id,
    )
    await db_session.commit()

    card = _owner_assigned(sent)[0]
    listed = card.context["flow_list"]
    assert f"另有 {extra} 条" in listed.zh
    assert f"and {extra} more" in listed.en
    # The overflow line is prose and needs its own translation — reusing
    # the Chinese one puts 「另有 N 条」 at the bottom of a Japanese card.
    assert f"ほか {extra} 件" in listed.ja
    assert "另有" not in listed.ja


@pytest.mark.asyncio
async def test_a_transfer_outside_any_open_round_is_announced_too(
    db_session, monkeypatch,
):
    """The path per-call-site wiring missed.

    Between rounds there is no confirmation item, so ``record_self_answer``
    writes the register directly. Wiring the announcement at each call
    site left exactly this one silent — which is why the push now lives
    in ``set_owner`` and defaults to announcing: a caller can opt out on
    purpose, but cannot forget.
    """
    alice = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com",
                          name="Alice")
    bob = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@t.com",
                        name="Bob")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="轮次外的流程",
        owner_person_id=alice.id,
    )
    await db_session.commit()
    assert await conf.current_open_round(db_session) is None

    sent = _capture_pushes(monkeypatch)
    result = await conf.record_self_answer(
        db_session, flow, actor=alice, keep=False,
        transfer_to_person_id=bob.id,
    )
    await db_session.commit()

    assert result["round"] is None
    cards = _owner_assigned(sent)
    assert len(cards) == 1
    assert cards[0].recipient.id == bob.id
    assert flow.owner_person_id == bob.id


@pytest.mark.asyncio
async def test_a_bulk_import_does_not_card_the_whole_company(
    db_session, monkeypatch,
):
    """Paste-import is how this register got seeded. Announcing there
    would push a card at most of the company in one click — the storm
    the round-opening window guard exists to prevent."""
    people = [
        await _person(db_session, f"i{i}-{uuid.uuid4().hex[:5]}@t.com",
                      name=f"Owner{i}")
        for i in range(4)
    ]
    codes = [str(uuid.uuid4()).upper() for _ in people]
    # Already in the register, ownerless — the shape the seeding paste
    # actually lands on, and the one whose owner assignment runs through
    # ``set_owner``. Creating them fresh would exercise a different
    # branch and pass without touching the opt-out at all.
    for i, code in enumerate(codes):
        await svc.upsert_discovered(
            db_session, approval_code=code, name=f"导入流程-{i}",
            source_group=None, observed_instances=1, last_instance_at=None,
        )
    await db_session.commit()
    rows = [
        {"approval_code": code, "name": f"导入流程-{i}",
         "owner_email": people[i].work_email}
        for i, code in enumerate(codes)
    ]

    sent = _capture_pushes(monkeypatch)
    result = await svc.import_flows(db_session, rows, actor=None)
    await db_session.commit()

    assert result["updated"] == 4
    for i, code in enumerate(codes):
        flow = await svc.get_by_code(db_session, code)
        assert flow is not None and flow.owner_person_id == people[i].id
    assert _owner_assigned(sent) == []


@pytest.mark.asyncio
async def test_registering_a_flow_against_somebody_tells_them(
    db_session, monkeypatch,
):
    """Naming an owner on the 「登记流程」 form is the same act as
    transferring one to them, and it was equally silent."""
    dave = await _person(db_session, f"d-{uuid.uuid4().hex[:6]}@t.com",
                         name="Dave")

    sent = _capture_pushes(monkeypatch)
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(),
        name="手工登记的流程", owner_person_id=dave.id,
    )
    await db_session.commit()

    cards = _owner_assigned(sent)
    assert len(cards) == 1
    assert cards[0].recipient.id == dave.id
    assert flow.name in cards[0].context["flow_list"].zh


@pytest.mark.asyncio
async def test_registering_an_ownerless_flow_tells_nobody(
    db_session, monkeypatch,
):
    """No owner, no addressee. The discovery sweep creates hundreds of
    these."""
    sent = _capture_pushes(monkeypatch)
    await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="无主流程",
    )
    await db_session.commit()
    assert _owner_assigned(sent) == []


@pytest.mark.asyncio
async def test_registering_a_request_says_it_once_not_twice(
    db_session, employee_user, monkeypatch,
):
    """The requester gets ``flow_request_registered`` — a message that
    knows what they asked for. A generic 「你现在是负责人」 next to it is
    the same fact twice, half a second apart."""
    person, _ = employee_user
    row = await reqs.submit_request(
        db_session, requester=person, proposed_name="双重通知检查",
        purpose="确认登记只发一条消息。",
    )
    await reqs.decide_request(db_session, row, approve=True, actor=None)
    await db_session.commit()
    row = await reqs.get_request(db_session, row.id)

    code = str(uuid.uuid4()).upper()
    # The adopt branch: the sweep already found this flow, owned by
    # nobody, and registration re-points it at the requester.
    await svc.upsert_discovered(
        db_session, approval_code=code, name="双重通知检查",
        source_group=None, observed_instances=1, last_instance_at=None,
    )
    await db_session.commit()

    sent = _capture_pushes(monkeypatch)
    await reqs.register_built_flow(
        db_session, row, approval_code=code, actor=None,
    )
    await db_session.commit()
    assert _owner_assigned(sent) == []


@pytest.mark.asyncio
async def test_a_broken_notifier_does_not_undo_the_handover(
    db_session, monkeypatch,
):
    """The router's rule is 「先 commit 业务，再通知」 — a push must never
    undo the thing it announces. Moving the push into ``set_owner`` buys
    「没有调用点会忘」 and costs that ordering, so the guard buys it back.

    Also pins the savepoint: a swallowed flush failure poisons the
    session, and the NEXT statement — the caller's — dies with an error
    about something it never did.
    """
    alice = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com")
    bob = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@t.com")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="通知会炸的流程",
        owner_person_id=alice.id,
    )
    await db_session.commit()

    async def _boom(self, env, *, feishu=True):  # noqa: ANN001
        raise RuntimeError("Feishu is down and so is the outbox")

    monkeypatch.setattr(
        "app.notifications.service.NotificationService.push_inbox_and_feishu",
        _boom,
    )
    await svc.set_owner(db_session, flow, owner_person_id=bob.id, actor=alice)
    # The session is still usable and the handover is durable.
    await db_session.commit()

    reloaded = await svc.get_flow(db_session, flow.id)
    assert reloaded.owner_person_id == bob.id


def test_the_handover_line_never_calls_the_operator_the_leaver():
    """On the offboarding path the actor is the IT admin (or a cron)
    executing the transfer — 「{admin} 离职」 about them is a false
    sentence, and the previous-owner fallback used to produce exactly
    that whenever the leaver could not be resolved by name."""
    line = notify._handover_line(
        notify.HANDOVER_OFFBOARDING, None, "Ada the IT admin",
    )
    assert "Ada" not in line.zh and "Ada" not in line.en
    assert "前负责人" in line.zh

    # The transfer path is the opposite case: there the actor IS the
    # outgoing owner, so falling back to their name is correct.
    line = notify._handover_line(notify.HANDOVER_TRANSFER, None, "Alice")
    assert "Alice" in line.zh and "Alice" in line.en


def test_an_unattributed_assignment_does_not_render_it_twice():
    """A cron or a script has no actor. 「IT(IT)」 reads like a bug."""
    line = notify._handover_line(notify.HANDOVER_ASSIGNED, None, None)
    assert "IT(IT)" not in line.zh
    assert "IT (IT)" not in line.en
    assert line.zh.startswith("IT把")


@pytest.mark.asyncio
async def test_assigning_a_flow_to_yourself_does_not_mail_you(
    db_session, monkeypatch,
):
    """IT registering a flow against themselves is the common case, and
    a card telling you what you just did is noise."""
    admin = await _person(db_session, f"ad-{uuid.uuid4().hex[:6]}@t.com",
                          name="Admin")

    sent = _capture_pushes(monkeypatch)
    await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="我自己登记的",
        owner_person_id=admin.id, actor=admin,
    )
    await db_session.commit()
    assert _owner_assigned(sent) == []


@pytest.mark.asyncio
async def test_set_owner_announces_without_being_asked_to(
    db_session, monkeypatch,
):
    """The default itself, which is the whole point of moving the push
    into the setter: IT re-pointing a flow from the console passes no
    ``announce`` at all, and the new owner still hears about it.

    Mutation-checked: flipping the default to ``None`` must turn this
    red. The transfer paths name their reason explicitly, so they would
    survive that flip and cannot stand in for this test.
    """
    alice = await _person(db_session, f"a-{uuid.uuid4().hex[:6]}@t.com",
                          name="Alice")
    bob = await _person(db_session, f"b-{uuid.uuid4().hex[:6]}@t.com",
                        name="Bob")
    admin = await _person(db_session, f"ad-{uuid.uuid4().hex[:6]}@t.com",
                          name="Admin")
    flow = await svc.create_flow(
        db_session, approval_code=str(uuid.uuid4()).upper(), name="控制台改派",
        owner_person_id=alice.id, actor=admin,
    )
    await db_session.commit()

    sent = _capture_pushes(monkeypatch)
    await svc.set_owner(
        db_session, flow, owner_person_id=bob.id, actor=admin,
    )
    await db_session.commit()

    cards = _owner_assigned(sent)
    assert len(cards) == 1
    assert cards[0].recipient.id == bob.id
    # IT re-pointed it, so the card says so rather than blaming Alice.
    assert "Admin" in cards[0].context["handover_line"].zh