"""年假到期提醒 —— 圈人口径、文案校验、渲染、群发与去重。

这是平台第一个交给 HR(非 global_admin)操作的通知群发面,而且收件人是几百名
员工,所以这里钉住的每一条都是「发错了会当场伤到人」的那一类:

  1. **圈人口径**:只发在职、有工作邮箱、余额过阈值、清零日在窗口内的人;
     已过期的桶(SF 常年不 close 的历史桶)默认不发;离职的人永远不发。
  2. **口径一致**:清零日 = ``booking_end_date or end_date``,和员工自己卡片上
     那个日期是同一个 —— 否则员工会拿着两个日期来问 HR。
  3. **文案护栏**:未知占位符 / 位置参数 / 空标题一律拒绝写入。第一条是
     「员工收到带大括号的 ``{name}``」,第二条会让**整批**投递抛异常。
  4. **快照语义**:HR 点确认时看到的数字冻结进批次,后台任务只读快照 ——
     中间跑一次 HRIS 同步也不会改变发出去的内容。
  5. **去重**:近期已提醒的人默认不再入选;测试发送不污染这个判定。
  6. **RBAC**:employee 一律 403;发送要求真实浏览器会话(PAT 打不开)。
"""
from __future__ import annotations

import pathlib
import uuid
from datetime import UTC, date, datetime, timedelta

import pytest
import pytest_asyncio
from sqlalchemy import select

from app.domains.hr import leave_reminder_service as svc
from app.domains.hr.leave_models import HrLeaveAccount
from app.domains.hr.leave_reminder_models import (
    BATCH_RESOURCE_TYPE,
    BATCH_STATUS_COMPLETED,
    HrLeaveReminderBatch,
)
from app.domains.hr.leave_reminder_schemas import (
    AccountSnapshot,
    ReminderFilters,
    TemplateCopy,
)
from app.identity.person_models import IdentityOrgMembership, IdentityPerson
from app.notifications.models import NotificationOutbox

TODAY = svc.today_sh()


@pytest.fixture(autouse=True)
def _freeze_today(monkeypatch):
    """Pin the service's date anchor to this module's ``TODAY``.

    ``TODAY`` is captured once at import, but the code under test re-reads
    ``today_sh()`` at call time. A full-suite run that crosses Shanghai
    midnight therefore builds fixtures against one date and asserts
    against the next, and every offset-from-``TODAY`` expectation in this
    file slips by a day (4 tests went red that way).

    Patching the module global covers every consumer: the service's own
    call sites resolve ``today_sh`` as a bare module-global at call time,
    and ``leave_reminder_router`` reaches it as ``svc.today_sh()``.
    """
    monkeypatch.setattr(svc, "today_sh", lambda: TODAY)


LEGAL = "CHN_AnnualLeave"
WELFARE = "CHN_PL"
COMP = "CHN_COMPL"


async def _person(
    db, *, name, status="active", email=True, lang=None, dept="Core",
):
    """``email=False`` = 空邮箱。

    ``identity_persons.work_email`` 是 NOT NULL 且唯一,所以「投不出去」在生产
    里唯一可达的形态是**空串**,不是 NULL —— 测试也必须照这个真值来构造,否则
    钉住的是一个不可能发生的场景。
    """
    pid = uuid.uuid4()
    db.add(IdentityPerson(
        id=pid,
        work_email=f"{name}-{pid}@test.com" if email else "",
        display_name=name,
        employment_status=status,
        source_system="successfactors",
        hris_person_id=str(pid)[:8],
        preferred_language=lang,
    ))
    db.add(IdentityOrgMembership(
        person_id=pid,
        source_system="successfactors",
        org_node_source_id=f"node-{dept}-{pid}",
        org_node_type="department",
        org_node_name=dept,
        org_path=f"PingCAP / Alpha / {dept}",
        is_primary=True,
        membership_status="active",
    ))
    return pid


def _account(
    db, pid, *, atype=LEGAL, balance=5.0, entitled=5.0, used=0.0,
    booking_end=None, end=None, fiscal_year=2025, closed=False,
):
    db.add(HrLeaveAccount(
        external_code=str(uuid.uuid4()),
        person_id=str(pid),
        account_type=atype,
        account_type_label=None,
        fiscal_year=fiscal_year,
        start_date=date(fiscal_year, 4, 1),
        end_date=end or date(fiscal_year + 1, 3, 31),
        booking_end_date=booking_end,
        balance_days=balance,
        entitled_days=entitled,
        used_days=used,
        is_current=False,
        account_closed=closed,
        synced_at=datetime.now(UTC),
    ))


def _filters(**kw) -> ReminderFilters:
    base = {
        "account_types": [LEGAL],
        "fiscal_years": [],
        "min_balance_days": 0.5,
        "expiring_within_days": 120,
    }
    base.update(kw)
    return ReminderFilters(**base)


# ── 1. 圈人口径 ─────────────────────────────────────────────────────────


@pytest_asyncio.fixture
async def cohort_data(db_session):
    """一个刚好能把每条口径都试出来的小样本。

    ``soon`` 30 天后清零(应入选);``later`` 300 天后(窗口外);
    ``expired`` 10 天前已过期;``tiny`` 只剩 0.2 天;``leaver`` 已离职;
    ``no_email`` 没有工作邮箱;``welfare_only`` 只有福利年假。
    """
    db = db_session
    ids = {}
    ids["soon"] = await _person(db, name="Soon")
    _account(db, ids["soon"], booking_end=TODAY + timedelta(days=30), balance=5)

    ids["later"] = await _person(db, name="Later")
    _account(db, ids["later"], booking_end=TODAY + timedelta(days=300), balance=3)

    ids["expired"] = await _person(db, name="Expired")
    _account(db, ids["expired"], booking_end=TODAY - timedelta(days=10), balance=2)

    ids["tiny"] = await _person(db, name="Tiny")
    _account(db, ids["tiny"], booking_end=TODAY + timedelta(days=20), balance=0.2)

    ids["leaver"] = await _person(db, name="Leaver", status="left")
    _account(db, ids["leaver"], booking_end=TODAY + timedelta(days=15), balance=8)

    ids["no_email"] = await _person(db, name="NoMail", email=False)
    _account(db, ids["no_email"], booking_end=TODAY + timedelta(days=15), balance=4)

    ids["welfare_only"] = await _person(db, name="WelfareOnly")
    _account(
        db, ids["welfare_only"], atype=WELFARE,
        booking_end=TODAY + timedelta(days=25), balance=5.5,
    )

    ids["both"] = await _person(db, name="Both")
    _account(db, ids["both"], booking_end=TODAY + timedelta(days=30), balance=1)
    _account(
        db, ids["both"], atype=WELFARE,
        booking_end=TODAY + timedelta(days=60), balance=2.5,
    )
    await db.flush()
    await db.commit()
    return ids


@pytest.mark.asyncio
async def test_cohort_selects_only_expiring_positive_balances(
    db_session, cohort_data,
):
    result = await svc.resolve_cohort(db_session, _filters())
    picked = {str(c.person_id) for c in result.candidates}
    assert str(cohort_data["soon"]) in picked
    assert str(cohort_data["both"]) in picked
    # 窗口外 / 已过期 / 余额不足 —— 都不该被提醒。
    assert str(cohort_data["later"]) not in picked
    assert str(cohort_data["expired"]) not in picked
    assert str(cohort_data["tiny"]) not in picked
    # 离职的人永远不发。
    assert str(cohort_data["leaver"]) not in picked
    # 没有工作邮箱 → 两个渠道都投不出去,如实计入 skipped 而不是硬塞进名单。
    assert str(cohort_data["no_email"]) not in picked
    assert result.summary.skipped_no_email == 1
    # 只有福利年假的人不在「法定年假」口径里。
    assert str(cohort_data["welfare_only"]) not in picked


@pytest.mark.asyncio
async def test_cohort_honours_welfare_type_when_hr_asks_for_it(
    db_session, cohort_data,
):
    """福利年假和法定年假一样会清零(HR 2026-08-18 确认),两者都能圈。

    口径仍然是 HR 的判断,不是代码替他做的判断 —— 只要他勾了,就要能圈出来。
    """
    result = await svc.resolve_cohort(
        db_session, _filters(account_types=[LEGAL, WELFARE]),
    )
    picked = {str(c.person_id) for c in result.candidates}
    assert str(cohort_data["welfare_only"]) in picked
    both = next(
        c for c in result.candidates if str(c.person_id) == str(cohort_data["both"])
    )
    assert len(both.accounts) == 2
    # 合计 = 两个桶相加;最早清零日 = 更近的那个。
    assert both.total_balance_days == pytest.approx(3.5)
    assert both.earliest_deadline == TODAY + timedelta(days=30)
    # 两个桶都是「不休就没了」—— 法定和福利都是一年半有效期,同日清零。
    welfare = next(a for a in both.accounts if a.account_type == WELFARE)
    assert welfare.forfeits is True
    legal = next(a for a in both.accounts if a.account_type == LEGAL)
    assert legal.forfeits is True


@pytest.mark.asyncio
async def test_expired_included_only_when_asked(db_session, cohort_data):
    result = await svc.resolve_cohort(
        db_session,
        _filters(include_expired=True, expiring_within_days=None),
    )
    picked = {str(c.person_id) for c in result.candidates}
    assert str(cohort_data["expired"]) in picked
    row = next(
        c for c in result.candidates
        if str(c.person_id) == str(cohort_data["expired"])
    )
    assert row.expires_in_days == -10


@pytest.mark.asyncio
async def test_deadline_prefers_booking_end_date(db_session):
    """清零日必须和员工自己卡片上那个日期同源。

    ``hr/router._leave_forfeit_signals`` 用的是 ``booking_end_date or
    end_date``;这里独立验一遍,免得两处哪天各自漂移。
    """
    pid = await _person(db_session, name="Booking")
    _account(
        db_session, pid,
        end=date(2026, 3, 31),
        booking_end=TODAY + timedelta(days=10),
        balance=4,
    )
    await db_session.flush()
    await db_session.commit()
    result = await svc.resolve_cohort(db_session, _filters())
    row = next(c for c in result.candidates if str(c.person_id) == str(pid))
    assert row.accounts[0].deadline == TODAY + timedelta(days=10)
    assert row.expires_in_days == 10


@pytest.mark.asyncio
async def test_closed_accounts_excluded_by_default(db_session):
    pid = await _person(db_session, name="Closed")
    _account(
        db_session, pid, booking_end=TODAY + timedelta(days=10),
        balance=4, closed=True,
    )
    await db_session.flush()
    await db_session.commit()
    assert not (await svc.resolve_cohort(db_session, _filters())).candidates
    included = await svc.resolve_cohort(
        db_session, _filters(include_closed_accounts=True),
    )
    assert len(included.candidates) == 1


@pytest.mark.asyncio
async def test_summary_reports_coverage_honestly(db_session, cohort_data):
    """覆盖面必须说真话。

    SF EC Time-Off 只承载中国主体 —— 「命中 2 人」不等于「全公司提醒到了」,
    汇总里必须同时给出在职总数和有假期数据的人数。
    """
    summary = (await svc.resolve_cohort(db_session, _filters())).summary
    assert summary.recipient_count == len(
        (await svc.resolve_cohort(db_session, _filters())).candidates
    )
    assert summary.active_person_count >= 7  # leaver 不算
    assert summary.persons_with_leave_data >= 6
    assert summary.persons_with_leave_data <= summary.active_person_count
    assert summary.as_of_date == TODAY
    assert summary.newest_synced_at is not None


# ── 2. 文案校验 ─────────────────────────────────────────────────────────


def _valid_copy(**kw) -> TemplateCopy:
    base = {
        "subject": "年假到期", "subject_en": "Expiring", "subject_ja": "失効",
        "body_text": "Hi {person_name} {account_lines}",
        "body_text_en": "Hi {person_name} {account_lines}",
        "body_text_ja": "Hi {person_name} {account_lines}",
    }
    base.update(kw)
    return TemplateCopy(**base)


def test_copy_rejects_unknown_placeholder():
    from app.core.exceptions import ValidationError

    with pytest.raises(ValidationError, match="unknown placeholder"):
        svc.validate_copy(_valid_copy(body_text="Hi {employee_name}"))


def test_copy_rejects_positional_and_format_spec():
    """``_safe_format`` 容忍缺键,但位置参数和格式说明符会**抛异常**。

    一个 ``{days:d}`` 就能让整批投递失败,所以它必须死在写入这一步。
    """
    from app.core.exceptions import ValidationError

    with pytest.raises(ValidationError, match="positional / attribute / index"):
        svc.validate_copy(_valid_copy(body_text="Hi {0}"))
    with pytest.raises(ValidationError, match="format spec or conversion"):
        svc.validate_copy(
            _valid_copy(body_text="{total_balance_days:d} days"),
        )
    # 复审补:这三种以前**完全绕过**白名单 —— 正则只认「纯标识符 + 紧跟右括号」。
    for sneaky in ("Hi {a.b}", "Hi { person_name }", "Hi {person_name.title}"):
        with pytest.raises(ValidationError):
            svc.validate_copy(_valid_copy(body_text=sneaky))


def test_copy_requires_all_three_languages():
    from app.core.exceptions import ValidationError

    with pytest.raises(ValidationError, match="body_text_ja"):
        svc.validate_copy(_valid_copy(body_text_ja=""))
    with pytest.raises(ValidationError, match="subject_en"):
        svc.validate_copy(_valid_copy(subject_en="   "))


def test_copy_accepts_auto_placeholders():
    svc.validate_copy(_valid_copy(
        body_text="{person_name} {cta_line} {portal_url} {hris_url}",
    ))


def test_builtin_copy_passes_its_own_validator():
    """出厂文案必须过自己的闸 —— 否则 HR 什么都没改就发不出去。"""
    svc.validate_copy(svc.builtin_copy())


def test_hris_url_validation():
    from app.core.exceptions import ValidationError

    assert svc.validate_hris_url("  ") == ""
    assert svc.validate_hris_url("https://hris.example.com/leave") == (
        "https://hris.example.com/leave"
    )
    with pytest.raises(ValidationError):
        svc.validate_hris_url("javascript:alert(1)")
    with pytest.raises(ValidationError):
        svc.validate_hris_url("hris.example.com")


# ── 3. 渲染 ─────────────────────────────────────────────────────────────


def _snap(**kw) -> AccountSnapshot:
    base = {
        "account_id": uuid.uuid4(),
        "account_type": LEGAL,
        "account_type_label": "法定年假",
        "fiscal_year": 2025,
        "balance_days": 5.0,
        "entitled_days": 5.0,
        "used_days": 0.0,
        "deadline": date(2026, 9, 30),
        "expires_in_days": 44,
        "forfeits": True,
    }
    base.update(kw)
    return AccountSnapshot(**base)


def test_context_is_localized_per_recipient():
    """``account_lines`` 是按收件人语言拼好的成品。

    一个 envelope 只有一个收件人,所以 context 可以本地化 —— 反过来说,把中文
    明细塞进英文邮件正是 i18n 规则要挡的混排。
    """
    zh = svc.build_context(
        display_name="张三", accounts=[_snap()], lang="zh",
        hris_url="", today=date(2026, 8, 17),
    )
    assert "法定年假 FY2025" in zh["account_lines"]
    assert "剩余 5 天" in zh["account_lines"]
    assert "清零" in zh["account_lines"]
    assert zh["deadline_month"] == "9 月"

    en = svc.build_context(
        display_name="Zhang San", accounts=[_snap()], lang="en",
        hris_url="", today=date(2026, 8, 17),
    )
    assert "Statutory annual leave FY2025" in en["account_lines"]
    assert "5 day(s) left" in en["account_lines"]
    assert en["deadline_month"] == "September"
    # 英文正文里不该出现中文标签。
    assert "法定年假" not in en["account_lines"]

    ja = svc.build_context(
        display_name="张三", accounts=[_snap()], lang="ja",
        hris_url="", today=date(2026, 8, 17),
    )
    # 断言真值表本身,而不是把译名抄一份 —— 译名以前端为准且有 parity 测试守着
    # (test_leave_type_label_parity),抄一份只会在改词时多一处要改的地方。
    from app.domains.hr.leave_service import LOCALIZED_TYPE_LABELS

    assert LOCALIZED_TYPE_LABELS[LEGAL]["ja"] in ja["account_lines"]
    assert ja["deadline_month"] == "9月"


def test_context_uses_earliest_deadline_and_total():
    ctx = svc.build_context(
        display_name="X",
        accounts=[
            _snap(deadline=date(2027, 9, 30), balance_days=2.5),
            _snap(deadline=date(2026, 9, 30), balance_days=1.0),
        ],
        lang="zh", hris_url="", today=date(2026, 8, 17),
    )
    assert ctx["deadline_date"] == "2026-09-30"
    assert ctx["total_balance_days"] == "3.5"
    assert ctx["expires_in_days"] == "44"


def test_fmt_days_trims_trailing_zeros():
    assert svc.fmt_days(5.0) == "5"
    assert svc.fmt_days(0.69) == "0.69"
    assert svc.fmt_days(5.50) == "5.5"
    assert svc.fmt_days(None) == "0"


def test_render_message_flags_unresolved_placeholders():
    """渲染后仍留着大括号 = 员工会看到 ``{foo}``,必须显式报出来。"""
    copy = _valid_copy(body_text="Hi {person_name}, see {account_lines}")
    ctx = svc.build_context(
        display_name="张三", accounts=[_snap()], lang="zh",
        hris_url="", today=date(2026, 8, 17),
    )
    ok = svc.render_message(copy=copy, context=ctx, lang="zh")
    assert "张三" in ok.body_text
    assert ok.unresolved_placeholders == []


def test_render_message_uses_builtin_copy_end_to_end():
    ctx = svc.build_context(
        display_name="张三", accounts=[_snap()], lang="zh",
        hris_url="https://hris.example.com", today=date(2026, 8, 17),
    )
    out = svc.render_message(
        copy=svc.builtin_copy(), context=ctx, lang="zh",
        accounts=[_snap()], hris_url="https://hris.example.com",
        portal_url="https://helix.example.com/me/settings?tab=leave",
    )
    assert "张三" in out.body_text
    assert "2026-09-30" in out.subject
    assert out.unresolved_placeholders == []
    # 卡片:一个账户一格,主数值是剩余天数(``[h]`` 标记在预览里已被剥掉)。
    assert len(out.card_stats) == 1
    assert "5 天" in out.card_stats[0]["value"]
    assert "[h]" not in out.card_stats[0]["value"]
    # 只有一个按钮,落在 HR 配的 HRIS(SF)入口上。
    assert [a["url"] for a in out.card_actions] == ["https://hris.example.com"]


def test_card_actions_without_hris_url_keeps_helix_button():
    """HR 没配 HRIS 入口时降级到 Helix 假期页 —— 一张没有按钮的提醒卡更坏。"""
    actions = svc.build_card_actions("", "zh")
    assert len(actions) == 1
    assert actions[0].url == svc.MY_LEAVE_PATH
    assert actions[0].style == "primary"


# ── 4. 发送 / 快照 / 去重 ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_send_freezes_numbers_and_delivers(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """发送的是 HR 确认时那份数字,不是投递那一刻重新算的。"""
    sent: list[dict] = []

    async def _fake_deliver(self, outbox, env, **kw):
        sent.append({
            "person_id": str(env.recipient.id),
            "context": dict(env.context),
            "template_key": env.template_key,
            "related": (env.related_resource_type, str(env.related_resource_id)),
            "stats": len(env.card_stats),
        })
        outbox.status = "delivered"
        outbox.delivered_channels = list(env.channels)
        outbox.delivered_at = datetime.now(UTC)

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _fake_deliver)

    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session,
        filters=_filters(),
        channels=["feishu", "email"],
        person_ids=[],
        expected_recipient_count=None,
        allow_resend=False,
        test_mode=False,
        actor_person_id=hr.id,
        trace_id="t-1",
    )
    await db_session.commit()
    frozen = {
        r["person_id"]: r["total_balance_days"]
        for r in batch.recipients_snapshot
    }

    # 快照落定之后 HRIS 又同步了一次,余额变了 —— 发出去的必须还是快照里的数。
    acct = (await db_session.execute(
        select(HrLeaveAccount).where(
            HrLeaveAccount.person_id == str(cohort_data["soon"]),
        )
    )).scalars().first()
    acct.balance_days = 99
    await db_session.commit()

    result = await svc.run_batch(db_session, batch.id)
    assert result["status"] == BATCH_STATUS_COMPLETED
    assert result["delivered"] == batch.recipient_count

    soon = next(s for s in sent if s["person_id"] == str(cohort_data["soon"]))
    assert "5 天" in soon["context"]["account_lines"]
    assert "99" not in soon["context"]["account_lines"]
    assert soon["context"]["total_balance_days"] == svc.fmt_days(
        frozen[str(cohort_data["soon"])],
    )
    assert soon["template_key"] == svc.TEMPLATE_KEY
    assert soon["related"] == (BATCH_RESOURCE_TYPE, str(batch.id))
    assert soon["stats"] >= 1

    refreshed = await db_session.get(
        HrLeaveReminderBatch, batch.id, populate_existing=True,
    )
    assert refreshed.status == BATCH_STATUS_COMPLETED
    assert refreshed.failed == 0
    assert refreshed.copy_snapshot["subject"]
    # 每个收件人都有一行 outbox,并挂回批次。
    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.related_resource_type == BATCH_RESOURCE_TYPE,
            NotificationOutbox.related_resource_id == str(batch.id),
        )
    )).scalars().all()
    assert len(rows) == refreshed.recipient_count
    # 正文存哨兵:站内通知按模板重渲染,才能是收件人自己的语言。
    assert all(r.body_text == "(see template)" for r in rows)


@pytest.mark.asyncio
async def test_recently_reminded_excluded_then_allowed(
    db_session, cohort_data, hr_user, monkeypatch,
):
    async def _noop_deliver(self, outbox, env, **kw):
        outbox.status = "delivered"
        outbox.delivered_channels = list(env.channels)

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _noop_deliver)
    hr, _ = hr_user
    hr_id = hr.id  # rollback() 会 expire 实例,见上面同名注释
    first = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr_id, trace_id="t-1",
    )
    await db_session.commit()
    # 同样先取成普通值:下面的 rollback 会 expire 这个批次实例。
    first_id = first.id
    first_count = first.recipient_count
    await svc.run_batch(db_session, first_id)
    assert first_count >= 2

    # 立刻再圈一次:所有人都被标为「近期已提醒」,默认选中数为 0。
    again = await svc.resolve_cohort(db_session, _filters())
    assert again.summary.reminded_recently_count == first_count
    assert again.summary.default_selected_count == 0
    assert all(c.reminded_recently for c in again.candidates)
    assert all(c.last_reminded_batch_id == first_id for c in again.candidates)

    # 默认口径下再发 → 无人可发,报错而不是静默发 0 封。
    from app.core.exceptions import ValidationError as AppValidationError

    with pytest.raises(AppValidationError, match="No recipients"):
        await svc.create_batch(
            db_session, filters=_filters(), channels=["feishu"], person_ids=[],
            expected_recipient_count=None, allow_resend=False, test_mode=False,
            actor_person_id=hr_id, trace_id="t-2",
        )
    await db_session.rollback()

    # HR 显式要求再催一轮 → 允许。护栏,不是硬闸。
    second = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=True, test_mode=False,
        actor_person_id=hr_id, trace_id="t-3",
    )
    assert second.recipient_count == first_count
    await db_session.rollback()


@pytest.mark.asyncio
async def test_resend_window_zero_disables_dedup(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """窗口设为 0 = 关掉去重提示,谁都不算「近期已提醒」。"""
    async def _noop_deliver(self, outbox, env, **kw):
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _noop_deliver)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-1",
    )
    await db_session.commit()
    await svc.run_batch(db_session, batch.id)

    zero = await svc.resolve_cohort(
        db_session, _filters(), resend_window_days=0,
    )
    assert zero.summary.reminded_recently_count == 0
    # 但「上次何时提醒过」这个事实仍然如实呈现。
    assert all(c.last_reminded_at is not None for c in zero.candidates)


@pytest.mark.asyncio
async def test_test_mode_sends_only_to_actor_and_skips_dedup(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """测试发送只发给发起人,且**不**让任何人被判成「已提醒」。

    发起人自己也给了一个命中口径的假期账户 —— 这一点是这条测试的关键:如果
    HR 不在名单里,那么「测试批次是否污染去重」这件事根本没有被观测到(第一版
    正是这样,把去重扫描里的 ``test_mode.is_(False)`` 去掉,测试照样绿)。
    """
    recipients: list[str] = []

    async def _capture(self, outbox, env, **kw):
        recipients.append(str(env.recipient.id))
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _capture)
    hr, _ = hr_user
    hr_id = hr.id

    # 让发起人也落在口径里,这样他既是测试收件人、又是真实候选人。
    _account(db_session, hr_id, booking_end=TODAY + timedelta(days=40), balance=3)
    await db_session.flush()
    await db_session.commit()
    before = await svc.resolve_cohort(db_session, _filters())
    assert str(hr_id) in {str(c.person_id) for c in before.candidates}

    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=True,
        actor_person_id=hr_id, trace_id="t-test",
    )
    await db_session.commit()
    batch_id = batch.id
    assert batch.test_mode is True
    assert batch.recipient_count == 1
    await svc.run_batch(db_session, batch_id)

    # 只发给发起人 —— 名单里其他人一个都没收到。
    assert recipients == [str(hr_id)]
    # 而且发起人自己也没有因此被算成「已提醒」:测试发送不进去重口径。
    after = await svc.resolve_cohort(db_session, _filters())
    assert after.summary.reminded_recently_count == 0
    assert all(c.last_reminded_at is None for c in after.candidates)
    me = next(c for c in after.candidates if str(c.person_id) == str(hr_id))
    assert me.reminded_recently is False
    assert me.last_reminded_batch_id is None


@pytest.mark.asyncio
async def test_explicit_selection_intersects_with_cohort(
    db_session, cohort_data, hr_user,
):
    """显式勾选只能收窄,不能把卡片发给口径外的人。"""
    from app.core.exceptions import ValidationError as AppValidationError

    hr, _ = hr_user
    # ``rollback()`` 下面要用好几次,而 rollback 会 expire 掉 ORM 实例 ——
    # 之后再读 ``hr.id`` 会触发一次同步 lazy refresh(MissingGreenlet)。
    # 先把 id 取成普通值。
    hr_id = hr.id
    only_one = await svc.create_batch(
        db_session, filters=_filters(),
        channels=["feishu"], person_ids=[cohort_data["soon"]],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr_id, trace_id="t-1",
    )
    assert only_one.recipient_count == 1
    assert only_one.recipients_snapshot[0]["person_id"] == str(
        cohort_data["soon"],
    )
    await db_session.rollback()

    # 口径外的人(已离职)被显式指定 → 拒绝整个请求,而不是静默丢掉他。
    with pytest.raises(AppValidationError, match="no longer in the cohort"):
        await svc.create_batch(
            db_session, filters=_filters(),
            channels=["feishu"],
            person_ids=[cohort_data["soon"], cohort_data["leaver"]],
            expected_recipient_count=None, allow_resend=False,
            test_mode=False, actor_person_id=hr_id, trace_id="t-2",
        )
    await db_session.rollback()


@pytest.mark.asyncio
async def test_expected_count_mismatch_is_conflict(
    db_session, cohort_data, hr_user,
):
    """HR 确认时看到 N 人,服务端算出 M 人 → 409,让他重新看一眼名单。"""
    from app.core.exceptions import ConflictError

    hr, _ = hr_user
    hr_id = hr.id
    with pytest.raises(ConflictError, match="cohort changed"):
        await svc.create_batch(
            db_session, filters=_filters(), channels=["feishu"], person_ids=[],
            expected_recipient_count=999, allow_resend=False, test_mode=False,
            actor_person_id=hr_id, trace_id="t-1",
        )
    await db_session.rollback()


@pytest.mark.asyncio
async def test_leaver_between_preview_and_send_is_skipped(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """快照冻结之后那个人离职了 —— 不发,并如实计入 failed/skipped。"""
    delivered: list[str] = []

    async def _capture(self, outbox, env, **kw):
        delivered.append(str(env.recipient.id))
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _capture)
    hr, _ = hr_user

    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-1",
    )
    await db_session.commit()
    assert batch.recipient_count >= 2

    person = await db_session.get(IdentityPerson, str(cohort_data["soon"]))
    person.deleted_at = datetime.now(UTC)
    await db_session.commit()

    result = await svc.run_batch(db_session, batch.id)
    assert str(cohort_data["soon"]) not in delivered
    assert result["skipped"] == 1


@pytest.mark.asyncio
async def test_drain_crash_marks_batch_failed_not_stuck_sending(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """扇出整体炸掉时,批次必须落成 ``failed`` 并带原因,而不是停在 ``sending``。

    停在 ``sending`` 有两个后果:HR 看不到失败原因,而且 in-flight 闸会把下一次
    发送挡住半小时。这条路径要求 except 分支先 rollback —— 否则后面写批次行的
    commit 会以 PendingRollbackError 二次爆掉
    (``reference_swallowed_db_error_poisons_transaction``)。
    """
    from app.notifications import fanout

    async def _boom(*a, **kw):
        raise RuntimeError("feishu integration is disabled")

    monkeypatch.setattr(fanout, "enqueue_then_drain", _boom)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-boom",
    )
    await db_session.commit()
    batch_id = batch.id

    result = await svc.run_batch(db_session, batch_id)
    assert result["status"] == "failed"
    refreshed = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert refreshed.status == "failed"
    assert "feishu integration is disabled" in (refreshed.error_summary or "")
    assert refreshed.finished_at is not None
    # session 仍然可用 —— 没有被吞掉的异常毒化。
    assert (await svc.resolve_cohort(db_session, _filters())).candidates


@pytest.mark.asyncio
async def test_in_flight_guard_blocks_double_send(db_session, hr_user):
    """双击「确认发送」不该发两遍。"""
    from app.core.exceptions import ConflictError

    hr, _ = hr_user
    db_session.add(HrLeaveReminderBatch(
        created_by_person_id=str(hr.id),
        status="sending",
        test_mode=False,
        channels=["feishu"],
        filters={},
        recipients_snapshot=[],
        recipient_count=0,
        created_at=datetime.now(UTC),
    ))
    await db_session.flush()
    with pytest.raises(ConflictError, match="still sending"):
        await svc.assert_no_batch_in_flight(db_session)


@pytest.mark.asyncio
async def test_stale_sending_batch_does_not_block_forever(db_session, hr_user):
    """进程被 kill 掉留下的 ``sending`` 残留不该永久挡住 HR。"""
    hr, _ = hr_user
    db_session.add(HrLeaveReminderBatch(
        created_by_person_id=str(hr.id),
        status="sending",
        test_mode=False,
        channels=["feishu"],
        filters={},
        recipients_snapshot=[],
        recipient_count=0,
        created_at=datetime.now(UTC) - timedelta(hours=3),
        # 心跳判据看的是 updated_at ——``run_batch`` 每块刷一次。停了这么久
        # 就是进程真的没了。
        updated_at=datetime.now(UTC) - timedelta(hours=3),
    ))
    await db_session.flush()
    await svc.assert_no_batch_in_flight(db_session)  # 不抛


# ── 5. RBAC / 端点 ──────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_config_and_preview_require_hr_scope(
    client, cohort_data, hr_headers, employee_headers,
):
    body = {"filters": _filters().model_dump(mode="json")}
    for path, payload in (
        ("/api/v1/hr/leave/expiry-reminder/config", None),
        ("/api/v1/hr/leave/expiry-reminder/preview", body),
    ):
        if payload is None:
            r = await client.get(path, headers=employee_headers)
        else:
            r = await client.post(path, json=payload, headers=employee_headers)
        assert r.status_code == 403, path

    r = await client.get(
        "/api/v1/hr/leave/expiry-reminder/config", headers=hr_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["template_key"] == svc.TEMPLATE_KEY
    assert data["template_copy"]["subject"]
    assert data["copy_overridden"] is False
    assert "person_name" in data["allowed_placeholders"]
    # 码表从真实数据里长出来,会清零的类型排在前面。
    codes = [o["code"] for o in data["account_types"]]
    assert LEGAL in codes
    assert codes[0] == LEGAL or data["account_types"][0]["forfeits"] is True

    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/preview",
        json=body, headers=hr_headers,
    )
    assert r.status_code == 200
    payload = r.json()["data"]
    assert payload["summary"]["recipient_count"] >= 1
    assert payload["candidates"][0]["accounts"]


@pytest.mark.asyncio
async def test_employee_cannot_send(client, cohort_data, employee_headers):
    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/send",
        json={
            "filters": _filters().model_dump(mode="json"),
            "channels": ["feishu"],
        },
        headers=employee_headers,
    )
    assert r.status_code == 403


@pytest.mark.asyncio
async def test_preview_message_renders_real_recipient(
    client, cohort_data, hr_headers,
):
    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/preview-message",
        json={"filters": _filters().model_dump(mode="json")},
        headers=hr_headers,
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["is_sample"] is False
    assert data["subject"]
    assert data["unresolved_placeholders"] == []
    assert data["card_stats"]
    # 邮件口径的 cta_line 渲染进了正文,HR 看到的是「话最多」的版本。
    assert data["context"]["cta_line"]


@pytest.mark.asyncio
async def test_preview_message_rejects_bad_draft_copy(
    client, cohort_data, hr_headers,
):
    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/preview-message",
        json={
            "filters": _filters().model_dump(mode="json"),
            "template_copy": _valid_copy(body_text="Hi {oops}").model_dump(),
        },
        headers=hr_headers,
    )
    assert r.status_code == 422
    assert "unknown placeholder" in r.text


@pytest.mark.asyncio
async def test_config_update_persists_copy_and_can_reset(
    client, db_session, hr_headers,
):
    from app.integrations.models import PlatformSetting

    r = await client.put(
        "/api/v1/hr/leave/expiry-reminder/config",
        json={
            "template_copy": _valid_copy(
                subject="HR 自己写的标题 {deadline_date}",
            ).model_dump(),
            "hris_url": "https://hris.example.com/leave",
            "resend_window_days": 14,
        },
        headers=hr_headers,
    )
    assert r.status_code == 200, r.text
    assert set(r.json()["data"]["changed"]) == {
        "copy", "hris_url", "resend_window_days",
    }

    copy, overridden = await svc.effective_copy(db_session)
    assert overridden is True
    assert copy.subject == "HR 自己写的标题 {deadline_date}"
    # 字段级合并:HR 只改了标题,英日正文仍在。
    assert copy.body_text_ja
    assert await svc.get_hris_url(db_session) == "https://hris.example.com/leave"
    assert await svc.get_resend_window_days(db_session) == 14

    r = await client.put(
        "/api/v1/hr/leave/expiry-reminder/config",
        json={"reset_copy": True}, headers=hr_headers,
    )
    assert r.status_code == 200
    assert await db_session.get(
        PlatformSetting, svc.TEMPLATE_SETTING_KEY,
    ) is None
    copy, overridden = await svc.effective_copy(db_session)
    assert overridden is False
    assert copy.subject == svc.builtin_copy().subject


@pytest.mark.asyncio
async def test_config_update_rejects_bad_copy_over_http(client, hr_headers):
    r = await client.put(
        "/api/v1/hr/leave/expiry-reminder/config",
        json={"template_copy": _valid_copy(body_text="{0}").model_dump()},
        headers=hr_headers,
    )
    assert r.status_code == 422
    assert "positional / attribute / index" in r.text


@pytest.mark.asyncio
async def test_send_and_batch_history_endpoints(
    client, db_session, cohort_data, hr_headers, monkeypatch,
):
    async def _noop_deliver(self, outbox, env, **kw):
        outbox.status = "delivered"
        outbox.delivered_channels = list(env.channels)
        outbox.delivered_at = datetime.now(UTC)

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _noop_deliver)

    # 后台投递在测试里手动 drain,断言才有确定性。``_spawn_background`` 是被
    # 路由**同步**调用的(它自己负责 create_task),所以替身也必须是同步函数 ——
    # 换成 async def 只会产出一个从没被 await 的 coroutine,批次永远停在
    # ``sending``。
    import app.domains.hr.leave_reminder_router as rtr

    spawned: list = []

    def _capture_spawn(coro, name=None):
        spawned.append(coro)

    monkeypatch.setattr(rtr, "_spawn_background", _capture_spawn)

    preview = await client.post(
        "/api/v1/hr/leave/expiry-reminder/preview",
        json={"filters": _filters().model_dump(mode="json")},
        headers=hr_headers,
    )
    expected = preview.json()["data"]["summary"]["default_selected_count"]

    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/send",
        json={
            "filters": _filters().model_dump(mode="json"),
            "channels": ["feishu", "email"],
            "expected_recipient_count": expected,
        },
        headers=hr_headers,
    )
    assert r.status_code == 200, r.text
    batch_id = r.json()["data"]["batch_id"]
    assert r.json()["data"]["recipient_count"] == expected
    # 请求立即返回 ``sending``;投递是后台的事。
    assert r.json()["data"]["status"] == "sending"
    assert len(spawned) == 1
    for coro in spawned:
        await coro

    r = await client.get(
        "/api/v1/hr/leave/expiry-reminder/batches", headers=hr_headers,
    )
    assert r.status_code == 200
    rows = r.json()["data"]
    assert rows[0]["id"] == batch_id
    assert rows[0]["created_by_name"] == "HR User"
    # 内部快照字段不外泄到 HR 的口径展示里。
    assert not any(k.startswith("_") for k in rows[0]["filters"])

    r = await client.get(
        f"/api/v1/hr/leave/expiry-reminder/batches/{batch_id}",
        headers=hr_headers,
    )
    assert r.status_code == 200
    detail = r.json()["data"]
    assert detail["batch"]["status"] == BATCH_STATUS_COMPLETED
    assert detail["copy_snapshot"]["subject"]
    assert len(detail["recipients"]) == expected
    first = detail["recipients"][0]
    assert first["status"] == "delivered"
    # 逐人「当时被告知剩几天」可回溯。
    assert first["balance_days"] is not None
    assert first["deadline"]


@pytest.mark.asyncio
async def test_batch_detail_404_for_unknown_id(client, hr_headers):
    r = await client.get(
        f"/api/v1/hr/leave/expiry-reminder/batches/{uuid.uuid4()}",
        headers=hr_headers,
    )
    assert r.status_code == 404


@pytest.mark.asyncio
async def test_reminder_template_is_actionable_and_mutable():
    """通知分类:action 桶(会驱动红点)+ 可静音,且不是 critical 模板。

    到期提醒的全部价值就是「在某个日期之前被看见」,落进 info 桶就等于把唯一
    需要被注意的通知藏起来了。同时它不该是 critical —— 员工有权关掉假期提醒。
    """
    from app.notifications.categories import BUCKET_ACTION, classify
    from app.notifications.service import is_critical_template

    cat = classify(svc.TEMPLATE_KEY)
    assert cat.bucket == BUCKET_ACTION
    assert cat.mutable is True
    assert cat.employee_facing is True
    assert is_critical_template(svc.TEMPLATE_KEY) is False


def test_template_groups_under_hr_in_the_email_editor():
    from app.notifications.template_catalog import _group_of

    assert _group_of(svc.TEMPLATE_KEY) == "hr"

# ── 6. 安全闸(复审补:这些闸删掉之前全绿) ─────────────────────────────


async def _mint_pat(db_session, user_id) -> str:
    from app.auth.tokens.service import PersonalAccessTokenService

    svc = PersonalAccessTokenService(db_session)
    _, plaintext = await svc.create(
        user_id=user_id, name="leave-reminder-test", expires_in_days=30,
    )
    await db_session.commit()
    return plaintext


@pytest.mark.asyncio
async def test_pat_cannot_send_or_edit_copy(client, db_session, hr_user):
    """群发是 catastrophic-if-stolen 动作,被偷的 PAT 不许触发。

    这也是「MCP 刻意没有发送工具」的全部理由 —— 如果 PAT 能打 /send,那 MCP
    有没有发送工具就无所谓了。此前把三处 ``require_browser_session`` 全删掉,
    整个 suite 依然全绿。
    """
    hr, _ = hr_user
    pat = await _mint_pat(db_session, hr.id)
    headers = {"Authorization": f"Bearer {pat}"}

    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/send",
        json={"filters": _filters().model_dump(mode="json"), "channels": ["feishu"]},
        headers=headers,
    )
    assert r.status_code == 403, r.text

    r = await client.put(
        "/api/v1/hr/leave/expiry-reminder/config",
        json={"hris_url": "https://hris.example.com"},
        headers=headers,
    )
    assert r.status_code == 403

    r = await client.post(
        f"/api/v1/hr/leave/expiry-reminder/batches/{uuid.uuid4()}/resend-failed",
        headers=headers,
    )
    assert r.status_code == 403

    # 同一个 PAT 打只读面是允许的(PAT 继承持有者的角色,这是设计)。
    r = await client.get(
        "/api/v1/hr/leave/expiry-reminder/config", headers=headers,
    )
    assert r.status_code == 200


@pytest.mark.asyncio
async def test_auditor_can_read_history_but_never_send(
    client, cohort_data, auditor_headers,
):
    """auditor 是「能看治理记录但绝不能发」的那个角色。

    此前把写端点的 ``require_hr_admin`` 降成 ``require_hr_read`` 全绿 —— 因为
    没有任何 auditor 打写端点的 403 测试。
    """
    r = await client.get(
        "/api/v1/hr/leave/expiry-reminder/batches", headers=auditor_headers,
    )
    assert r.status_code == 200

    for method, path, payload in (
        ("post", "/api/v1/hr/leave/expiry-reminder/send",
         {"filters": _filters().model_dump(mode="json"), "channels": ["feishu"]}),
        ("put", "/api/v1/hr/leave/expiry-reminder/config",
         {"resend_window_days": 3}),
        # 圈人名单是全量逐人 PII(姓名/邮箱/部门 + 每个账户的余额与清零日),
        # 刻意窄于 hr_read。
        ("post", "/api/v1/hr/leave/expiry-reminder/preview",
         {"filters": _filters().model_dump(mode="json")}),
        ("post", "/api/v1/hr/leave/expiry-reminder/preview-message",
         {"filters": _filters().model_dump(mode="json")}),
    ):
        r = await getattr(client, method)(path, json=payload, headers=auditor_headers)
        assert r.status_code == 403, f"{method} {path} -> {r.status_code}"


@pytest.mark.asyncio
async def test_mcp_candidates_tool_requires_a_governance_role(db_session):
    """MCP 只读工具的门必须和 HTTP 读侧一致,且普通员工进不来。"""
    from app.core.exceptions import ForbiddenError
    from app.core.rbac import AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, HR_ROLE, has_role

    # 门本身(工具体内那一行)的判据:employee 不含任何治理角色。
    assert not has_role(["employee"], GLOBAL_ADMIN_ROLE, HR_ROLE, AUDITOR_ROLE)
    assert has_role(["hr"], GLOBAL_ADMIN_ROLE, HR_ROLE, AUDITOR_ROLE)

    # 工具确实注册了,而且**没有**任何发送工具。取法与 test_mcp_tools_surface
    # 一致(FastMCP 的注册表在 _tool_manager 上)。
    from app.mcp.server import mcp_server

    manager = mcp_server._tool_manager
    names = set(getattr(manager, "_tools", {}) or {})
    assert "list_leave_expiry_reminder_candidates" in names
    assert not [
        n for n in names
        if "expiry_reminder" in n and ("send" in n or "batch" in n)
    ], "a mass employee-facing send must never be exposed over MCP"
    _ = ForbiddenError  # 门抛的类型


# ── 7. 边界与状态机(复审补:这些改成 <= / >= 之前全绿) ─────────────────


@pytest.mark.asyncio
async def test_min_balance_boundary_is_inclusive(db_session):
    """余额恰好等于阈值的人必须入选(``balance < min`` 才排除)。"""
    pid = await _person(db_session, name="Exactly")
    _account(
        db_session, pid, booking_end=TODAY + timedelta(days=10), balance=0.5,
    )
    await db_session.flush()
    await db_session.commit()
    picked = await svc.resolve_cohort(db_session, _filters(min_balance_days=0.5))
    assert [str(c.person_id) for c in picked.candidates] == [str(pid)]
    # 再高一点点就出局。
    out = await svc.resolve_cohort(db_session, _filters(min_balance_days=0.51))
    assert out.candidates == []


@pytest.mark.asyncio
async def test_expiry_window_boundary_is_inclusive(db_session):
    """清零日恰好落在窗口第 N 天的人必须入选。"""
    pid = await _person(db_session, name="Edge")
    _account(db_session, pid, booking_end=TODAY + timedelta(days=30), balance=3)
    await db_session.flush()
    await db_session.commit()
    inside = await svc.resolve_cohort(db_session, _filters(expiring_within_days=30))
    assert [str(c.person_id) for c in inside.candidates] == [str(pid)]
    outside = await svc.resolve_cohort(db_session, _filters(expiring_within_days=29))
    assert outside.candidates == []


@pytest.mark.asyncio
async def test_fiscal_year_filter_narrows(db_session):
    pid = await _person(db_session, name="TwoYears")
    _account(
        db_session, pid, fiscal_year=2025,
        booking_end=TODAY + timedelta(days=20), balance=2,
    )
    _account(
        db_session, pid, fiscal_year=2026,
        booking_end=TODAY + timedelta(days=40), balance=4,
    )
    await db_session.flush()
    await db_session.commit()
    only_2025 = await svc.resolve_cohort(db_session, _filters(fiscal_years=[2025]))
    row = only_2025.candidates[0]
    assert [a.fiscal_year for a in row.accounts] == [2025]
    assert row.total_balance_days == pytest.approx(2)


@pytest.mark.asyncio
async def test_accounts_without_a_deadline_never_ship(db_session, hr_user):
    """没有清零日的桶不能进发送 —— 否则正文会写「· — 清零」。

    窗口条件为空时它们能进**预览**(HR 有理由看见),但 ``create_batch`` 必须
    把它们剔掉。
    """
    from app.core.exceptions import ValidationError as AppValidationError

    pid = await _person(db_session, name="NoDeadline")
    db_session.add(HrLeaveAccount(
        external_code=str(uuid.uuid4()),
        person_id=str(pid),
        account_type=LEGAL,
        fiscal_year=2025,
        start_date=date(2025, 4, 1),
        end_date=None,
        booking_end_date=None,
        balance_days=4,
        entitled_days=4,
        used_days=0,
        is_current=False,
        account_closed=False,
        synced_at=datetime.now(UTC),
    ))
    await db_session.flush()
    await db_session.commit()

    preview = await svc.resolve_cohort(
        db_session, _filters(expiring_within_days=None),
    )
    assert [str(c.person_id) for c in preview.candidates] == [str(pid)]
    assert preview.candidates[0].expires_in_days is None

    hr, _ = hr_user
    with pytest.raises(AppValidationError, match="no deadline"):
        await svc.create_batch(
            db_session, filters=_filters(expiring_within_days=None),
            channels=["feishu"], person_ids=[],
            expected_recipient_count=None, allow_resend=False,
            test_mode=False, actor_person_id=hr.id, trace_id="t-nd",
        )
    await db_session.rollback()


@pytest.mark.asyncio
async def test_expired_accounts_never_ship_even_when_previewed(
    db_session, cohort_data, hr_user,
):
    """勾了「含已过期」也不会把「距清零还有 0 天 / 最晚可用于(过去的日期)」发出去。"""
    from app.core.exceptions import ValidationError as AppValidationError

    hr, _ = hr_user
    hr_id = hr.id  # rollback 会 expire 实例
    f = _filters(include_expired=True, expiring_within_days=None)
    preview = await svc.resolve_cohort(db_session, f)
    assert any(
        str(c.person_id) == str(cohort_data["expired"]) for c in preview.candidates
    )
    # 显式勾选它 → 报「已过清零日」这条**专门**的错,而不是含糊的「名单变了、
    # 请 reload」——后者是假话(什么都没变),照它 reload 会陷入死循环。
    with pytest.raises(AppValidationError, match="past its deadline"):
        await svc.create_batch(
            db_session, filters=f, channels=["feishu"],
            person_ids=[cohort_data["expired"]],
            expected_recipient_count=None, allow_resend=False,
            test_mode=False, actor_person_id=hr_id, trace_id="t-exp",
        )
    await db_session.rollback()

    # 不显式勾选(发全量)时,它也不在发出去的名单里。
    batch = await svc.create_batch(
        db_session, filters=f, channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False,
        test_mode=False, actor_person_id=hr_id, trace_id="t-exp2",
    )
    assert str(cohort_data["expired"]) not in {
        r["person_id"] for r in batch.recipients_snapshot
    }
    assert batch.filters["_excluded_undated"] >= 1
    await db_session.rollback()


@pytest.mark.asyncio
async def test_partial_status_when_some_recipients_skipped(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """有人被 skip、其余成功 → ``partial``(此前把它改成 completed 全绿)。"""
    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"
        outbox.delivered_channels = list(env.channels)

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _ok)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-partial",
    )
    await db_session.commit()
    batch_id = batch.id
    assert batch.recipient_count >= 2

    # 用**离职**(而不是软删)让一个人掉出去 —— Helix 的离职不软删,判据必须走
    # ``is_effectively_departed``,这条以前漏了会给刚离职的人发「快去休假」。
    # 注意不能直接写 ``employment_status='left'``:平台的 LeaverBlocked
    # ``before_flush`` 监听器会拦下任何绕过 flip_to_left() 的直写(它本来就该拦)。
    # ``leaver_pending`` + 已过的离职日是 HRIS 同步真实会产生的中间态。
    person = await db_session.get(IdentityPerson, str(cohort_data["soon"]))
    person.employment_status = "leaver_pending"
    person.termination_date = TODAY - timedelta(days=1)
    person.country = "China"
    await db_session.commit()

    result = await svc.run_batch(db_session, batch_id)
    assert result["skipped"] == 1
    refreshed = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert refreshed.status == "partial"
    assert "left the company" in (refreshed.error_summary or "")


@pytest.mark.asyncio
async def test_all_recipients_departed_reports_the_real_reason(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """全员在预览→发送之间离职时,别把人引去查飞书集成。"""
    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _ok)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-allgone",
    )
    await db_session.commit()
    batch_id = batch.id
    # 走 ``is_effectively_departed`` 的另一条分支:HRIS 已经不再承载这个人
    # (全量对账把行 tombstone 掉了)。同样不碰 employment_status,免得撞上
    # LeaverBlocked 的直写闸。
    for row in batch.recipients_snapshot:
        person = await db_session.get(IdentityPerson, row["person_id"])
        person.hris_record_state = "purged"
        person.hris_purged_at = datetime.now(UTC)
    await db_session.commit()

    await svc.run_batch(db_session, batch_id)
    refreshed = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert refreshed.status == "failed"
    assert "left the company" in (refreshed.error_summary or "")
    assert "Feishu" not in (refreshed.error_summary or "")


@pytest.mark.asyncio
async def test_run_batch_is_idempotent(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """已收尾的批次再跑一次不会重造一整套 outbox 行。"""
    calls = {"n": 0}

    async def _count(self, outbox, env, **kw):
        calls["n"] += 1
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _count)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-idem",
    )
    await db_session.commit()
    batch_id = batch.id
    await svc.run_batch(db_session, batch_id)
    first = calls["n"]
    assert first >= 2

    again = await svc.run_batch(db_session, batch_id)
    assert again.get("already_finished") is True
    assert calls["n"] == first


@pytest.mark.asyncio
async def test_recipient_cap_rejects_oversized_batch(
    db_session, cohort_data, hr_user, monkeypatch,
):
    from app.core.exceptions import ValidationError as AppValidationError

    monkeypatch.setattr(svc, "MAX_RECIPIENTS_PER_BATCH", 1)
    hr, _ = hr_user
    with pytest.raises(AppValidationError, match="exceeds the per-batch cap"):
        await svc.create_batch(
            db_session, filters=_filters(), channels=["feishu"], person_ids=[],
            expected_recipient_count=None, allow_resend=False, test_mode=False,
            actor_person_id=hr.id, trace_id="t-cap",
        )
    await db_session.rollback()


# ── 8. 重发失败项 / 收尸 ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_resend_failed_only_touches_real_failures(
    client, db_session, cohort_data, hr_headers, monkeypatch,
):
    """``pending`` 的行不是失败 —— 重发它们等于给员工发第二遍。

    静默时段被 park 成 pending 的行、以及这批还没 drain 到的行,都长这个样子。
    """
    from app.notifications.models import NotificationOutbox

    sent: list[str] = []

    async def _mixed(self, outbox, env, **kw):
        sent.append(str(env.recipient.id))
        outbox.status = "delivered"
        outbox.delivered_channels = list(env.channels)

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _mixed)
    import app.domains.hr.leave_reminder_router as rtr

    spawned: list = []
    monkeypatch.setattr(rtr, "_spawn_background", lambda coro, name=None: spawned.append(coro))

    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/send",
        json={"filters": _filters().model_dump(mode="json"), "channels": ["feishu"]},
        headers=hr_headers,
    )
    assert r.status_code == 200
    batch_id = r.json()["data"]["batch_id"]
    for coro in spawned:
        await coro

    rows = (await db_session.execute(
        select(NotificationOutbox).where(
            NotificationOutbox.related_resource_type == BATCH_RESOURCE_TYPE,
            NotificationOutbox.related_resource_id == batch_id,
        )
    )).scalars().all()
    assert len(rows) >= 2
    # 一行摆成「静默时段被 park」的样子(pending),一行摆成真失败。
    rows[0].status = "pending"
    rows[0].last_error = "deferred — recipient quiet hours"
    rows[1].status = "failed"
    rows[1].last_error = "feishu 99991400"
    await db_session.commit()

    resent_ids: list[str] = []

    async def _capture_resend(self, ids, **kw):
        resent_ids.extend(str(i) for i in ids)
        return {"resent": len(ids), "skipped": 0, "new_ids": [], "skipped_detail": []}

    monkeypatch.setattr(NotificationService, "resend_outbox_rows", _capture_resend)
    r = await client.post(
        f"/api/v1/hr/leave/expiry-reminder/batches/{batch_id}/resend-failed",
        headers=hr_headers,
    )
    assert r.status_code == 200, r.text
    assert resent_ids == [str(rows[1].id)]
    assert str(rows[0].id) not in resent_ids


@pytest.mark.asyncio
async def test_resend_failed_refuses_while_batch_is_still_sending(
    client, db_session, hr_user, hr_headers,
):
    hr, _ = hr_user
    batch = HrLeaveReminderBatch(
        created_by_person_id=str(hr.id),
        status="sending",
        test_mode=False,
        channels=["feishu"],
        filters={},
        recipients_snapshot=[],
        recipient_count=3,
        created_at=datetime.now(UTC),
    )
    db_session.add(batch)
    await db_session.flush()
    await db_session.commit()
    r = await client.post(
        f"/api/v1/hr/leave/expiry-reminder/batches/{batch.id}/resend-failed",
        headers=hr_headers,
    )
    assert r.status_code == 409
    assert "still sending" in r.text


@pytest.mark.asyncio
async def test_reaper_closes_a_batch_whose_task_never_started(
    db_session, hr_user,
):
    """最坏的一种:批次行已提交,进程在 spawn 投递之前就没了 —— 一个人都没收到。

    没有 reaper 的话 HR 看到的是「正在发送」,永远,且没有任何信号。
    """
    hr, _ = hr_user
    batch = HrLeaveReminderBatch(
        created_by_person_id=str(hr.id),
        status="sending",
        test_mode=False,
        channels=["feishu"],
        filters={},
        recipients_snapshot=[{"person_id": str(hr.id)}],
        recipient_count=1,
        created_at=datetime.now(UTC) - timedelta(hours=2),
        updated_at=datetime.now(UTC) - timedelta(hours=2),
        started_at=datetime.now(UTC) - timedelta(hours=2),
    )
    db_session.add(batch)
    await db_session.flush()
    await db_session.commit()
    batch_id = batch.id

    out = await svc.reap_stale_batches(db_session)
    assert out["reaped"] == 1
    refreshed = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert refreshed.status == "failed"
    assert "never started" in (refreshed.error_summary or "")
    assert "nobody was notified" in (refreshed.error_summary or "")
    assert refreshed.finished_at is not None

    # 幂等:再跑一次不动已收尾的行。
    assert (await svc.reap_stale_batches(db_session))["reaped"] == 0


@pytest.mark.asyncio
async def test_reaper_leaves_a_live_batch_alone(db_session, hr_user):
    """心跳还新的批次是「正在正常发」,不许被收尸。"""
    hr, _ = hr_user
    batch = HrLeaveReminderBatch(
        created_by_person_id=str(hr.id),
        status="sending",
        test_mode=False,
        channels=["feishu"],
        filters={},
        recipients_snapshot=[],
        recipient_count=500,
        created_at=datetime.now(UTC) - timedelta(hours=3),
        updated_at=datetime.now(UTC),  # 刚刷过心跳
    )
    db_session.add(batch)
    await db_session.flush()
    assert (await svc.reap_stale_batches(db_session))["reaped"] == 0
    # 同一条也不该挡住 in-flight 闸之外的判断:它确实在跑,所以要挡。
    from app.core.exceptions import ConflictError

    with pytest.raises(ConflictError):
        await svc.assert_no_batch_in_flight(db_session)


@pytest.mark.asyncio
async def test_test_mode_batch_never_blocks_a_real_send(db_session, hr_user):
    """一次崩掉的测试发送不该把真实群发挡在门外。"""
    hr, _ = hr_user
    db_session.add(HrLeaveReminderBatch(
        created_by_person_id=str(hr.id),
        status="sending",
        test_mode=True,
        channels=["feishu"],
        filters={},
        recipients_snapshot=[],
        recipient_count=1,
        created_at=datetime.now(UTC),
        updated_at=datetime.now(UTC),
    ))
    await db_session.flush()
    await svc.assert_no_batch_in_flight(db_session)  # 不抛


# ── 9. 去重只认真的送达 ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_dedup_ignores_a_batch_that_delivered_nothing(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """一次全军覆没的批次不该让整个 cohort 在去重窗口内被默认排除。

    否则「集成挂了一次」= 接下来 7 天 HR 什么都发不出去,而员工从没收到过。
    """
    async def _fail(self, outbox, env, **kw):
        raise RuntimeError("feishu down")

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _fail)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-dead",
    )
    await db_session.commit()
    await svc.run_batch(db_session, batch.id)

    again = await svc.resolve_cohort(db_session, _filters())
    assert again.summary.reminded_recently_count == 0
    assert again.summary.default_selected_count == again.summary.recipient_count


@pytest.mark.asyncio
async def test_no_email_count_only_covers_people_who_would_otherwise_qualify(
    db_session,
):
    """「因为没邮箱投不出去」这个数字不该把本来也不入选的人算进来。"""
    tiny = await _person(db_session, name="NoMailTiny", email=False)
    _account(db_session, tiny, booking_end=TODAY + timedelta(days=10), balance=0.1)
    await db_session.flush()
    await db_session.commit()
    summary = (await svc.resolve_cohort(db_session, _filters())).summary
    assert summary.skipped_no_email == 0


# ── 10. 审计 ───────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_send_and_config_write_audit_rows(
    client, db_session, cohort_data, hr_headers, monkeypatch,
):
    """规则 6(每个关键动作可追溯)在这两条写路径上被钉住。"""
    from app.audit.models import AuditLog

    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _ok)
    import app.domains.hr.leave_reminder_router as rtr

    monkeypatch.setattr(rtr, "_spawn_background", lambda coro, name=None: coro.close())

    await client.put(
        "/api/v1/hr/leave/expiry-reminder/config",
        json={"resend_window_days": 9}, headers=hr_headers,
    )
    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/send",
        json={"filters": _filters().model_dump(mode="json"), "channels": ["feishu"]},
        headers=hr_headers,
    )
    assert r.status_code == 200
    batch_id = r.json()["data"]["batch_id"]

    actions = {
        a for (a,) in (await db_session.execute(
            select(AuditLog.action).where(
                AuditLog.action.like("hr.leave_expiry_reminder%"),
            )
        )).all()
    }
    assert "hr.leave_expiry_reminder.send" in actions
    assert "hr.leave_expiry_reminder.config_update" in actions

    row = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "hr.leave_expiry_reminder.send",
            AuditLog.resource_id == batch_id,
        )
    )).scalars().first()
    assert row is not None
    assert row.detail["recipient_count"] >= 1
    assert row.detail["channels"] == ["feishu"]


# ── 11. 文案与投递路径不打架 ───────────────────────────────────────────


@pytest.mark.asyncio
async def test_copy_cannot_be_edited_while_a_batch_is_sending(
    client, db_session, hr_user, hr_headers,
):
    """drain 期间改文案 = 同一批人前后收到两种说法。"""
    hr, _ = hr_user
    db_session.add(HrLeaveReminderBatch(
        created_by_person_id=str(hr.id),
        status="sending",
        test_mode=False,
        channels=["feishu"],
        filters={},
        recipients_snapshot=[],
        recipient_count=5,
        created_at=datetime.now(UTC),
        updated_at=datetime.now(UTC),
    ))
    await db_session.flush()
    await db_session.commit()

    r = await client.put(
        "/api/v1/hr/leave/expiry-reminder/config",
        json={"template_copy": _valid_copy().model_dump()},
        headers=hr_headers,
    )
    assert r.status_code == 409
    assert "two different" in r.text

    # 不碰文案的设置项仍可改 —— 闸只保护「同一批人两种说法」这一件事。
    r = await client.put(
        "/api/v1/hr/leave/expiry-reminder/config",
        json={"resend_window_days": 5}, headers=hr_headers,
    )
    assert r.status_code == 200


def test_unresolved_detection_covers_card_fields_too():
    """卡片副标题 / 按钮文字同样会被渲染,漏检就等于漏报。"""
    copy = _valid_copy(card_subtitle="来自 {hris_url} 的提醒")
    ctx = svc.build_context(
        display_name="张三", accounts=[_snap()], lang="zh",
        hris_url="", today=date(2026, 8, 17),
    )
    out = svc.render_message(copy=copy, context=ctx, lang="zh")
    # hris_url 是合法占位符且 context 里有(空串),所以不该报 unresolved。
    assert out.unresolved_placeholders == []
    assert out.card_subtitle == "来自  的提醒"


def test_literal_braces_are_accepted_and_rendered():
    """``{{`` 转义必须能写 —— 旧的窄正则会把它误判成未知占位符而拒绝保存。"""
    copy = _valid_copy(body_text="Hi {person_name}, use {{curly}} freely")
    svc.validate_copy(copy)
    ctx = svc.build_context(
        display_name="张三", accounts=[_snap()], lang="zh",
        hris_url="", today=date(2026, 8, 17),
    )
    out = svc.render_message(copy=copy, context=ctx, lang="zh")
    assert "use {curly} freely" in out.body_text
    # ``{curly}`` 不在白名单里,所以它是字面量而不是「没替换掉的占位符」。
    assert out.unresolved_placeholders == []


# ── 12. 第二轮复审的修复(每一条都对应一个会伤到员工的失败场景) ──────────


@pytest.mark.asyncio
async def test_drain_enqueues_the_whole_batch_before_delivering_any(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """phase 1 必须是**全批**入队 —— 这是「进程被 kill 后剩下的行仍由 outbox
    worker 兜住」的全部来源。

    此前为了刷心跳把 fanout 切成每 25 人一块,于是 phase 1 被挪进循环:第 5 块中途
    被部署重启,第 126..500 号收件人**根本没有 outbox 行**,重试 worker 无从兜住,
    而 resend-failed 只能挑已存在的行 ⇒ 那些人无法通过任何 UI 补发。
    """
    from app.notifications.models import NotificationOutbox

    rows_at_first_delivery: list[int] = []

    async def _peek(self, outbox, env, **kw):
        if not rows_at_first_delivery:
            # 第一次投递时,这一批的 outbox 行应该**已经全部**在库里了。
            n = len((await db_session.execute(
                select(NotificationOutbox.id).where(
                    NotificationOutbox.related_resource_type == BATCH_RESOURCE_TYPE,
                    NotificationOutbox.related_resource_id == str(env.related_resource_id),
                )
            )).all())
            rows_at_first_delivery.append(n)
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _peek)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-phase1",
    )
    await db_session.commit()
    expected = batch.recipient_count
    assert expected >= 2
    await svc.run_batch(db_session, batch.id)
    assert rows_at_first_delivery == [expected], (
        "phase 1 must durably enqueue every recipient before phase 2 delivers "
        f"the first one (saw {rows_at_first_delivery} of {expected})"
    )


@pytest.mark.asyncio
async def test_heartbeat_advances_updated_at_during_the_drain(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """心跳必须在投递过程中真的推进 —— in-flight 闸与 reaper 都靠它区分
    「还在正常发」和「进程没了」。"""
    beats: list[int] = []

    async def _count(self, outbox, env, **kw):
        beats.append(1)
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _count)
    monkeypatch.setattr(svc, "_HEARTBEAT_EVERY", 1)  # 每行都刷,断言才确定
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-beat",
    )
    await db_session.commit()
    before = (await db_session.get(
        HrLeaveReminderBatch, batch.id, populate_existing=True,
    )).updated_at
    await svc.run_batch(db_session, batch.id)
    after = (await db_session.get(
        HrLeaveReminderBatch, batch.id, populate_existing=True,
    )).updated_at
    assert len(beats) >= 2
    assert before is not None and after is not None
    assert after >= before


@pytest.mark.asyncio
async def test_counters_add_up_when_a_row_never_gets_enqueued(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """``delivered + failed`` 必须等于 ``recipient_count``,状态不能判「已完成」。

    ``enqueue_then_drain`` 对单行 enqueue 失败是 rollback+log+continue(**不抛**),
    所以此前 ``run_batch`` 会算出 delivered=N-1 / failed=0 → 状态 completed ——
    HR 看到「收件人 3 · 已送达 2 · 失败 0 · 已完成」,一个人凭空消失且页面上没有
    任何线索。
    """
    from app.notifications import fanout
    from app.notifications.service import NotificationService

    real_enqueue = NotificationService.enqueue
    state = {"n": 0}

    async def _flaky_enqueue(self, env):
        state["n"] += 1
        if state["n"] == 1:  # 第一行入队失败
            raise RuntimeError("simulated enqueue failure")
        return await real_enqueue(self, env)

    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"

    monkeypatch.setattr(NotificationService, "enqueue", _flaky_enqueue)
    monkeypatch.setattr(NotificationService, "deliver", _ok)
    assert fanout.enqueue_then_drain  # 用的是真 fanout,不是替身

    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-enq",
    )
    await db_session.commit()
    batch_id, total = batch.id, batch.recipient_count
    assert total >= 2
    await svc.run_batch(db_session, batch_id)
    fresh = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert fresh.delivered + fresh.failed == total, (
        f"delivered={fresh.delivered} failed={fresh.failed} total={total} "
        f"enqueued={fresh.enqueued} status={fresh.status}"
    )
    assert fresh.failed >= 1
    # 其余人照常送达 —— 这一条把测试钉在「一个人凭空消失」而不是「整批都挂了」。
    assert fresh.delivered >= 1, (
        f"the other recipients must still go out (delivered={fresh.delivered})"
    )
    # 关键:不能报「已完成」。
    assert fresh.status == "partial", fresh.status
    assert fresh.enqueued < total, "enqueued 必须如实小于收件人数"


@pytest.mark.asyncio
async def test_explicit_selection_still_honours_the_dedup_guard(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """显式勾选**也**要过去重护栏。

    此前是 ``elif``,而前端在正常路径上总是带 ``person_ids`` ⇒ 护栏在真实路径上
    从不生效:HR 发完一轮、页面把刚发过的人重新勾满,再点一次就是同一批人第二
    张卡,零警告。
    """
    from app.core.exceptions import ConflictError

    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"
        outbox.delivered_channels = list(env.channels)

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _ok)
    hr, _ = hr_user
    hr_id = hr.id
    first = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr_id, trace_id="t-dedup1",
    )
    await db_session.commit()
    sent_ids = [uuid.UUID(r["person_id"]) for r in first.recipients_snapshot]
    await svc.run_batch(db_session, first.id)

    # 带着**同一批** person_ids 再发一次 → 必须被挡住并点名。
    with pytest.raises(ConflictError, match="already.*reminded recently"):
        await svc.create_batch(
            db_session, filters=_filters(), channels=["feishu"],
            person_ids=sent_ids, expected_recipient_count=None,
            allow_resend=False, test_mode=False,
            actor_person_id=hr_id, trace_id="t-dedup2",
        )
    await db_session.rollback()

    # HR 显式说「就是要再催一轮」→ 放行。
    again = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"],
        person_ids=sent_ids, expected_recipient_count=None,
        allow_resend=True, test_mode=False,
        actor_person_id=hr_id, trace_id="t-dedup3",
    )
    assert again.recipient_count == len(sent_ids)
    await db_session.rollback()


def test_non_forfeiting_and_closed_buckets_never_say_forfeited():
    """机器生成的那两串必须尊重 ``forfeits`` / ``account_closed``。

    HR 可以刻意提醒不清零的类型(用户要求三种假期都能选),但 Helix 自己拼的
    「· 2027-09-30 清零」对一个不清零的桶是假话 —— 而 Helix 手里正好有这个字段。
    """
    for lang, expiring_word, safe_word in (
        ("zh", "清零", "最晚可用于"),
        ("en", "forfeited", "last bookable"),
        ("ja", "失効", "まで取得可"),
    ):
        forfeiting = svc.build_context(
            display_name="X", accounts=[_snap(forfeits=True)], lang=lang,
            hris_url="", today=date(2026, 8, 17),
        )["account_lines"]
        assert expiring_word in forfeiting

        welfare = svc.build_context(
            display_name="X",
            accounts=[_snap(account_type=WELFARE, forfeits=False)],
            lang=lang, hris_url="", today=date(2026, 8, 17),
        )["account_lines"]
        assert safe_word in welfare
        assert expiring_word not in welfare, (
            f"[{lang}] a non-forfeiting bucket must not be described as forfeited"
        )

        closed = svc.build_context(
            display_name="X",
            accounts=[_snap(forfeits=True, account_closed=True)],
            lang=lang, hris_url="", today=date(2026, 8, 17),
        )["account_lines"]
        assert safe_word in closed
        assert expiring_word not in closed, (
            f"[{lang}] a closed bucket must not be described as forfeited — the "
            "employee's own leave card shows it as history with no countdown"
        )


def test_broken_entitled_used_breakdown_is_omitted_not_shipped():
    """``entitled - used != balance`` 时不写这两个派生数。

    ``balance_days`` 是 SF 的直接真值,另两个是 Helix 从 posting 派生的;一条
    非消耗类型的负向 posting(期末冲销/结转出/手工负调)就能让三个数互相矛盾。
    与其把矛盾发给员工,不如少说两个数。
    """
    ok = svc.build_context(
        display_name="X",
        accounts=[_snap(balance_days=5, entitled_days=8, used_days=3)],
        lang="zh", hris_url="", today=date(2026, 8, 17),
    )["account_lines"]
    assert "累计授予 8" in ok and "已用 3" in ok

    broken = svc.build_context(
        display_name="X",
        accounts=[_snap(balance_days=5, entitled_days=10, used_days=3)],
        lang="zh", hris_url="", today=date(2026, 8, 17),
    )["account_lines"]
    assert "剩余 5 天" in broken
    assert "累计授予" not in broken and "已用" not in broken


def test_context_separates_total_from_the_first_deadline_and_states_as_of():
    """多桶时「合计」与「最早那个桶」必须是两个数,并且信里要有口径日期。"""
    ctx = svc.build_context(
        display_name="X",
        accounts=[
            _snap(deadline=date(2026, 9, 30), balance_days=1.0),
            _snap(deadline=date(2027, 9, 30), balance_days=2.5),
        ],
        lang="zh", hris_url="", today=date(2026, 8, 17),
        as_of=date(2026, 8, 17),
    )
    assert ctx["total_balance_days"] == "3.5"
    assert ctx["expiring_balance_days"] == "1"   # 只有最早那个桶
    assert ctx["deadline_date"] == "2026-09-30"
    assert ctx["as_of_date"] == "2026-08-17"
    # 两个余额占位符仍然是 HR 可用的词汇,即使出厂正文不再用它们 —— HR
    # 2026-08-18 把余额明细整个从正文里拿掉了,数字只留在卡片格子里。
    assert "expiring_balance_days" in svc.ALLOWED_PLACEHOLDERS
    assert "total_balance_days" in svc.ALLOWED_PLACEHOLDERS
    body = svc.builtin_copy().body_text
    assert "{expiring_balance_days}" not in body
    # 口径日期仍然要写在信里,否则员工会拿快照数字和实时卡片的差额来问 HR。
    assert "{as_of_date}" in body


def test_expires_in_days_is_not_clamped_to_zero():
    """负数要显形,不要被 ``max(0, …)`` 抹成「还有 0 天」。

    clamp 会把两条真实路径伪装成正常:① 23:59 建批、00:00 之后 drain;
    ② HR 勾了「含已过期」后在文案 tab 看到的那份预览(排序把负得最多的那个人
    排在第一位,而 ``preview-message`` 默认取第一个人)。
    """
    ctx = svc.build_context(
        display_name="X", accounts=[_snap(deadline=date(2026, 8, 10))],
        lang="zh", hris_url="", today=date(2026, 8, 17),
    )
    assert ctx["expires_in_days"] == "-7"


def test_hris_url_normalises_scheme_and_rejects_userinfo():
    from app.core.exceptions import ValidationError

    # 消费侧的 ``startswith`` 是大小写敏感的,所以协议要归一化,否则
    # ``HTTPS://…`` 会被当成相对路径拼成 ``https://helix…/HTTPS://…``。
    assert svc.validate_hris_url("HTTPS://hris.example.com/x") == (
        "https://hris.example.com/x"
    )
    with pytest.raises(ValidationError, match="userinfo"):
        svc.validate_hris_url("https://helix.pingcap.net@evil.com/")


def test_copy_fields_have_length_caps():
    """2 MB 的正文不该能过校验 —— 它会被复制进每个批次快照并原样投递。"""
    import pydantic

    with pytest.raises(pydantic.ValidationError):
        TemplateCopy(subject="x" * 5000)
    with pytest.raises(pydantic.ValidationError):
        TemplateCopy(body_text="x" * 100_000)


def test_hr_button_label_reaches_the_real_card_button():
    """HR 改 ``card_button_label`` 必须真的改到按钮上。

    ``_build_interactive_card`` 是 ``if extra_actions: … elif url:``,而我们无条件
    返回至少一个 action ⇒ 模板的主按钮分支永不执行。不把这个字段接进
    ``build_card_actions``,HR 改完保存就是看不到任何变化。
    """
    from app.notifications.service import resolve_card_text

    copy = _valid_copy(
        card_button_label="去 HRIS 提交",
        card_button_label_en="File it in HRIS",
        card_button_label_ja="HRIS で申請",
    )
    actions = svc.build_card_actions("https://hris.example.com", "zh", copy)
    assert resolve_card_text(actions[0].label, "zh") == "去 HRIS 提交"
    assert resolve_card_text(actions[0].label, "en") == "File it in HRIS"
    assert resolve_card_text(actions[0].label, "ja") == "HRIS で申請"
    # 没配 HRIS 入口时只有「查看我的假期」,标签不受这个字段影响。
    only_helix = svc.build_card_actions("", "zh", copy)
    assert len(only_helix) == 1
    assert only_helix[0].url == svc.MY_LEAVE_PATH


@pytest.mark.asyncio
async def test_batch_detail_and_mcp_tool_are_hr_admin_only(
    client, db_session, cohort_data, hr_headers, auditor_headers, monkeypatch,
):
    """逐人 PII 的两个面都必须窄于 ``require_hr_read``。

    `/preview` 早就收窄了,理由是全量逐人 PII;而批次明细回的是同一类数据
    (姓名/邮箱/余额/清零日,一批最多 1000 人),MCP 工具回的更是 `/preview` 的
    payload 本身 —— 前门关了后门开着没有意义。
    """
    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _ok)
    import app.domains.hr.leave_reminder_router as rtr

    monkeypatch.setattr(rtr, "_spawn_background", lambda coro, name=None: coro.close())

    r = await client.post(
        "/api/v1/hr/leave/expiry-reminder/send",
        json={"filters": _filters().model_dump(mode="json"), "channels": ["feishu"]},
        headers=hr_headers,
    )
    assert r.status_code == 200
    batch_id = r.json()["data"]["batch_id"]

    # 聚合列表:治理可见性,auditor 可读。
    assert (await client.get(
        "/api/v1/hr/leave/expiry-reminder/batches", headers=auditor_headers,
    )).status_code == 200
    # 逐人明细:必须 403。
    assert (await client.get(
        f"/api/v1/hr/leave/expiry-reminder/batches/{batch_id}",
        headers=auditor_headers,
    )).status_code == 403
    # hr 自己仍然能看。
    assert (await client.get(
        f"/api/v1/hr/leave/expiry-reminder/batches/{batch_id}", headers=hr_headers,
    )).status_code == 200

    # MCP 工具的门必须与 /preview 一致(global_admin + hr),不含 auditor/management。
    from app.core.rbac import AUDITOR_ROLE, GLOBAL_ADMIN_ROLE, HR_ROLE, has_role

    assert not has_role([AUDITOR_ROLE], GLOBAL_ADMIN_ROLE, HR_ROLE)
    src = pathlib.Path(
        rtr.__file__,
    ).parent.parent.parent / "mcp" / "tools.py"
    tool_src = src.read_text()
    idx = tool_src.index("list_leave_expiry_reminder_candidates")
    window = tool_src[idx:idx + 4000]
    assert "_has_role(rt.user.role_names, _ADMIN, _HR)" in window, (
        "the MCP candidates tool must mirror POST /preview's gate — it returns "
        "the same per-person PII"
    )


@pytest.mark.asyncio
async def test_heartbeat_leaves_the_batch_row_readable(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """心跳必须真的把 DB 时钟写进 ``updated_at``,且批次行之后仍然可读。

    这两点是 in-flight 闸与 reaper 的前提:它们都拿 ``updated_at`` 和
    ``now() - _STALE_HEARTBEAT`` 比较。

    **诚实说明这条测试覆盖了什么、没覆盖什么**:第一版心跳是「``db.get`` 出实例 →
    ``beat.updated_at = func.now()`` → commit」。把 SQL 表达式赋给 ORM 属性之后,
    SQLAlchemy 不知道服务端算出的值,会把该属性标成 expired ⇒ 之后在**异步**代码里
    读它就是 ``MissingGreenlet``(用独立脚本直接实测确认过)。但那是一个**潜在**
    陷阱而不是当时已经在爆的 bug —— ``run_batch`` 恰好没有在 commit 之后再读那个
    实例,所以把实现换回旧写法,这条测试**不会**转红(我试过)。改成 Core UPDATE 的
    理由因此是「消掉一个只要有人多写一行读取就会炸的雷,并省掉每次心跳一条
    SELECT」,不是「修一个已在生产爆掉的缺陷」。
    """
    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _ok)
    monkeypatch.setattr(svc, "_HEARTBEAT_EVERY", 1)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-beat2",
    )
    await db_session.commit()
    batch_id = batch.id
    before = (await db_session.execute(
        select(HrLeaveReminderBatch.updated_at).where(
            HrLeaveReminderBatch.id == batch_id,
        )
    )).scalar_one()

    await svc.run_batch(db_session, batch_id)

    # 读属性不炸(这就是那个 MissingGreenlet 会出现的地方),且时间前进了。
    fresh = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert isinstance(fresh.updated_at, datetime)
    assert svc._as_utc(fresh.updated_at) >= svc._as_utc(before)
    # in-flight 闸能拿这个值做比较而不抛。
    assert fresh.finished_at is not None


# ── 13. 第三轮复审的修复 ─────────────────────────────────────────────


def test_card_stat_is_balance_only_and_reads_at_a_glance():
    """额度格子 = SF 的叫法 + 一个放大的余额。别的什么都不写。

    HR 定的形状(2026-08-18):清零日/已用/累计授予挤在两列网格里只会把「你还
    剩几天」这一个数淹掉。此前那条 hint(「2026-09-30 清零 · 已用 0」)因此整个
    去掉了 —— 这个测试钉住「去掉」本身,免得哪天有人顺手把细节加回来。

    标题的措辞见 ``test_card_stat_label_is_hr_wording_in_three_languages``
    (HR 第二轮把它从「法定年假 2025」改成了「截至目前未休法定年假(FY26)」)。
    """
    from app.notifications.service import resolve_card_text

    stat = svc.build_card_stats(
        [_snap(fiscal_year=2025, period_end=date(2026, 3, 31))], "zh",
    )[0]
    assert resolve_card_text(stat.label, "zh") == "截至目前未休法定年假(FY26)"
    assert stat.hint is None
    # 主数值走 ``[h]``(飞书 ``heading_3``),纯文本渠道降级成加粗 —— 不管走哪条
    # 渠道,标记本身都不该漏给员工。
    assert resolve_card_text(stat.value, "zh") == "[h]5[/h] 天"
    for channel, expected in (
        ("feishu_card", "<font size='heading_3'>**5**</font> 天"),
        ("email_text", "**5** 天"),
    ):
        from app.notifications.service import _apply_inline_markup

        rendered = _apply_inline_markup("[h]5[/h] 天", channel=channel)
        assert rendered == expected, channel
        assert "[h]" not in rendered

    # 整格的最终 markdown。**上一版这里是错的**:格子本身还会再包一层 ``**``,
    # 于是变成 ``**<font …>**5**</font> 天**`` —— 嵌套的 ``**`` 让飞书把加粗和
    # 放大一起丢掉(2026-08-18 真机截图实证)。判据见
    # ``test_stat_value_is_not_double_bolded_into_oblivion``。
    from app.notifications.service import _render_stat_grid

    cell = _render_stat_grid([stat])["columns"][0]["elements"][0]["content"]
    assert cell == (
        "<font color='grey'>截至目前未休法定年假(FY26)</font>\n"
        "<font size='heading_3'>**5**</font> 天"
    )


def test_account_lines_still_respect_forfeits_after_the_hint_went_away():
    """卡片没有 hint 了,但正文明细仍然是 HR 可选的 —— 判据不能跟着消失。

    ``forfeits`` / ``account_closed`` 的读取此前有两个消费者(明细 + hint)。
    删掉一个不该把另一个也放松掉:一句「XX 2027-09-30 清零」对一个不清零的桶
    仍然是 Helix 自己拼出来的假话。
    """
    def _line(acct, lang="zh"):
        return svc.build_context(
            display_name="X", accounts=[acct], lang=lang,
            hris_url="", today=date(2026, 8, 17),
        )["account_lines"]

    assert "清零" in _line(_snap(forfeits=True))
    assert "清零" not in _line(_snap(account_type=WELFARE, forfeits=False))
    assert "最晚可用于" in _line(_snap(account_type=WELFARE, forfeits=False))
    assert "清零" not in _line(_snap(forfeits=True, account_closed=True))


@pytest.mark.asyncio
async def test_tally_prefers_delivered_when_rows_share_a_second(
    db_session, hr_user,
):
    """同一秒里「失败源行 + 成功克隆行」必须判成已送达。

    `notification_outbox.created_at` 精度只到秒,而重发是批量克隆 —— 源行与克隆行
    经常同秒。此前平票只靠 uuid4 字符串兜底 = 抛硬币:实测 40 组同秒配对有 22 组
    被判成未送达,于是重发全成功之后台账还写着「失败 22」。
    """
    from app.notifications.models import NotificationOutbox

    hr, _ = hr_user
    batch = HrLeaveReminderBatch(
        created_by_person_id=str(hr.id), status="partial", test_mode=False,
        channels=["feishu"], filters={}, recipients_snapshot=[],
        recipient_count=1, finished_at=datetime.now(UTC),
    )
    db_session.add(batch)
    await db_session.flush()

    same_second = datetime.now(UTC).replace(microsecond=0)
    for status in ("failed", "delivered"):
        db_session.add(NotificationOutbox(
            recipient_person_id=hr.id,
            template_key=svc.TEMPLATE_KEY,
            title="t", body_text="(see template)",
            channels=["feishu"], status=status,
            related_resource_type=BATCH_RESOURCE_TYPE,
            related_resource_id=batch.id,
            created_at=same_second,
        ))
    await db_session.flush()

    tally = await svc.batch_delivery_tally(db_session, batch.id)
    assert tally.persons == 1
    assert tally.delivered == 1, (
        "一个人只要有一行真的推出去了,他的处境就是「收到了」——平票必须偏向已送达"
    )
    assert tally.undelivered == 0
    await db_session.rollback()


@pytest.mark.asyncio
async def test_resend_skips_people_whose_latest_row_already_delivered(
    client, db_session, hr_user, hr_headers, monkeypatch,
):
    """重发不能给「上一次重发已经送达」的人再发一封。

    `_resend_one` 从不改源行状态,所以那条陈旧 failed 行会一直躺在那;不按人判
    「最新一行是否已送达」的话,每点一次「只重发失败项」他就多收一封。
    """
    from app.notifications.models import NotificationOutbox
    from app.notifications.service import NotificationService

    hr, _ = hr_user
    batch = HrLeaveReminderBatch(
        created_by_person_id=str(hr.id), status="partial", test_mode=False,
        channels=["feishu"], filters={}, recipients_snapshot=[],
        recipient_count=1, delivered=0, failed=1,
        finished_at=datetime.now(UTC),
    )
    db_session.add(batch)
    await db_session.flush()
    stale = NotificationOutbox(
        recipient_person_id=hr.id, template_key=svc.TEMPLATE_KEY,
        title="t", body_text="(see template)", channels=["feishu"],
        status="failed", related_resource_type=BATCH_RESOURCE_TYPE,
        related_resource_id=batch.id,
        created_at=datetime.now(UTC) - timedelta(minutes=10),
    )
    clone = NotificationOutbox(
        recipient_person_id=hr.id, template_key=svc.TEMPLATE_KEY,
        title="t", body_text="(see template)", channels=["feishu"],
        status="delivered", related_resource_type=BATCH_RESOURCE_TYPE,
        related_resource_id=batch.id, created_at=datetime.now(UTC),
    )
    db_session.add_all([stale, clone])
    await db_session.flush()
    await db_session.commit()
    batch_id = batch.id

    resent: list[str] = []

    async def _capture(self, ids, **kw):
        resent.extend(str(i) for i in ids)
        return {"resent": len(ids), "skipped": 0, "new_ids": [], "skipped_detail": []}

    monkeypatch.setattr(NotificationService, "resend_outbox_rows", _capture)
    r = await client.post(
        f"/api/v1/hr/leave/expiry-reminder/batches/{batch_id}/resend-failed",
        headers=hr_headers,
    )
    assert r.status_code == 200, r.text
    assert resent == [], "这个人最新一行已送达,不该再被重发"

    # 而且台账要被修正:失败归零、状态从 failed/partial 翻回 completed。
    fresh = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert fresh.failed == 0
    assert fresh.delivered == 1
    assert fresh.status == "completed"
    assert fresh.error_summary is None


@pytest.mark.asyncio
async def test_completed_batch_clears_a_stale_reaper_error(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """drain 成功收尾时要清掉 reaper 留下的那条已经不成立的说明。

    否则页面会显示「已完成」+ 红色「一个人都没收到,请重跑」。
    """
    async def _ok(self, outbox, env, **kw):
        outbox.status = "delivered"
        outbox.delivered_channels = list(env.channels)

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _ok)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-stale-err",
    )
    # 模拟 reaper 在我们投递期间先下了结论
    batch.error_summary = (
        "The send task never started (process restart) — nobody was notified."
    )
    await db_session.commit()
    batch_id = batch.id

    await svc.run_batch(db_session, batch_id)
    fresh = await db_session.get(
        HrLeaveReminderBatch, batch_id, populate_existing=True,
    )
    assert fresh.status == "completed"
    assert fresh.error_summary is None, (
        "「已完成」不能同时挂着「一个人都没收到」——HR 会据此重跑整轮"
    )


# ── 14. 试发给指定的人(HR 反馈:默认全选,想只挑几个人试发很难) ────────


@pytest.mark.asyncio
async def test_test_send_goes_to_the_people_you_picked(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """试发可以指定几个人,每人收到**他自己**的真实数字。

    原来 test_mode 硬编码「只发给发起人自己」,于是 HR 想「挑两个同事试一下」只能
    去 281 行的默认全选里逐个取消 —— 这是他实际反馈的痛点。给别人发**他自己**的
    额度不是隐私问题;原来那条限制针对的是「拿别人的余额去演示」(发起人不在名单里
    时用示例数据的那个分支)。
    """
    recipients: list[str] = []
    contexts: dict[str, str] = {}

    async def _capture(self, outbox, env, **kw):
        recipients.append(str(env.recipient.id))
        contexts[str(env.recipient.id)] = env.context["account_lines"]
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _capture)
    hr, _ = hr_user
    picked = [cohort_data["soon"], cohort_data["both"]]

    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"],
        person_ids=picked, expected_recipient_count=None,
        allow_resend=False, test_mode=True,
        actor_person_id=hr.id, trace_id="t-pick",
    )
    await db_session.commit()
    assert batch.test_mode is True
    assert batch.recipient_count == 2
    await svc.run_batch(db_session, batch.id)

    assert sorted(recipients) == sorted(str(p) for p in picked)
    # 每人拿到的是自己的数字:soon 剩 5 天,both 是 1 天(法定年假那个桶)。
    assert "剩余 5 天" in contexts[str(cohort_data["soon"])]
    assert "剩余 1 天" in contexts[str(cohort_data["both"])]

    # 试发仍然**不计入**去重 —— 之后正式群发这些人照样会被选中。
    after = await svc.resolve_cohort(db_session, _filters())
    assert after.summary.reminded_recently_count == 0


@pytest.mark.asyncio
async def test_test_send_is_capped_so_it_cannot_become_a_mass_send(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """试发不计入去重,所以必须有个小上限 —— 否则「勾 281 人 + test_mode」就是
    一条绕开「近期已提醒」护栏的群发通道。"""
    from app.core.exceptions import ValidationError as AppValidationError

    monkeypatch.setattr(svc, "MAX_TEST_RECIPIENTS", 1)
    hr, _ = hr_user
    with pytest.raises(AppValidationError, match="at most 1 people"):
        await svc.create_batch(
            db_session, filters=_filters(), channels=["feishu"],
            person_ids=[cohort_data["soon"], cohort_data["both"]],
            expected_recipient_count=None, allow_resend=False, test_mode=True,
            actor_person_id=hr.id, trace_id="t-cap",
        )
    await db_session.rollback()


@pytest.mark.asyncio
async def test_test_send_reaches_people_outside_the_cohort(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """名单**外**的同事也能收到试发信,用的仍是他自己的真数字。

    这是 HR 报的那个「搜不到人」的根因:试发的选择器只列名单内的人,于是搜一个
    在 300 天后才清零(``later``)或只剩 0.2 天(``tiny``)的同事 → ``暂无数据``,
    看起来像搜索坏了。而「要提醒谁」(口径)和「我想给谁看一眼样信」(试发)本来
    就是两件无关的事。
    """
    contexts: dict[str, str] = {}

    async def _capture(self, outbox, env, **kw):
        contexts[str(env.recipient.id)] = env.context["account_lines"]
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _capture)
    hr, _ = hr_user
    picked = [cohort_data["later"], cohort_data["tiny"]]

    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"],
        person_ids=picked, expected_recipient_count=None,
        allow_resend=False, test_mode=True,
        actor_person_id=hr.id, trace_id="t-outside",
    )
    await db_session.commit()
    assert batch.recipient_count == 2
    await svc.run_batch(db_session, batch.id)

    assert set(contexts) == {str(p) for p in picked}
    # 各自的**真实**余额,不是示例数字。
    assert "剩余 3 天" in contexts[str(cohort_data["later"])]
    assert "剩余 0.2 天" in contexts[str(cohort_data["tiny"])]
    # 名单口径本身没被放宽 —— 真实群发仍然不含这两个人。
    cohort = await svc.resolve_cohort(db_session, _filters())
    picked_ids = {str(c.person_id) for c in cohort.candidates}
    assert str(cohort_data["later"]) not in picked_ids
    assert str(cohort_data["tiny"]) not in picked_ids


@pytest.mark.asyncio
async def test_test_send_preflight_explains_every_outcome(
    db_session, cohort_data, hr_user,
):
    """预检逐人给出结局,**和发送路径共用同一个函数**。

    前端拿这份结果在发送**之前**逐人显示,所以「为什么这个人收不到」不再是点了
    发送才弹出来的一句 422。
    """
    hr, _ = hr_user
    rows = await svc.resolve_test_recipients(
        db_session,
        filters=_filters(),
        person_ids=[
            cohort_data["soon"],        # 名单内
            cohort_data["later"],       # 窗口外,但有真额度
            cohort_data["tiny"],        # 余额低于阈值,但有真额度
            cohort_data["expired"],     # 额度已过清零日
            cohort_data["leaver"],      # 已离职
            cohort_data["no_email"],    # 没有工作邮箱
            cohort_data["welfare_only"],  # 没有所选类型的额度
            hr.id,                      # 发起人自己,无额度 → 示例数字
        ],
        actor_person_id=hr.id,
    )
    by_id = {str(r.person_id): r for r in rows}
    # 顺序 = HR 勾选的顺序(前端按这个顺序显示)。
    assert [str(r.person_id) for r in rows][0] == str(cohort_data["soon"])

    assert by_id[str(cohort_data["soon"])].source == svc.TEST_SOURCE_COHORT
    for key in ("later", "tiny"):
        assert by_id[str(cohort_data[key])].source == svc.TEST_SOURCE_OUT_OF_COHORT
        assert by_id[str(cohort_data[key])].row is not None

    blocked = {
        "expired": svc.BLOCK_PAST_DEADLINE,
        "leaver": svc.BLOCK_INACTIVE,
        "no_email": svc.BLOCK_NO_EMAIL,
        "welfare_only": svc.BLOCK_NO_MATCHING_LEAVE,
    }
    for key, reason in blocked.items():
        row = by_id[str(cohort_data[key])]
        assert row.source == svc.TEST_SOURCE_BLOCKED, key
        assert row.reason == reason, key
        assert row.row is None, key
        # 发不了也要能显示成一行,所以名字不能是裸 UUID。
        assert row.display_name and row.display_name != str(row.person_id)

    # 发起人自己没有可提醒的额度 → 明确标注的示例数字(只有他能拿到)。
    assert by_id[str(hr.id)].source == svc.TEST_SOURCE_SAMPLE


@pytest.mark.asyncio
async def test_test_send_refuses_the_people_who_cannot_receive_it(
    db_session, cohort_data, hr_user,
):
    """真的挑了发不出去的人 → 明确报错并**点名**,而不是静默少发一个。

    前端会先按预检把这些人跳过,所以这条路径只在竞态时命中(预检之后那个人恰好
    离职 / 清零日过了)—— 但发送侧必须自己守住,不能依赖前端。
    """
    from app.core.exceptions import ValidationError as AppValidationError

    hr, _ = hr_user
    with pytest.raises(AppValidationError, match="not an active employee"):
        await svc.create_batch(
            db_session, filters=_filters(), channels=["feishu"],
            person_ids=[cohort_data["soon"], cohort_data["leaver"]],
            expected_recipient_count=None, allow_resend=False, test_mode=True,
            actor_person_id=hr.id, trace_id="t-blocked",
        )
    await db_session.rollback()


@pytest.mark.asyncio
async def test_test_send_never_puts_sample_numbers_in_someone_elses_letter(
    db_session, cohort_data, hr_user,
):
    """示例数字只给发起人自己 —— 给第三方寄一封数字是假的「你的年假即将清零」
    是不能做的事,所以别人一律 blocked。"""
    hr, _ = hr_user
    rows = await svc.resolve_test_recipients(
        db_session,
        filters=_filters(),
        person_ids=[cohort_data["welfare_only"]],
        actor_person_id=hr.id,
    )
    assert [r.source for r in rows] == [svc.TEST_SOURCE_BLOCKED]
    assert all(r.source != svc.TEST_SOURCE_SAMPLE for r in rows)


@pytest.mark.asyncio
async def test_preflight_endpoint_is_hr_only_and_returns_per_person_rows(
    client, cohort_data, hr_headers, employee_headers,
):
    """端点门与 ``/preview`` 一致(响应里带真实姓名与余额)。"""
    path = "/api/v1/hr/leave/expiry-reminder/test-recipients"
    body = {
        "filters": _filters().model_dump(mode="json"),
        "person_ids": [str(cohort_data["soon"]), str(cohort_data["leaver"])],
    }
    r = await client.post(path, json=body, headers=employee_headers)
    assert r.status_code == 403

    r = await client.post(path, json=body, headers=hr_headers)
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["max_recipients"] == svc.MAX_TEST_RECIPIENTS
    assert data["sendable_count"] == 1
    sources = {row["person_id"]: row["source"] for row in data["recipients"]}
    assert sources[str(cohort_data["soon"])] == "cohort"
    assert sources[str(cohort_data["leaver"])] == "blocked"


@pytest.mark.asyncio
async def test_test_send_without_a_pick_still_goes_to_me(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """不指定人时保持原行为:只发给发起人自己(「先发给我自己看看」)。"""
    recipients: list[str] = []

    async def _capture(self, outbox, env, **kw):
        recipients.append(str(env.recipient.id))
        outbox.status = "delivered"

    from app.notifications.service import NotificationService

    monkeypatch.setattr(NotificationService, "deliver", _capture)
    hr, _ = hr_user
    hr_id = hr.id
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=True,
        actor_person_id=hr_id, trace_id="t-self",
    )
    await db_session.commit()
    await svc.run_batch(db_session, batch.id)
    assert recipients == [str(hr_id)]


# ── 14. HR 2026-08-18 的改版(圈人门槛 / 卡片 / 文案) ──────────────────


@pytest.mark.asyncio
async def test_zero_balance_bucket_still_shows_but_never_wins_the_deadline(
    db_session,
):
    """「法定 15 天 + 福利 0 天」的人:两个桶都要出现在卡片上,0 也照显示。

    HR 要的是「这一轮 9/30 到期的两个额度」的全貌 —— 少一格会让员工以为自己
    根本没有那个额度。但 0 天的桶只是展示项:它没有任何东西会在那天失效,所以
    不许它决定「最早清零日」(否则标题里的日期指向一个什么都不会损失的日子),
    也不许它把 ``expiring_balance_days`` 拉成 0。
    """
    pid = await _person(db_session, name="ZeroWelfare")
    _account(
        db_session, pid, booking_end=TODAY + timedelta(days=44), balance=15,
    )
    _account(
        db_session, pid, atype=WELFARE,
        booking_end=TODAY + timedelta(days=20), balance=0,
    )
    await db_session.flush()
    await db_session.commit()

    result = await svc.resolve_cohort(
        db_session, _filters(account_types=[LEGAL, WELFARE]),
    )
    row = next(c for c in result.candidates if str(c.person_id) == str(pid))
    assert {a.account_type for a in row.accounts} == {LEGAL, WELFARE}
    assert row.total_balance_days == pytest.approx(15)
    # 更近的那个清零日属于 0 天的桶 —— 它不该成为「即将到期」的那一天。
    assert row.earliest_deadline == TODAY + timedelta(days=44)

    # 卡片两格,0 那格如实写 0;信里的日期与名单行一致。
    stats = svc.build_card_stats(row.accounts, "zh")
    assert len(stats) == 2
    assert any("[h]0[/h]" in str(s.value) for s in stats)
    ctx = svc.build_context(
        display_name="X", accounts=row.accounts, lang="zh",
        hris_url="", today=TODAY,
    )
    assert ctx["deadline_date"] == (TODAY + timedelta(days=44)).isoformat()
    assert ctx["expiring_balance_days"] == "15"


@pytest.mark.asyncio
async def test_person_whose_every_matched_bucket_is_zero_is_never_reminded(
    db_session,
):
    """两个桶都是 0 的人一封都不发(HR 的第一条要求)。

    门槛是**人**的合计,不是账户 —— 按账户筛的老写法会把「15 + 0」的人砍成
    一格,而这个人(0 + 0)在两种写法下都必须出局。
    """
    pid = await _person(db_session, name="AllZero")
    _account(db_session, pid, booking_end=TODAY + timedelta(days=44), balance=0)
    _account(
        db_session, pid, atype=WELFARE,
        booking_end=TODAY + timedelta(days=44), balance=0,
    )
    await db_session.flush()
    await db_session.commit()

    result = await svc.resolve_cohort(
        db_session, _filters(account_types=[LEGAL, WELFARE]),
    )
    assert [str(c.person_id) for c in result.candidates] == []
    # 合计门槛:0.3 + 0.3 = 0.6 ≥ 0.5 的人则入选,两格都展示。
    pid2 = await _person(db_session, name="TwoSmall")
    _account(
        db_session, pid2, booking_end=TODAY + timedelta(days=44), balance=0.3,
    )
    _account(
        db_session, pid2, atype=WELFARE,
        booking_end=TODAY + timedelta(days=44), balance=0.3,
    )
    await db_session.flush()
    await db_session.commit()
    result = await svc.resolve_cohort(
        db_session, _filters(account_types=[LEGAL, WELFARE]),
    )
    row = next(c for c in result.candidates if str(c.person_id) == str(pid2))
    assert len(row.accounts) == 2


def test_default_copy_has_no_subtitle_and_no_helix_link():
    """HR 要求关掉的两样东西,用测试钉住「关掉了」。

    ① 卡片副标题那行灰字(「余额来自 HRIS(SuccessFactors)同步」);
    ② 邮件正文里那行 ``请到 Helix 查看: https://…``(走 ``{cta_line}``)。
    留一个空字符串是不够的:文案覆盖是**字段级合并**,空值会被当成「没覆盖」
    而回落到内置默认 —— 所以内置默认本身必须是空的。
    """
    copy = svc.builtin_copy()
    assert copy.card_subtitle == ""
    assert copy.card_subtitle_en == ""
    assert copy.card_subtitle_ja == ""
    for body in (copy.body_text, copy.body_text_en, copy.body_text_ja):
        # 正文里不留任何链接:``{cta_line}`` 是那行「请到 Helix 查看: https://…」
        # 的来源,``{portal_url}`` 是它的裸 URL 版本。(正文仍然**提到** Helix ——
        # 「Helix 中的数据仅作参考」是 HR 要的原话,那是文字不是链接。)
        assert "{cta_line}" not in body
        assert "{portal_url}" not in body
        assert "http" not in body
    # 三语标题都用占位符写日期,不写死 2026-09-30。
    for subject in (copy.subject, copy.subject_en, copy.subject_ja):
        assert "{deadline_date}" in subject
        assert "FY26" in subject
    svc.validate_copy(copy)  # 出厂文案必须过自己的校验


def test_default_filters_cover_both_forfeiting_annual_buckets():
    """出厂口径含福利年假 —— 它和法定年假一样 9/30 清零。"""
    from app.domains.hr.leave_insights_service import forfeits_at_period_end

    assert forfeits_at_period_end(WELFARE, {}) is True
    assert set(svc.default_filters().account_types) == {LEGAL, WELFARE, "MY_AnnuL"}


@pytest.mark.asyncio
async def test_hris_url_defaults_to_successfactors_until_hr_overrides_it(
    db_session,
):
    """按钮只剩一个,所以它的落点不能是空的。

    出厂值是 HR 指定的入口(2026-08-18 起是飞书 applink);HR 在页面上填了别的就用
    HR 的 —— 仍然是配置项,换 HRIS / 换深链不需要改代码。
    """
    from app.integrations.models import PlatformSetting

    assert await svc.get_hris_url(db_session) == svc.DEFAULT_HRIS_URL
    # 出厂值本身必须过写侧那道校验 —— 它是要挂在几百张卡片按钮上的 URL,而
    # HR 从 IM 里复制来的链接经常带着**发链接的人自己**的会话态(第一版 SF 链接
    # 就带着 ``?_s.crb=…`` 登录令牌,原样发给 500 个人只会各自失效)。
    assert svc.validate_hris_url(svc.DEFAULT_HRIS_URL) == svc.DEFAULT_HRIS_URL
    assert "_s.crb" not in svc.DEFAULT_HRIS_URL
    db_session.add(PlatformSetting(
        key=svc.SETTING_HRIS_URL, value="https://hris.example.com/timeoff",
    ))
    await db_session.commit()
    assert await svc.get_hris_url(db_session) == "https://hris.example.com/timeoff"


@pytest.mark.asyncio
async def test_test_send_preflight_says_zero_balance_not_no_deadline(
    db_session, hr_user,
):
    """试发预检要说真原因:「一天都不剩」不是「没有清零日」。

    余额 0 的桶现在会进快照(卡片要画那一格),但它不参与「最早清零日」——
    于是一个所有桶都是 0 的人 ``expires_in_days is None``,如果照搬那条分支就会
    被告知「他的额度没有清零日」,而他的额度明明有清零日、只是空了。HR 拿着这条
    原因去问当事人会问出一段莫名其妙的对话。
    """
    hr, _ = hr_user
    pid = await _person(db_session, name="ZeroEverything")
    _account(db_session, pid, booking_end=TODAY + timedelta(days=44), balance=0)
    await db_session.flush()
    await db_session.commit()

    resolved = await svc.resolve_test_recipients(
        db_session, filters=_filters(), person_ids=[pid],
        actor_person_id=hr.id,
    )
    assert [r.source for r in resolved] == [svc.TEST_SOURCE_BLOCKED]
    assert resolved[0].reason == svc.BLOCK_ZERO_BALANCE


# ── 15. HR 第二轮:排除名单 / 寄语 / FY 标签 / 大数字 ────────────────────


def test_fiscal_label_comes_from_the_period_end_not_the_fiscal_year_column():
    """FY 标签的判据是**周期结束年**,不是 ``fiscal_year`` 列。

    生产实测(2026-08-18):清零日同为 2026-09-30 的桶里,261 个是
    ``fiscal_year=2025``(2025-04-01 ~ 2026-03-31),另有 **7 个是
    ``fiscal_year=2026``**(2026-01-12 ~ 2026-03-31,1~3 月入职的同事)。周期和
    清零日完全一样,``fiscal_year`` 却差 1 —— 按 ``fiscal_year + 1`` 命名会把那
    7 个人的同一份额度标成「FY27」。

    另外这套 FY 编号和 ``core/fiscal_year.py`` 的**差一年**(那边 FY2026 =
    2026-04…2027-03,按开始年命名;HR 按结束年命名),所以这里不能去复用它。
    """
    period = date(2026, 3, 31)
    assert svc.fiscal_label(_snap(fiscal_year=2025, period_end=period)) == "FY26"
    assert svc.fiscal_label(_snap(fiscal_year=2026, period_end=period)) == "FY26"
    # 下一个周期
    assert svc.fiscal_label(
        _snap(fiscal_year=2026, period_end=date(2027, 3, 31)),
    ) == "FY27"
    # 没有 end_date 时退回 fiscal_year + 1(至少对 261 个人是对的)
    assert svc.fiscal_label(_snap(fiscal_year=2025, period_end=None)) == "FY26"
    assert svc.fiscal_label(
        _snap(fiscal_year=None, period_end=None),
    ) == ""


def test_card_stat_label_is_hr_wording_in_three_languages():
    """HR 定的措辞:「截至目前未休法定年假(FY26)」。"""
    from app.notifications.service import resolve_card_text

    acct = _snap(fiscal_year=2025, period_end=date(2026, 3, 31))
    zh = resolve_card_text(svc.build_card_stats([acct], "zh")[0].label, "zh")
    assert zh == "截至目前未休法定年假(FY26)"
    en = resolve_card_text(svc.build_card_stats([acct], "en")[0].label, "en")
    # ``localized_type_label`` 给的英文标签首字母大写,所以标签必须在句首 ——
    # ``Unused Statutory annual leave`` 是句中大写。
    assert en.endswith("still unused (FY26)")
    assert "Unused " not in en
    ja = resolve_card_text(svc.build_card_stats([acct], "ja")[0].label, "ja")
    assert ja.startswith("未消化の") and ja.endswith("(FY26)")


def test_stat_value_is_not_double_bolded_into_oblivion():
    """带标记的格子值**不再**被外层 ``**…**`` 二次包裹。

    生产实测(2026-08-18 第一版上线后的真机截图):余额渲染成
    ``**<font size='heading_3'>**4**</font> 天**`` —— 嵌套的 ``**`` 让飞书把
    **加粗和放大一起丢掉**,员工看到的是正文大小的「4 天」。同一个潜在缺陷从
    stat grid 上线起就一直在悄悄吃掉每一个 ``[r]`` 红色加粗值。

    没有标记的值仍然要加粗(老行为不能被这次修复顺手改掉)。
    """
    from app.notifications.service import CardStat, _render_stat_grid

    def _cell(value: str) -> str:
        grid = _render_stat_grid([CardStat(label="L", value=value)])
        return grid["columns"][0]["elements"][0]["content"].split("\n")[1]

    assert _cell("[h]4[/h] 天") == "<font size='heading_3'>**4**</font> 天"
    assert _cell("[r]$140.76[/r]") == "<font color='red'>**$140.76**</font>"
    # 无标记 → 照旧加粗
    assert _cell("23 人") == "**23 人**"


@pytest.mark.asyncio
async def test_excluded_person_still_listed_but_not_default_selected(
    db_session, hr_user,
):
    """排除是**默认值不是硬闸**:人还在名单里,只是默认不勾。"""
    hr, _ = hr_user
    keep = await _person(db_session, name="Keep")
    _account(db_session, keep, booking_end=TODAY + timedelta(days=30), balance=5)
    skip = await _person(db_session, name="Skip")
    _account(db_session, skip, booking_end=TODAY + timedelta(days=30), balance=7)
    await db_session.flush()
    await db_session.commit()

    await svc.set_exclusion(
        db_session, person_id=skip, note="长期病假", actor_person_id=hr.id,
    )
    await db_session.commit()

    result = await svc.resolve_cohort(db_session, _filters())
    picked = {str(c.person_id): c for c in result.candidates}
    # 仍然在名单里 —— HR 要看得见「他被跳过了」,而不是人间蒸发。
    assert str(skip) in picked
    assert picked[str(skip)].excluded is True
    assert picked[str(skip)].excluded_note == "长期病假"
    assert picked[str(keep)].excluded is False
    # 汇总口径和前端默认勾选同源。
    assert result.summary.recipient_count == 2
    assert result.summary.excluded_count == 1
    assert result.summary.default_selected_count == 1


@pytest.mark.asyncio
async def test_send_without_explicit_picks_skips_the_excluded(
    db_session, hr_user, monkeypatch,
):
    """不传 ``person_ids`` 的那条路径**也**要尊重排除名单。

    前端正常路径总是带 person_ids,所以这条分支很容易被写成只挡「近期已提醒」——
    然后一个脚本 / 一次「全选后清空搜索再发」就把 HR 明确排除掉的人发出去了,
    而页面上他明明是不勾的。
    """
    hr, _ = hr_user
    keep = await _person(db_session, name="KeepMe")
    _account(db_session, keep, booking_end=TODAY + timedelta(days=30), balance=5)
    skip = await _person(db_session, name="SkipMe")
    _account(db_session, skip, booking_end=TODAY + timedelta(days=30), balance=7)
    await db_session.flush()
    await db_session.commit()
    await svc.set_exclusion(
        db_session, person_id=skip, note=None, actor_person_id=hr.id,
    )
    await db_session.commit()

    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"], person_ids=[],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-exc",
    )
    await db_session.commit()
    ids = {r["person_id"] for r in batch.recipients_snapshot}
    assert ids == {str(keep)}


@pytest.mark.asyncio
async def test_exclusion_requires_a_real_person_and_can_be_cleared(
    db_session, hr_user,
):
    """打错的 id 会安静地躺在表里、永远匹配不到人,而 HR 以为自己排除掉了。"""
    from app.core.exceptions import NotFoundError

    hr, _ = hr_user
    with pytest.raises(NotFoundError):
        await svc.set_exclusion(
            db_session, person_id=uuid.uuid4(), note=None, actor_person_id=hr.id,
        )
    pid = await _person(db_session, name="Someone")
    await db_session.commit()
    await svc.set_exclusion(
        db_session, person_id=pid, note="  交接中  ", actor_person_id=hr.id,
    )
    await db_session.commit()
    assert (await svc.load_exclusions(db_session))[str(pid)] == "交接中"
    # 重复加入 = 就地改备注,不是第二行。
    await svc.set_exclusion(
        db_session, person_id=pid, note="已离岗", actor_person_id=hr.id,
    )
    await db_session.commit()
    rows = await svc.list_exclusions(db_session)
    assert len(rows) == 1 and rows[0][0].note == "已离岗"
    assert await svc.remove_exclusion(db_session, pid) is True
    await db_session.commit()
    assert await svc.load_exclusions(db_session) == {}
    # 删一个不存在的要如实回报,而不是假装成功。
    assert await svc.remove_exclusion(db_session, pid) is False


def test_signoff_validation_and_picking():
    """寄语会落进 platform_settings 再原样发给几百个人,所以先过校验。"""
    from app.core.exceptions import ValidationError

    cfg = svc.default_signoff()
    assert svc.validate_signoff(cfg) == cfg
    for bad in (
        {"mode": "sometimes", "lines": [{"zh": "a"}]},        # 未知模式
        {"mode": "random", "lines": []},                       # 空列表
        {"mode": "random", "lines": [{"zh": "", "en": "", "ja": ""}]},  # 整行全空
        {"mode": "fixed", "fixed_index": 9, "lines": [{"zh": "a"}]},    # 越界
        {"mode": "random", "lines": [{"zh": "x" * 201}]},      # 超长
    ):
        with pytest.raises(ValidationError):
            svc.validate_signoff(bad)

    fixed = {**cfg, "mode": "fixed", "fixed_index": 1}
    assert svc.pick_signoff(fixed, seed="a") == svc.pick_signoff(fixed, seed="b")
    assert svc.pick_signoff(fixed, seed="a")["zh"] == cfg["lines"][1]["zh"]
    # 随机是**可复现**的:同一批次重渲染(站内回看、resend 克隆)要拿到同一句。
    assert svc.pick_signoff(cfg, seed="batch-1") == svc.pick_signoff(cfg, seed="batch-1")
    seen = {
        svc.signoff_text(svc.pick_signoff(cfg, seed=f"b{i}"), "zh")
        for i in range(40)
    }
    assert len(seen) > 1, "random mode must actually vary across batches"
    # 语言阶梯:缺失时 ja → en → zh。
    assert svc.signoff_text({"zh": "中", "en": "", "ja": ""}, "ja") == "中"


@pytest.mark.asyncio
async def test_signoff_is_frozen_per_batch_and_reaches_the_card(
    db_session, cohort_data, hr_user, monkeypatch,
):
    """寄语在建批次时冻结,投递时读快照 —— 与余额/HRIS 入口同一套语义。"""
    envelopes: list = []

    async def _capture(self, outbox, env, **kw):
        envelopes.append(env)
        outbox.status = "delivered"

    from app.notifications.service import NotificationService, resolve_card_text

    monkeypatch.setattr(NotificationService, "deliver", _capture)
    hr, _ = hr_user
    batch = await svc.create_batch(
        db_session, filters=_filters(), channels=["feishu"],
        person_ids=[str(cohort_data["soon"])],
        expected_recipient_count=None, allow_resend=False, test_mode=False,
        actor_person_id=hr.id, trace_id="t-sign",
    )
    await db_session.commit()
    frozen = batch.filters["_signoff"]
    assert frozen["zh"] in [line["zh"] for line in svc.default_signoff()["lines"]]

    await svc.run_batch(db_session, batch.id)
    assert envelopes, "expected one envelope"
    footer = resolve_card_text(envelopes[0].card_footer, "zh")
    assert footer == frozen["zh"]
    # 平台默认的落款被换掉了 —— 这封信是催人去休假的,不是行政通知。
    assert "Helix IT" not in footer


@pytest.mark.asyncio
async def test_exclusion_endpoints_end_to_end_and_audited(
    client, db_session, hr_headers, employee_headers, hr_user,
):
    """走真实 HTTP 面 —— 两个 500 只有这样才会暴露。

    本地起服务点一遍抓到的(单测全绿的情况下):
    ① 名单**为空**时 `IN ("")` 撞上 ``GUID`` 的 bind ⇒ 整个 GET 500。空名单恰好是
       这个页面最常见的状态,也是最容易漏测的那个。
    ② `AuditService.log()` 少传 `actor_type` / `status` ⇒ 写侧 500。服务层单测调的是
       `svc.set_exclusion`,压根走不到路由里的审计那一行。
    """
    from app.audit.models import AuditLog

    path = "/api/v1/hr/leave/expiry-reminder/exclusions"
    # ① 空名单
    r = await client.get(path, headers=hr_headers)
    assert r.status_code == 200, r.text
    assert r.json()["data"] == {"items": [], "total": 0}

    pid = await _person(db_session, name="OnLeave")
    await db_session.commit()

    # 员工进不来(读写都不行)
    assert (await client.get(path, headers=employee_headers)).status_code == 403
    r = await client.post(
        path, json={"person_id": str(pid), "note": "长期病假"},
        headers=employee_headers,
    )
    assert r.status_code == 403

    # ② 写 + 审计
    r = await client.post(
        path, json={"person_id": str(pid), "note": "长期病假"}, headers=hr_headers,
    )
    assert r.status_code == 200, r.text
    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "hr.leave_expiry_reminder.exclusion_set",
        )
    )).scalars().all()
    assert len(rows) == 1 and str(rows[0].resource_id) == str(pid)

    r = await client.get(path, headers=hr_headers)
    items = r.json()["data"]["items"]
    assert len(items) == 1
    assert items[0]["note"] == "长期病假"
    assert items[0]["created_by_name"]  # 谁加的要看得见

    r = await client.delete(f"{path}/{pid}", headers=hr_headers)
    assert r.status_code == 200, r.text
    assert (await client.get(path, headers=hr_headers)).json()["data"]["total"] == 0
    # 删第二次 = 404,而不是假装成功。
    assert (await client.delete(f"{path}/{pid}", headers=hr_headers)).status_code == 404