"""Scheduled-job handlers for the role-change coordination domain
(issue #623).

Four cron keys, two of which can reach a human:

  * ``role_change.dispatch_reconcile`` (03:20) — replays org moves the
    inline coordinator swallowed. Writes only, notifies nobody.
  * ``role_change.review_backfill`` (03:40) — builds the review packet
    for any move that lacks one. Writes only, notifies nobody.
  * ``role_change.review_notify`` (09:30) — 🔴 pushes a packet to the
    mover's NEW MANAGER.
  * ``role_change.scope_review_sweep`` (10:00) — 🔴 pushes stale
    ``pending_review`` cloud grants to the scope owner + every active
    global admin.

🔴 **Both push handlers are gated on
``platform_settings['role_change.review']`` and both default to OFF**
(migration 610, 2026-09-05). The gate is read at SEND time, so
re-enabling the ``scheduled_jobs`` row is not sufficient to reach
anybody — an operator must also state a baseline policy.

Why: ``provision_policies`` is empty in production, so the platform has
no statement of what any role *should* hold. The packet therefore
reported the mover's whole **inventory** as though it were the
consequence of the move. On 2026-09-05 that produced nine cards — five
of them to the CTO, one to the CEO — against 223 items that have never
received a single decision. See ``mover_review_settings`` for the rule
and the reasoning.

Handler raises on per-row failure rather than returning
``status='failed'`` per the platform's handler-status discipline
(memory ``feedback_handler_status_must_raise``).
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID, uuid4

from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.models import CmdbCloudAccessGrant, CmdbCloudScope
from app.identity.person_models import IdentityPerson
from app.notifications.models import NotificationOutbox
from app.notifications.service import NotificationEnvelope, NotificationService
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

# Window for "stalled" rows. 30 days is the issue-spec default; the
# live value comes from ``platform_settings['role_change.review']
# .stale_grant_days`` so an operator can widen it without a deploy.
_REVIEW_STALE_DAYS = 30

# Outbox keys for dedup.
_TEMPLATE_KEY = "role_change_review_30d"
_RELATED_TYPE = "cmdb_cloud_access_grant"


@register_scheduled_job(
    key="role_change.scope_review_sweep",
    display_name="转岗 pending_review 扫描",
    description=(
        "每天 10:00 (Asia/Shanghai) 扫描 cmdb_cloud_access_grants 中 "
        "review_status='pending_review' 且 review_status_changed_at "
        "已超过阈值(platform_settings['role_change.review'].stale_grant_days,"
        "默认 30 天)的行,向 scope owner + 全部在职 global_admin 推卡催决断。\n"
        "🔴 **2026-09-05 起默认停用**:发送前重读 "
        "platform_settings['role_change.review'].stale_grant_push_enabled"
        "(默认 False),关着时整个 handler 直接空转返回,把这一行 enabled "
        "打开也发不出东西。当时积压 35 条 grant,满 30 天会一次产生约 175 张卡。\n"
        "通过 NotificationOutbox 的 (template_key, related_resource_id) "
        "做幂等去重,所以即便 cron 多次触发(如 worker 启动 backfill), "
        "scope owner 不会被同一行的同一卡片连续轰炸。"
    ),
)
async def run_role_change_scope_review_sweep(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Scan and notify. Raises RuntimeError on any per-row failure
    so the run row lands ``status='failed'`` (handler-status discipline).
    """
    from app.hr_lifecycle.mover_review_settings import load_review_settings

    # Same gate as ``review_notify``, same reason. This one fans out to
    # the scope owner AND every active global admin, so on 2026-09-05 it
    # was sitting on 35 grants that would have produced ~175 cards the
    # moment they crossed 30 days.
    settings = await load_review_settings(db)
    if not settings.stale_grant_push_enabled:
        return {
            "scanned": 0,
            "notified": 0,
            "skipped_already_sent": 0,
            "skipped_push_disabled": True,
            "reason": (
                "platform_settings['role_change.review']."
                "stale_grant_push_enabled is false"
            ),
        }

    stale_days = settings.stale_grant_days
    cutoff = datetime.now(UTC) - timedelta(days=stale_days)

    # Select the candidate grants. The composite index added in
    # migration 233 (review_status, review_status_changed_at) makes
    # this lookup index-only.
    stmt = (
        select(CmdbCloudAccessGrant)
        .where(
            CmdbCloudAccessGrant.review_status == "pending_review",
            CmdbCloudAccessGrant.review_status_changed_at < cutoff,
        )
        .order_by(CmdbCloudAccessGrant.review_status_changed_at.asc())
    )
    candidates = list((await db.execute(stmt)).scalars().all())

    scanned = len(candidates)
    notified = 0
    skipped_already_sent = 0
    errors: list[str] = []

    if not candidates:
        return {
            "scanned": 0,
            "notified": 0,
            "skipped_already_sent": 0,
        }

    # Pre-load the IT admin pool once — used as fallback recipient
    # when a scope has no ``owner_person_id``.
    it_admins = await _load_it_admin_pool(db)

    for grant in candidates:
        try:
            # Dedup: skip if an outbox row already exists for this
            # (template_key, related_resource_id). ``related_resource_id``
            # is CHAR(36) WITHOUT a GUID adapter on the column — values
            # round-trip as strings, so we MUST bind ``str(grant.id)``
            # for the WHERE clause to match (memory
            # ``reference_asyncmy_uuid_bind``: UUID bind on String/CHAR
            # silently returns 0 rows).
            already = await db.scalar(
                select(NotificationOutbox.id)
                .where(
                    NotificationOutbox.template_key == _TEMPLATE_KEY,
                    NotificationOutbox.related_resource_type == _RELATED_TYPE,
                    NotificationOutbox.related_resource_id == str(grant.id),
                )
                .limit(1)
            )
            if already is not None:
                skipped_already_sent += 1
                continue

            # CmdbCloudScope.id is Uuid; ``grant.scope_id`` is String(36)
            # — cast at boundary so the bind doesn't return 0 rows
            # (memory ``reference_asyncmy_uuid_bind``).
            try:
                scope_uuid = UUID(grant.scope_id)
            except (TypeError, ValueError):
                scope_uuid = None
            scope = None
            if scope_uuid is not None:
                scope = await db.scalar(
                    select(CmdbCloudScope).where(
                        CmdbCloudScope.id == scope_uuid,
                    )
                )
            recipients: list[tuple[IdentityPerson, bool]] = []
            if scope is not None and scope.owner_person_id:
                try:
                    owner_uuid = UUID(scope.owner_person_id)
                except (TypeError, ValueError):
                    owner_uuid = None
                if owner_uuid is not None:
                    owner = await db.scalar(
                        select(IdentityPerson).where(
                            IdentityPerson.id == owner_uuid,
                            IdentityPerson.deleted_at.is_(None),
                            IdentityPerson.employment_status == "active",
                        )
                    )
                    if owner is not None:
                        recipients.append((owner, True))
            # Always also notify IT admin pool — the issue spec asks
            # the card to land on both scope owner AND IT.
            for admin in it_admins:
                if not any(r.id == admin.id for r, _ in recipients):
                    recipients.append((admin, False))

            if not recipients:
                # No-one to tell. Don't raise — this is operationally
                # benign (the org has neither a scope owner nor an
                # IT admin pool); just log and move on.
                logger.warning(
                    "role_change.scope_review_sweep: no recipients "
                    "for grant=%s scope=%s — skipping notify",
                    grant.id, grant.scope_id,
                )
                continue

            notify = NotificationService(db)
            # ``recipients`` is a list of ``(person, is_scope_owner)``:
            # the two audiences need DIFFERENT landing pages, and one
            # shared URL is why this card was unactionable for its
            # primary recipient. An admin goes to the console; the scope
            # owner goes to their own scope page, which is the surface
            # they can actually open (the console route is admin-gated)
            # and where the decision now lives for them.
            for recipient, is_owner in recipients:
                target_url = (
                    # The owner's own inbox tab — the surface that lists
                    # exactly the grants this card is about, and one they
                    # can actually open. The scope detail page was the
                    # obvious guess and the wrong one: its `?tab=pending`
                    # is pending *requests* awaiting their approval, a
                    # different kind of pending entirely, so it would
                    # have landed them on a page that does not contain
                    # what they were asked about.
                    "/my/todos?tab=cloudGrantReview"
                    if is_owner else "/admin/role-changes"
                )
                result = await notify.dispatch(
                    NotificationEnvelope(
                        recipient=recipient,
                        title="云资源授权 pending_review > 30 天,请尽快决断",
                        # Fallback body only — the template
                        # (``role_change_review_30d``) is what renders,
                        # and it builds its own CTA from ``url``. Kept
                        # free of a hardcoded path so the two audiences'
                        # different landing pages cannot disagree with
                        # the prose.
                        body_text=(
                            f"角色 {grant.role_display_name or grant.role_key} "
                            f"在 scope {grant.scope_id} 上的持有人已转岗,"
                            f"该授权挂在 pending_review 状态已超过 "
                            f"{stale_days} 天,请确认保留还是回收。"
                        ),
                        channels=["feishu", "email"],
                        template_key=_TEMPLATE_KEY,
                        related_resource_type=_RELATED_TYPE,
                        related_resource_id=grant.id,
                        context={
                            "grant_id": str(grant.id),
                            "scope_id": grant.scope_id,
                            "role_key": grant.role_key,
                            "role_display_name": grant.role_display_name,
                            # Template placeholders: ``role_display_name`` is
                            # nullable (a raw {role_display_name} would render
                            # "None"), and the staleness threshold was only a
                            # module constant, never on the context.
                            "role_label": (
                                grant.role_display_name or grant.role_key
                            ),
                            "stale_days": stale_days,
                            "review_status_changed_at": (
                                grant.review_status_changed_at.isoformat()
                                if grant.review_status_changed_at else None
                            ),
                        },
                        url=target_url,
                    ),
                    trace_id=f"system:cron:role_change.scope_review_sweep:{uuid4()}",
                    actor_user_id=None,
                )
                logger.info(
                    "role_change.scope_review_sweep: dispatched to %s "
                    "delivered=%s feishu_err=%s email_err=%s",
                    recipient.work_email, result.delivered_channels,
                    result.feishu_error, result.email_error,
                )
            notified += 1
            # Per-grant commit so a partial failure later in the loop
            # doesn't lose the outbox rows we already wrote (review C1):
            # NotificationService.dispatch adds outbox rows to the
            # session; a single final commit would let one bad row
            # roll back N already-dispatched cards AND break dedup
            # next run (no outbox row to find).
            await db.commit()
        except Exception as exc:  # noqa: BLE001 — collect per-row, raise at end
            # Roll back THIS row's failed transaction so the session
            # is usable for the next iteration's commit. Without
            # rollback, asyncmy raises PendingRollbackError on the
            # next statement.
            try:
                await db.rollback()
            except Exception:  # noqa: BLE001
                pass
            logger.exception(
                "role_change.scope_review_sweep: per-row failure "
                "for grant=%s", grant.id,
            )
            errors.append(f"grant={grant.id}: {exc}")

    if errors:
        # Per ``feedback_handler_status_must_raise`` — raise here so the
        # run lands ``status='failed'`` and the failure notifier fires.
        # Bare ``return {"status": "failed", ...}`` would silently
        # mark the run ``ok``.
        raise RuntimeError(
            f"role_change.scope_review_sweep partial failure "
            f"({len(errors)}/{scanned}): " + " | ".join(errors[:5])
        )

    return {
        "scanned": scanned,
        "notified": notified,
        "skipped_already_sent": skipped_already_sent,
        "cutoff_utc": cutoff.isoformat(),
    }


async def _load_it_admin_pool(db: AsyncSession) -> list[IdentityPerson]:
    """Return active IT admins (global_admin role). The role-change
    sweep CC's them on every card so an absent scope owner can't
    park a stalled review forever."""
    from app.auth.models import AuthRole, AuthUserRole
    from app.core.rbac import GLOBAL_ADMIN_ROLE

    rows = (
        await db.execute(
            select(IdentityPerson)
            .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id)
            .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
            .where(
                AuthRole.name == GLOBAL_ADMIN_ROLE,
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
            )
            .distinct()
        )
    ).scalars().all()
    return list(rows)


# ── Straggler reconciler ────────────────────────────────────────────
#
# The confirm path fires ``handle_role_change`` inline and the
# coordinator swallows its own exceptions, so a failure there is
# invisible: the staged row still reaches ``confirmed`` and the move is
# applied to the person, but no lifecycle event exists and no cloud
# grant is flagged for review. That hole is not hypothetical — it is
# the SAME shape as the two gate bugs migration 565 closes, and it
# survives them. A batch that half-dispatched (process killed between
# the person commit and the coordinator) leaves exactly this residue.
#
# So detection and dispatch are separated: staging records the drift
# flags durably on the staged row, and this sweep asserts that every
# recorded drift eventually has a lifecycle event to show for it.
_RECONCILE_LOOKBACK_DAYS = 180
_RECONCILE_LIMIT = 200
_RECONCILE_TRIGGER = "reconcile"


def _staged_org_drift(person_data: dict[str, Any] | None) -> bool:
    """True when the staged row recorded a manager / dept / title move.

    ``is True`` rather than truthiness: rows staged before the flags
    existed carry ``None``, and ``None`` must read as "this row never
    made a claim", not as "no move" — the counters below report those
    separately so an operator can tell an empty sweep from a blind one.
    """
    pd = person_data or {}
    return any(
        pd.get(key) is True
        for key in ("manager_changed", "dept_changed", "title_changed")
    )


@register_scheduled_job(
    key="role_change.dispatch_reconcile",
    display_name="转岗派发对账(补发漏掉的组织变动)",
    description=(
        "每天 03:20 (Asia/Shanghai) 复核最近 180 天内**已确认**的 HRIS 暂存事件:"
        "凡是暂存时记录了组织变动(manager / dept / title 任一为 true)、"
        "却没有对应 role_change 生命周期事件的,原样补发一次协调器 —— "
        "写事件、把该员工的云授权标成 pending_review、跑一次坐席差异预览。\n"
        "为什么需要它:确认路径是**内联**调用协调器的,而协调器按平台惯例"
        "自吞异常。于是一次失败是完全隐形的——暂存行照样变成 confirmed、"
        "人员行照样被改,但事件没写、授权没标。进程在「人员已提交」和"
        "「协调器已跑」之间被杀,留下的正是这个残留。检测(暂存时落标记)"
        "与派发(这里)因此被分开:标记是持久的,这个扫描保证每一条被记录的"
        "变动最终都有一个事件与之对应。\n"
        "补发的事件按**当时观察到的日期**落 effective_date,不backdate 到今天;"
        "trigger_source='reconcile' 与内联写入区分。按 "
        "(person_id, event_type, effective_date) 去重,重复运行是空转 —— "
        "已经被管理员判成 confirmed_keep 的授权不会被重新翻成待复核。\n"
        "只读外部:不打厂商接口、不调用模型、不发通知。"
    ),
)
async def run_role_change_dispatch_reconcile(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core.coordinators import role_change_coordinator
    from app.domains.hr.models import HrLifecycleEvent, HrSyncBatch, HrSyncStagedEvent
    from app.identity.identity_reconcile import resolve_person_by_hris_user_id

    lookback = int(config.get("lookback_days") or _RECONCILE_LOOKBACK_DAYS)
    limit = int(config.get("limit") or _RECONCILE_LIMIT)
    cutoff = datetime.now(UTC) - timedelta(days=lookback)

    # Drift is filtered in SQL, not in Python, so ``limit`` bounds the
    # rows that MATTER. Filtering afterwards would let the overwhelming
    # majority — confirmed updates that recorded no move — consume the
    # budget and starve the handful this sweep exists for, and it would
    # do so silently.
    #
    # ``JSON_UNQUOTE(JSON_EXTRACT(...)) = 'true'`` on MySQL 8 and TiDB:
    # JSON true → 'true', false → 'false', an absent key → NULL (so a
    # row staged before the flags existed is excluded rather than
    # guessed at — the same distinction ``_staged_org_drift`` draws).
    #
    # NOT ``= CAST('true' AS JSON)``, which reads correctly and does not
    # work: SQLAlchemy binds the literal THROUGH the JSON type, so
    # ``"true"`` is re-encoded as the JSON *string* ``"true"`` and never
    # equals the JSON *boolean*. The predicate then silently matches
    # nothing — caught here only because the reconciler tests assert on
    # a non-zero ``drifted`` count rather than merely "did not crash".
    #
    # A JSON string ``"true"`` would also match this form; the flags are
    # written from Python bools on one code path, so that value cannot
    # arise, and ``_staged_org_drift`` re-checks with ``is True`` anyway.
    drift_in_sql = or_(*[
        func.json_unquote(
            func.json_extract(HrSyncStagedEvent.person_data, f"$.{key}")
        ) == "true"
        for key in ("manager_changed", "dept_changed", "title_changed")
    ])

    rows = (
        await db.execute(
            select(HrSyncStagedEvent, HrSyncBatch.hris_instance_id)
            .join(HrSyncBatch, HrSyncBatch.id == HrSyncStagedEvent.batch_id)
            .where(
                HrSyncStagedEvent.status == "confirmed",
                HrSyncStagedEvent.action.in_(("update", "reactivate")),
                HrSyncStagedEvent.existing_person_id.is_not(None),
                HrSyncStagedEvent.created_at >= cutoff,
                drift_in_sql,
            )
            .order_by(HrSyncStagedEvent.created_at.asc())
            .limit(limit)
        )
    ).all()

    scanned = 0
    drifted = 0
    replayed = 0
    already_present = 0
    unrecoverable = 0
    grants_marked = 0
    errors: list[dict[str, Any]] = []

    for staged, instance_id in rows:
        scanned += 1
        person_data = staged.person_data or {}
        if not _staged_org_drift(person_data):
            continue
        drifted += 1

        # ``created_at`` is when Helix OBSERVED the move. SF carries no
        # separate "move effective on" date on this payload, and dating
        # the event today would put a July move in today's access
        # context — the one thing a forensic timeline must not do.
        event_date = staged.created_at.date()
        person_id = staged.existing_person_id

        exists = await db.scalar(
            select(HrLifecycleEvent.id).where(
                HrLifecycleEvent.person_id == person_id,
                HrLifecycleEvent.event_type
                == role_change_coordinator.ROLE_CHANGE_EVENT_TYPE,
                HrLifecycleEvent.effective_date == event_date,
            ).limit(1)
        )
        if exists is not None:
            already_present += 1
            continue

        prior = person_data.get("_prior_role")
        if not prior:
            # Staged before the before/after blob existed. A role change
            # with no stated "before" is not a role change anyone can
            # review, and inventing one from the person's CURRENT row
            # would record the move as a no-op. Counted, not guessed.
            unrecoverable += 1
            continue

        jobs = (staged.employment_data or {}).get("_jobs_snapshot") or []
        head = jobs[0] if jobs else {}
        new_mgr_hris = head.get("managerId")
        new_mgr_person_id: str | None = None
        if new_mgr_hris:
            # Resolve exactly the way the confirm path does — alias
            # aware, instance scoped. A manager referenced by a
            # secondary userId otherwise reads as "manager removed".
            resolved = await resolve_person_by_hris_user_id(
                db, str(new_mgr_hris), hris_instance_id=instance_id,
            )
            new_mgr_person_id = str(resolved.id) if resolved else None

        try:
            summary = await role_change_coordinator.handle_role_change(
                db,
                person_id,
                before=prior,
                after={
                    "dept_code": head.get("department"),
                    "job_title": person_data.get("job_title"),
                    "manager_person_id": new_mgr_person_id,
                },
                actor_user_id=None,
                effective_date=event_date,
                trigger_source=_RECONCILE_TRIGGER,
            )
        except Exception as exc:  # noqa: BLE001 — one row must not end the sweep
            logger.exception(
                "role_change.dispatch_reconcile: replay failed for "
                "person=%s staged=%s", person_id, staged.id,
            )
            errors.append({"staged_event_id": str(staged.id), "error": str(exc)})
            continue

        if summary.get("error"):
            # The coordinator swallowed it — which is the exact failure
            # mode this sweep exists to catch, so surface it rather than
            # counting the row as replayed.
            errors.append({
                "staged_event_id": str(staged.id),
                "error": str(summary["error"]),
            })
            continue
        if summary.get("event_written"):
            replayed += 1
            grants_marked += int(summary.get("grants_marked") or 0)
        else:
            already_present += 1

    await db.commit()

    result = {
        # ``scanned`` counts rows the SQL drift filter already admitted,
        # so ``scanned == limit`` means the window is deeper than one
        # run — reported rather than silently truncated.
        "scanned": scanned,
        "truncated": scanned >= limit,
        "drifted": drifted,
        "replayed": replayed,
        "already_present": already_present,
        "unrecoverable_no_prior_role": unrecoverable,
        "grants_marked": grants_marked,
        "errors": errors[:10],
        "error_count": len(errors),
    }
    if errors:
        # Handler-status discipline: a swallowed per-row failure must
        # still turn the JOB red, or the sweep that exists to find
        # invisible failures becomes invisible itself.
        raise RuntimeError(
            f"role_change.dispatch_reconcile: {len(errors)} of {drifted} "
            f"drifted rows failed to replay — {errors[:3]}"
        )
    return result


# ── Review packets: top-up and notify ───────────────────────────────
#
# The packet (migration 568) is built inline by the coordinator, which
# swallows its own exceptions — so an inline-only build fails
# invisibly, the same hole PR #3315 closed for the lifecycle event.
# These two sweeps close it for the packet: one makes sure every
# access-relevant role change HAS a packet, the other makes sure a
# packet that exists actually reaches the person who can answer it.
_PACKET_LOOKBACK_DAYS = 90
_PACKET_LIMIT = 50
_NOTIFY_LIMIT = 50
_REVIEW_TEMPLATE = "role_change_review_open"


@register_scheduled_job(
    key="role_change.review_backfill",
    display_name="转岗复核包裹补建",
    description=(
        "每天 03:40 (Asia/Shanghai) 复核最近 90 天内的 role_change 生命周期"
        "事件:凡是**跨部门或换经理**、却没有对应复核包裹的,补建一份。\n"
        "为什么需要它:包裹由协调器内联构建,而协调器按平台惯例自吞异常 —— "
        "一次失败完全隐形,人员行照样改了、事件照样写了,只是没人被问过。"
        "内联构建保证「移动落地即有包裹」,这个扫描保证「没有一次移动没有包裹」。\n"
        "只改职级的移动**不建包裹**(权限跟的是汇报关系和组织,不是职级字符串),"
        "这与协调器不为纯职级变动翻云授权是同一条规则。\n"
        "按 (person_id, effective_date) 幂等;重复运行是空转,已有包裹只会被"
        "重扫补项,不会覆盖任何已经做出的决定。\n"
        "**同时重算开放包裹的差异**(2026-09-05 合并后复审补):差异引擎跑在"
        "建包裹里,而这个扫描此前只为「没有包裹的移动」建包裹 —— 于是既有包裹"
        "的差异被冻结在建它的那一刻,而引擎的全部承诺正是「空策略表时建的包裹,"
        "会在策略配好的那天开始有意义」。没有这一步,那句承诺需要有人记得逐份"
        "点「重新扫描」。重算按 delta_computed_at 最旧优先(NULL 最先),"
        "只写 delta_* 四个字段,幂等,复活不了任何决定或 withheld 戳,"
        "也不会让任何包裹变得可推送(发送路径仍然要过 may_push)。\n"
        "只读外部,不发通知。"
    ),
)
async def run_role_change_review_backfill(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core.coordinators.role_change_coordinator import (
        ROLE_CHANGE_EVENT_TYPE,
    )
    from app.domains.hr.models import (
        ROLE_REVIEW_OPEN,
        HrLifecycleEvent,
        HrRoleChangeReview,
    )
    from app.hr_lifecycle.mover_review_service import build_review_packet

    lookback = int(config.get("lookback_days") or _PACKET_LOOKBACK_DAYS)
    limit = int(config.get("limit") or _PACKET_LIMIT)
    cutoff = (datetime.now(UTC) - timedelta(days=lookback)).date()

    rows = (
        await db.execute(
            select(HrLifecycleEvent)
            .outerjoin(
                HrRoleChangeReview,
                (HrRoleChangeReview.person_id == HrLifecycleEvent.person_id)
                & (
                    HrRoleChangeReview.effective_date
                    == HrLifecycleEvent.effective_date
                ),
            )
            .where(
                HrLifecycleEvent.event_type == ROLE_CHANGE_EVENT_TYPE,
                HrLifecycleEvent.effective_date >= cutoff,
                HrRoleChangeReview.id.is_(None),
            )
            .order_by(HrLifecycleEvent.effective_date.asc())
            .limit(limit)
        )
    ).scalars().all()

    built = 0
    not_packet_worthy = 0
    reannotated = 0
    errors: list[dict[str, Any]] = []
    for event in rows:
        detail = event.detail or {}
        before = detail.get("before") or {}
        after = detail.get("after") or {}
        try:
            packet = await build_review_packet(
                db, event.person_id,
                before=before, after=after,
                effective_date=event.effective_date,
                lifecycle_event_id=event.id,
            )
        except Exception as exc:  # noqa: BLE001 — one row must not end the sweep
            logger.exception(
                "role_change.review_backfill: failed for event=%s", event.id,
            )
            errors.append({"event_id": str(event.id), "error": str(exc)})
            continue
        if packet is None:
            # A title-only move, or a person since removed. Counted so an
            # operator can tell "nothing to build" from "built nothing".
            not_packet_worthy += 1
        else:
            built += 1

    # ── Re-annotate the OPEN packets ────────────────────────────────
    #
    # Found in the post-merge review: ``build_review_packet`` runs the
    # delta engine, but this sweep only calls it for moves that have NO
    # packet. An existing packet's delta was therefore frozen at the
    # moment it was built — and the whole promise of the engine is that
    # a packet built while ``provision_policies`` was empty starts
    # meaning something the day policies are configured. Without this
    # loop that promise needed somebody to remember to press 「重新扫描」
    # on each packet, one at a time.
    #
    # Re-annotation reads the CURRENT policy table and org tree, writes
    # ``delta_*`` and nothing else, and is idempotent. It cannot resurrect
    # a decision or a withheld stamp, and it cannot make a packet
    # pushable on its own — the send path still asks ``may_push``.
    from app.hr_lifecycle.mover_delta import annotate_review

    open_packets = (
        await db.execute(
            select(HrRoleChangeReview)
            .where(HrRoleChangeReview.status == ROLE_REVIEW_OPEN)
            # NOT ``.nulls_first()``: MySQL / TiDB reject the
            # ``NULLS FIRST`` syntax outright (1064). MySQL does sort
            # NULLs first under a plain ASC, but relying on that leaves
            # the intent invisible — the explicit boolean key says
            # "never-annotated packets go to the front" and is portable.
            .order_by(
                HrRoleChangeReview.delta_computed_at.is_(None).desc(),
                HrRoleChangeReview.delta_computed_at.asc(),
            )
            .limit(limit)
        )
    ).scalars().all()
    for packet in open_packets:
        try:
            await annotate_review(db, packet)
            reannotated += 1
        except Exception as exc:  # noqa: BLE001 — one packet must not end the sweep
            logger.exception(
                "role_change.review_backfill: re-annotation failed for "
                "review=%s", packet.id,
            )
            errors.append({"review_id": str(packet.id), "error": str(exc)})

    await db.commit()
    result = {
        "scanned": len(rows),
        "built": built,
        "skipped_not_packet_worthy": not_packet_worthy,
        # Oldest-first by ``delta_computed_at`` (NULLs first), so a
        # backlog larger than ``limit`` drains rather than starving the
        # same tail every night.
        "reannotated": reannotated,
        "reannotate_truncated": len(open_packets) >= limit,
        "errors": errors[:10],
        "error_count": len(errors),
        "truncated": len(rows) >= limit,
    }
    if errors:
        raise RuntimeError(
            f"role_change.review_backfill: {len(errors)} failure(s) across "
            f"{len(rows)} packet build(s) and {len(open_packets)} "
            f"re-annotation(s) — {errors[:3]}"
        )
    return result


@register_scheduled_job(
    key="role_change.review_notify",
    display_name="转岗复核 · 通知新经理",
    description=(
        "每天 09:30 (Asia/Shanghai) 把**尚未通知过**的开放复核包裹推给它的"
        "复核人 —— 也就是接手这位同事的**新经理**。\n"
        "🔴 **2026-09-05 起默认停用,并且即使把这一行 enabled 打开也发不出东西。**"
        "发送前会重读 platform_settings['role_change.review']:"
        "``employee_push_enabled`` 默认 False,``require_baseline_for_push`` 默认 True。"
        "原因是生产上 ``provision_policies`` 是空表 —— 平台没有任何「某个岗位该有什么」"
        "的基线,于是包裹报的是这个人的**存量**(24 类持有物)而不是这次转岗造成的"
        "**增量**。9/5 那天发出去的 9 张卡里,CTO 一个人收了 5 张共 115 项;"
        "223 个 item 至今 0 次决断。**没有基线,就不要推送。**\n"
        "为什么发给经理而不是 IT:一个管理员看到「这个人有 9 条云授权」"
        "无从判断他还需不需要;唯一知道的人是刚把他接过来的那位。新经理已离职、"
        "或这次移动没留下经理时,回落给在职 global_admin —— 一个指向动不了手的人的"
        "收件人字段比空的更糟,因为它看起来像已经有人管了。\n"
        "卡片是**摘要**:真实包裹是几百项跨十来个类别,把它们列进卡片就是这个"
        "功能想避免的复核风暴。卡片给总量、类别分布和(如果生成过)一段模型"
        "摘要,处理在页面上按类别整批做。\n"
        "每份包裹只通知一次(``notified_at``);被 ``notify_suppressed_at`` 盖过戳的"
        "包裹永远不再进入队列 —— 迁移 610 给 9/5 之前的全部积压盖了戳,"
        "重新启用不会把积压一次性轰出去。"
    ),
)
async def run_role_change_review_notify(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from sqlalchemy import func as _func

    from app.domains.hr.models import (
        ROLE_REVIEW_OPEN,
        HrRoleChangeReview,
        HrRoleChangeReviewItem,
    )
    from app.hr_lifecycle.mover_holdings import REVIEW as _REVIEW_DISPOSITION
    from app.hr_lifecycle.mover_review_settings import load_review_settings
    from app.notifications import fanout
    from app.scheduled_jobs.failure_notifier import list_admin_recipients

    # 🔴 The gate. Read at SEND time, not at schedule time, so turning
    # the cron row back on is not by itself sufficient to reach an
    # employee — the operator has to state a baseline policy as well.
    # Returning early (rather than filtering rows) keeps the run cheap
    # and makes the reason legible in the run record.
    settings = await load_review_settings(db)
    if not settings.employee_push_enabled:
        return {
            "pending": 0,
            "notified": 0,
            "skipped_push_disabled": True,
            "reason": (
                "platform_settings['role_change.review'].employee_push_enabled "
                "is false — no baseline, no push (2026-09-05)"
            ),
        }

    limit = int(config.get("limit") or _NOTIFY_LIMIT)
    reviews = (
        await db.execute(
            select(HrRoleChangeReview)
            .where(
                HrRoleChangeReview.status == ROLE_REVIEW_OPEN,
                HrRoleChangeReview.notified_at.is_(None),
                # Withheld packets never re-enter the queue. Without
                # this, re-enabling the push would fire the entire
                # accumulated backlog at once — the failure mode that
                # made migration 610 necessary in the first place.
                HrRoleChangeReview.notify_suppressed_at.is_(None),
                HrRoleChangeReview.reviewable_items > 0,
            )
            .order_by(HrRoleChangeReview.created_at.asc())
            .limit(limit)
        )
    ).scalars().all()

    # The per-packet half of the gate. ``employee_push_enabled`` above
    # says "this org has decided pushes are allowed"; this says "and
    # this particular packet has something real to say".
    #
    # ``baseline_source`` is ``'none'`` on every packet built before the
    # delta engine, so with no baseline configured this filter empties
    # the batch — which is the whole point. Filtered in Python rather
    # than SQL because ``may_push`` is the single definition of the rule
    # and a WHERE clause would be a second, drifting copy of it.
    considered = len(reviews)
    reviews = [
        r for r in reviews
        if settings.may_push(
            baseline_source=r.baseline_source, delta_items=r.delta_items,
        )
    ]
    skipped_no_delta = considered - len(reviews)

    if not reviews:
        return {
            "pending": 0,
            "notified": 0,
            "considered": considered,
            "skipped_no_delta": skipped_no_delta,
        }

    envelopes: list[NotificationEnvelope] = []
    fallback_admins: list[IdentityPerson] | None = None
    notified = 0
    unroutable = 0

    for review in reviews:
        mover = await db.get(IdentityPerson, review.person_id)
        if mover is None:
            continue

        recipients: list[IdentityPerson] = []
        if review.reviewer_person_id:
            reviewer = await db.get(IdentityPerson, review.reviewer_person_id)
            if (
                reviewer is not None
                and reviewer.deleted_at is None
                and reviewer.employment_status == "active"
            ):
                recipients = [reviewer]
        if not recipients:
            if fallback_admins is None:
                fallback_admins = await list_admin_recipients(db)
            recipients = list(fallback_admins)
            unroutable += 1
        if not recipients:
            logger.warning(
                "role_change.review_notify: no recipient for review=%s",
                review.id,
            )
            continue

        counts = (
            await db.execute(
                select(
                    HrRoleChangeReviewItem.resource_type,
                    _func.count(),
                )
                .where(
                    HrRoleChangeReviewItem.review_id == review.id,
                    HrRoleChangeReviewItem.disposition == _REVIEW_DISPOSITION,
                )
                .group_by(HrRoleChangeReviewItem.resource_type)
                .order_by(_func.count().desc())
            )
        ).all()
        # Top five categories by volume, then a tail count. A card that
        # lists ten categories is a card nobody finishes reading.
        head = counts[:5]
        summary = "、".join(f"{rt} {n}" for rt, n in head)
        if len(counts) > len(head):
            summary += f" 等 {len(counts)} 类"

        ai_line = ""
        if review.ai_summary_zh:
            ai_line = f"AI 初判:{review.ai_summary_zh}\n\n"

        context = {
            "person_name": mover.display_name or "同事",
            "before_dept": review.before_dept_path
            or review.before_dept_code
            or "(未知)",
            "after_dept": review.after_dept_path
            or review.after_dept_code
            or "(未知)",
            "reviewable": review.reviewable_items,
            "category_count": len(counts),
            "category_summary": summary or "(无)",
            "ai_line": ai_line,
        }
        for person in recipients:
            envelopes.append(NotificationEnvelope(
                recipient=person,
                title="转岗权限复核",
                body_text="",
                channels=["feishu", "email"],
                template_key=_REVIEW_TEMPLATE,
                context=context,
                related_resource_type="hr_role_change_review",
                related_resource_id=review.id,
                url=f"/admin/role-changes?review={review.id}",
            ))
        review.notified_at = datetime.now(UTC)
        notified += 1

    await db.commit()
    if envelopes:
        await fanout.enqueue_then_drain(
            db, envelopes, trace_id="",
            label="role_change.review_notify",
        )
    return {
        "pending": len(reviews),
        "considered": considered,
        # Packets whose delta could not be established (no baseline) or
        # was empty. A high number here is not a fault — it is the gate
        # doing its job, and the fix is a baseline, not a louder cron.
        "skipped_no_delta": skipped_no_delta,
        "notified": notified,
        "envelopes": len(envelopes),
        # Packets whose new manager could not be reached went to IT.
        # Reported because a rising number here means the org data is
        # losing manager lines, not that IT is getting more diligent.
        "routed_to_admins": unroutable,
    }