"""Unified cross-domain action inbox for the mobile workspace.

``build_action_inbox`` collects every "waiting for ME to act" item the
caller can see, across domains, into one prioritised list. It is the
backing logic for both ``GET /api/v1/m/aggregate/inbox`` and the
``mobile_action_inbox`` MCP tool — neither duplicates the query.

Design rules:

  * READ ONLY. Each item carries a ``deep_link`` to the page that
    actually mutates it (and, where a one-tap action is safe, an inline
    ``actions`` descriptor). The aggregator never writes and never
    becomes a source of truth — it reads each domain's own pending state
    so a decision made elsewhere is reflected on the next refresh.
  * ONLY WHAT *I* CAN ACT ON. The bar is strict: an item belongs here
    only when the CALLER is the person who acts on it — the recipient of
    a transfer, the assignee of a task, the owner deciding a grant, the
    auditor disposing a finding. A company-wide queue the caller merely
    has *read* authority over is an OBSERVATION surface (the dedicated
    admin/overview page), NOT a personal to-do. Concretely: Feishu-
    mirrored ``ApprovalRequest`` rows are NOT surfaced here — Helix keeps
    no per-approver attribution for them (``ApprovalStep.approver_id`` is
    never populated; the approver acts in the Feishu app, which is their
    real inbox), so listing every company-wide ``pending`` approval as
    "my to-do" was noise the caller could neither own nor clear. The
    admin approvals *overview* page carries that company-level view.
  * ROLE-SCOPED off ``role_names``. ``get_current_user`` has already
    applied the ``X-View-Role`` narrowing, so a global_admin "viewing as
    员工" sees only the employee-scope sources here — same contract as
    the rest of the platform.
  * EXTENSIBLE. A new item type is a new provider function appended to
    the gated list in ``build_action_inbox``; the schema is type-agnostic
    so no Pydantic change is needed to add a source.

Provider coverage (this revision):

  self (every caller)   asset_transfer_confirm, provisioning_ack,
                        close_tasks, access_attestation, support_reply,
                        attendance_confirm, inventory_response,
                        onboarding_ack
  auditor               compliance_finding
  global_admin          support_assigned
  leaver-blocked admin  leaver_blocker
  dual-sign signer      leaver_override_signoff (named secondary signer)
  HR Ops / HRBP duty    github_id_approval (named approver; HR sees all)
  scope owner/cloud     cloud_grant_approval (owner duty; cloud_admin sees all)
  cloud_admin           cloud_scope_approval

Every surfaced type deep-links to a mobile page that can actually perform
the action — an item is never a dead-end. The cloud grant/scope approvals
are decidable in-app: grant approve/reject carry no step-up, and the
scope-request step-up (``require_browser_session``) distinguishes PAT from
JWT — the Feishu webview's OAuth-minted JWT passes it — so the mobile
decision pages host approve/reject inline rather than handing off.
"""
from __future__ import annotations

from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import UTC, date, datetime
from uuid import UUID

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

from app.core.rbac import (
    AUDITOR_ROLE,
    GLOBAL_ADMIN_ROLE,
    can_decide_github_id_change,
    can_delete_any_mail_group,
    can_manage_cloud,
    can_manage_leaver_blocked,
    can_secondary_sign_leaver_override,
    has_role,
)
from app.domains.asset.models import AssetSoftwareCatalog
from app.domains.attendance.constants import (
    STATUS_PENDING_INTERN,
    STATUS_PENDING_LEADER,
)
from app.domains.attendance.models import AttendanceCycle, AttendanceReport
from app.domains.auditor.models import (
    AccessCertification,
    AccessCertificationItem,
    AuditorFinding,
)
from app.domains.cmdb.models import (
    GRANT_STATUS_PENDING,
    SCOPE_REQUEST_STATUS_FEISHU_SENT,
    SCOPE_REQUEST_STATUS_PENDING,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbCloudScopeRequest,
)
from app.domains.email_group.models import (
    EmailGroup,
    EmailGroupJoinRequest,
    EmailGroupOwner,
)
from app.domains.github_org.models import GithubIdChangeRequest
from app.domains.leaver_blocked.models import (
    DUAL_SIGN_PENDING_SECONDARY,
    LeaverDualSignOverride,
)
from app.domains.offboarding.models import (
    PROC_CANCELLED,
    PROC_COMPLETED,
    RESO_PENDING,
    SEVERITY_BLOCKER,
    XFER_PENDING,
    AssetTransferRequest,
    OffboardingBlocker,
    OffboardingProcess,
)
from app.domains.provisioning.models import (
    STATUS_IN_PROGRESS,
    STATUS_NOTIFIED,
    STATUS_PENDING,
    ProvisioningTask,
)
from app.domains.support.models import (
    STATUS_IN_PROGRESS as SUPPORT_IN_PROGRESS,
)
from app.domains.support.models import (
    STATUS_OPEN as SUPPORT_OPEN,
)
from app.domains.support.models import (
    STATUS_TRIAGING as SUPPORT_TRIAGING,
)
from app.domains.support.models import (
    STATUS_WAITING_REQUESTER as SUPPORT_WAITING_REQUESTER,
)
from app.domains.support.models import (
    SupportTicket,
)
from app.identity.person_models import IdentityPerson
from app.mobile.schemas import (
    MobileActionDescriptor,
    MobileActionInboxResponse,
    MobileActionItem,
)

# Item types. Kept as module constants so providers, tests, and the
# frontend i18n keys (``mobile.inbox.type.<value>``) all reference the
# same literals.
TYPE_ASSET_TRANSFER = "asset_transfer_confirm"
TYPE_PROVISIONING = "provisioning_ack"
TYPE_FINDING = "compliance_finding"
TYPE_LEAVER_BLOCKER = "leaver_blocker"
TYPE_CLOSE_TASKS = "close_tasks"
TYPE_ACCESS_ATTESTATION = "access_attestation"
TYPE_SUPPORT_REPLY = "support_reply"
TYPE_SUPPORT_ASSIGNED = "support_assigned"
TYPE_ATTENDANCE_CONFIRM = "attendance_confirm"
TYPE_INVENTORY_RESPONSE = "inventory_response"
TYPE_ONBOARDING_ACK = "onboarding_ack"
TYPE_LEAVER_OVERRIDE = "leaver_override_signoff"
TYPE_CLOUD_GRANT_APPROVAL = "cloud_grant_approval"
TYPE_CLOUD_SCOPE_APPROVAL = "cloud_scope_approval"
TYPE_CLOUD_RENEWAL_APPROVAL = "cloud_renewal_approval"
TYPE_EMAIL_GROUP_JOIN_APPROVAL = "email_group_join_approval"
TYPE_GITHUB_ID_APPROVAL = "github_id_approval"

# Sort priority: lower = more urgent = surfaces first.
_SEV_RANK = {"urgent": 0, "warning": 1, "info": 2}

# Per-source cap. The inbox is a TRIAGE surface, not a queue browser —
# when a single source exceeds this, the dedicated per-domain page is the
# full list. Counts are computed off the (capped) candidate set, so a
# source larger than the cap badges as the cap; the home aggregate's
# COUNT-based badges remain the exact figure.
_PER_SOURCE_CAP = 200

# All compliance findings deep-link to the unified mobile findings
# worklist (W4-PR14), where they're decidable (acknowledge / accept_risk
# / resolve) — not to the per-control read-only detector pages. One
# decidable surface across every control mirrors the desktop worklist.
_FINDING_DEEP_LINK = "/m/auditor/findings"


@dataclass
class _Cand:
    """Internal candidate carrying a naive-UTC sort timestamp.

    The public schema serialises ``created_at`` to an ISO string; sorting
    on the string is unsafe because aware (``+00:00``) and naive forms —
    which MySQL/TiDB hand back depending on fresh-load vs commit — sort
    differently. We keep the comparable datetime here and serialise once.
    """

    item: MobileActionItem
    sort_dt: datetime


def _naive(dt: datetime) -> datetime:
    """Drop tzinfo for cross-source comparison.

    MySQL round-trips ``DateTime(timezone=True)`` as naive on fresh load
    but aware right after commit; normalising avoids a
    ``can't compare offset-naive and offset-aware`` crash when one source
    was just written and another was loaded from disk.
    """
    return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt


def _finding_severity(severity: str) -> str:
    if severity in ("critical", "high"):
        return "urgent"
    if severity == "medium":
        return "warning"
    return "info"


async def _provider_asset_transfer(db: AsyncSession, person_id: UUID) -> list[_Cand]:
    rows = (
        await db.execute(
            select(AssetTransferRequest)
            .where(
                and_(
                    AssetTransferRequest.recipient_id == person_id,
                    AssetTransferRequest.status == XFER_PENDING,
                )
            )
            .order_by(AssetTransferRequest.created_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_ASSET_TRANSFER}:{r.id}",
                type=TYPE_ASSET_TRANSFER,
                title=r.asset_label,
                subtitle=r.reason,
                severity="info",
                created_at=r.created_at.isoformat(),
                deep_link=f"/m/asset-transfer/{r.id}",
                actions=[
                    MobileActionDescriptor(key="confirm", style="primary"),
                    MobileActionDescriptor(key="reject", style="danger"),
                ],
            ),
            sort_dt=_naive(r.created_at),
        )
        for r in rows
    ]


async def _provider_provisioning(db: AsyncSession, person_id: UUID) -> list[_Cand]:
    rows = (
        await db.execute(
            select(ProvisioningTask, AssetSoftwareCatalog.name)
            .outerjoin(
                AssetSoftwareCatalog,
                AssetSoftwareCatalog.id == ProvisioningTask.software_id,
            )
            .where(
                and_(
                    # The ASSIGNEE is the person who fulfils the task (the
                    # software owner / IT admin who clicks complete/fail) —
                    # NOT user_id (the subject the access is *for*). The
                    # detail page only shows the complete/fail actions to
                    # the assignee, so surfacing by user_id gave the subject
                    # a dead-end inbox row while the actual actor never saw
                    # it. Matches the legacy /provisioning/tasks/my source.
                    ProvisioningTask.assignee_id == person_id,
                    ProvisioningTask.status.in_(
                        [STATUS_PENDING, STATUS_IN_PROGRESS, STATUS_NOTIFIED],
                    ),
                )
            )
            .order_by(ProvisioningTask.created_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).all()
    out: list[_Cand] = []
    for task, software_name in rows:
        out.append(
            _Cand(
                item=MobileActionItem(
                    id=f"{TYPE_PROVISIONING}:{task.id}",
                    type=TYPE_PROVISIONING,
                    # Software name when the task targets a catalogued
                    # SaaS; otherwise the target_type enum, which the
                    # client localises.
                    title=software_name or task.target_type,
                    subtitle=task.action,  # grant | revoke — client maps
                    severity="warning" if task.status == STATUS_NOTIFIED else "info",
                    created_at=task.created_at.isoformat(),
                    deep_link=f"/m/provisioning-task/{task.id}",
                ),
                sort_dt=_naive(task.created_at),
            )
        )
    return out


async def _provider_findings(db: AsyncSession, _person_id: UUID) -> list[_Cand]:
    rows = (
        await db.execute(
            select(AuditorFinding)
            .where(AuditorFinding.status == "open")
            .order_by(AuditorFinding.last_seen_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_FINDING}:{r.id}",
                type=TYPE_FINDING,
                title=r.title,
                subtitle=r.subject_display or r.control_key,
                severity=_finding_severity(r.severity),
                created_at=r.last_seen_at.isoformat(),
                deep_link=_FINDING_DEEP_LINK,
            ),
            sort_dt=_naive(r.last_seen_at),
        )
        for r in rows
    ]


async def _provider_leaver_blockers(db: AsyncSession, _person_id: UUID) -> list[_Cand]:
    rows = (
        await db.execute(
            # Terminal processes excluded (2026-09-05) — see the same
            # filter in ``mobile/aggregate.py``. A cancelled process keeps
            # its pending blockers forever (cancel does not resolve them,
            # and ``reconcile_blockers`` refuses terminal processes), so
            # without this they surface as ``urgent`` action items that no
            # operator can ever clear — and, because they are ordered
            # newest-first against a ``_PER_SOURCE_CAP``, a freshly
            # cancelled leaver's dead rows would push the live ones that
            # DO need action out of the inbox entirely.
            select(OffboardingBlocker)
            .join(
                OffboardingProcess,
                OffboardingProcess.id == OffboardingBlocker.process_id,
            )
            .where(
                OffboardingBlocker.resolution == RESO_PENDING,
                OffboardingProcess.status.notin_(
                    [PROC_COMPLETED, PROC_CANCELLED],
                ),
            )
            .order_by(OffboardingBlocker.created_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_LEAVER_BLOCKER}:{r.id}",
                type=TYPE_LEAVER_BLOCKER,
                title=r.label,
                subtitle=r.resource_type,
                severity="urgent" if r.severity == SEVERITY_BLOCKER else "warning",
                created_at=r.created_at.isoformat(),
                deep_link=f"/m/admin/offboarding/{r.process_id}",
            ),
            sort_dt=_naive(r.created_at),
        )
        for r in rows
    ]


async def _provider_close_tasks(db: AsyncSession, person_id: UUID) -> list[_Cand]:
    """One aggregate item when the caller owns open month-end close tasks.

    Pull-only (surfaces in the todo inbox, never a push). Cheap for the
    99% who own none — the owner lookup is a single indexed MAX query that
    returns nothing. Severity escalates if any owned task is overdue.
    """
    from app.domains.finance import close_self_service

    payload = await close_self_service.my_close_tasks(db, person_id=str(person_id))
    open_tasks = [
        t for t in payload["tasks"]
        if t.status not in ("completed", "na")
    ]
    if not open_tasks:
        return []
    today = date.today()
    deadlines = [t.deadline for t in open_tasks if t.deadline is not None]
    earliest = min(deadlines) if deadlines else None
    overdue = any(d < today for d in deadlines)
    soon = any(0 <= (d - today).days <= 2 for d in deadlines)
    severity = "urgent" if overdue else ("warning" if soon else "info")
    # Sort key: surface earlier when something is due sooner.
    sort_dt = (
        datetime.combine(earliest, datetime.min.time())
        if earliest else datetime.now(UTC).replace(tzinfo=None)
    )
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_CLOSE_TASKS}:{payload['period']}",
                type=TYPE_CLOSE_TASKS,
                title=payload["label"] or payload["period"],
                subtitle=earliest.isoformat() if earliest else None,
                severity=severity,
                created_at=(payload["close_date"] or today.isoformat()),
                deep_link="/m/close-tasks",
            ),
            sort_dt=_naive(sort_dt),
        )
    ]


async def _provider_access_attestation(
    db: AsyncSession, person_id: UUID,
) -> list[_Cand]:
    """One aggregate item when a self-attestation campaign is open AND the
    caller still has undecided items in it.

    Self-scope (every caller): the half-yearly access self-attestation is
    a personal obligation regardless of the active perspective — an admin
    still has to attest their own access. Pull-only. Cheap for the 99% of
    the year with no open campaign — the campaign lookup returns None and
    we skip the per-person count entirely. Severity escalates as the
    campaign due date approaches / passes.
    """
    cert = (
        await db.execute(
            select(AccessCertification)
            .where(
                AccessCertification.campaign_type == "self_attestation",
                AccessCertification.status == "in_progress",
            )
            .order_by(AccessCertification.started_at.desc())
            .limit(1)
        )
    ).scalar_one_or_none()
    if cert is None:
        return []
    pending = int(
        (
            await db.execute(
                select(func.count())
                .select_from(AccessCertificationItem)
                .where(
                    and_(
                        AccessCertificationItem.certification_id == cert.id,
                        AccessCertificationItem.person_id == person_id,
                        AccessCertificationItem.decision == "pending",
                    )
                )
            )
        ).scalar()
        or 0
    )
    if pending == 0:
        return []
    due = _naive(cert.due_at) if cert.due_at else None
    days_left = (due.date() - date.today()).days if due else None
    if days_left is not None and days_left < 0:
        severity = "urgent"
    elif days_left is not None and days_left <= 7:
        severity = "warning"
    else:
        severity = "info"
    # Sort by the due date so a closer deadline surfaces first; fall back
    # to "now" when the campaign carries no due date.
    sort_dt = due if due else datetime.now(UTC).replace(tzinfo=None)
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_ACCESS_ATTESTATION}:{cert.id}",
                type=TYPE_ACCESS_ATTESTATION,
                title=cert.label,
                # Undecided-item count — the client localises the unit.
                subtitle=str(pending),
                severity=severity,
                deadline=due.isoformat() if due else None,
                created_at=_naive(cert.started_at).isoformat(),
                deep_link="/m/access-review",
            ),
            sort_dt=sort_dt,
        )
    ]


async def _provider_support_reply(db: AsyncSession, person_id: UUID) -> list[_Cand]:
    """Help-desk tickets I RAISED that are waiting on my reply.

    Self-scope (every caller): a ticket in ``waiting_requester`` is
    blocked on the requester answering IT's question — so it is the
    requester's to-do, not IT's. Deep-links to the employee ticket page.
    """
    rows = (
        await db.execute(
            select(SupportTicket)
            .where(
                and_(
                    SupportTicket.requester_id == person_id,
                    SupportTicket.status == SUPPORT_WAITING_REQUESTER,
                )
            )
            .order_by(SupportTicket.updated_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_SUPPORT_REPLY}:{r.id}",
                type=TYPE_SUPPORT_REPLY,
                title=r.title,
                subtitle=r.ticket_number,
                severity="warning",
                created_at=r.created_at.isoformat(),
                deep_link=f"/m/support/{r.id}",
            ),
            sort_dt=_naive(r.updated_at or r.created_at),
        )
        for r in rows
    ]


async def _provider_support_assigned(db: AsyncSession, person_id: UUID) -> list[_Cand]:
    """Help-desk tickets ASSIGNED to me as IT that need my work.

    Gated on the IT workbench role (global_admin) upstream. Scoped to
    ``assignee_id == me`` so an admin only sees the tickets they own, not
    the whole unassigned queue (that's the workbench page's job). Excludes
    ``waiting_requester`` — there the ball is in the requester's court
    (surfaced to them via :func:`_provider_support_reply`), not mine.
    """
    rows = (
        await db.execute(
            select(SupportTicket)
            .where(
                and_(
                    SupportTicket.assignee_id == person_id,
                    SupportTicket.status.in_(
                        [SUPPORT_OPEN, SUPPORT_TRIAGING, SUPPORT_IN_PROGRESS],
                    ),
                )
            )
            .order_by(SupportTicket.updated_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_SUPPORT_ASSIGNED}:{r.id}",
                type=TYPE_SUPPORT_ASSIGNED,
                title=r.title,
                subtitle=r.ticket_number,
                # Un-started (open/triaging) needs a first response →
                # warning; actively-worked (in_progress) → info.
                severity="info" if r.status == SUPPORT_IN_PROGRESS else "warning",
                created_at=r.created_at.isoformat(),
                deep_link=f"/m/admin/support/tickets/{r.id}",
            ),
            sort_dt=_naive(r.updated_at or r.created_at),
        )
        for r in rows
    ]


async def _provider_attendance_confirm(
    db: AsyncSession, person_id: UUID,
) -> list[_Cand]:
    """Intern attendance reports waiting on MY confirmation.

    Self-scope (interns + their leaders). Two-stage workflow: a
    ``pending_intern`` report is the intern's to-do (``person_id == me``);
    a ``pending_leader`` report is the leader's (``leader_person_id ==
    me``). Cheap: both legs are single-column-indexed.
    """
    rows = (
        await db.execute(
            select(AttendanceReport, AttendanceCycle.cycle_key)
            .join(
                AttendanceCycle,
                AttendanceCycle.id == AttendanceReport.cycle_id,
            )
            .where(
                (
                    (AttendanceReport.status == STATUS_PENDING_INTERN)
                    & (AttendanceReport.person_id == person_id)
                )
                | (
                    (AttendanceReport.status == STATUS_PENDING_LEADER)
                    & (AttendanceReport.leader_person_id == person_id)
                )
            )
            .order_by(AttendanceReport.updated_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).all()
    out: list[_Cand] = []
    for r, cycle_key in rows:
        ts = r.generated_at or r.created_at
        out.append(
            _Cand(
                item=MobileActionItem(
                    id=f"{TYPE_ATTENDANCE_CONFIRM}:{r.id}",
                    type=TYPE_ATTENDANCE_CONFIRM,
                    # Period label (YYYY-MM); the client renders the month.
                    title=cycle_key,
                    subtitle=r.status,  # pending_intern | pending_leader
                    severity="warning",
                    created_at=ts.isoformat(),
                    deep_link=f"/m/attendance/{r.id}",
                ),
                sort_dt=_naive(r.updated_at or ts),
            )
        )
    return out


async def _provider_inventory_response(
    db: AsyncSession, person_id: UUID,
) -> list[_Cand]:
    """Asset-inventory (盘点) items routed to me and still pending my
    response, in active cycles.

    Self-scope (asset holders). Collapsed into ONE aggregate item — the
    holder responds to all of them on the single ``/m/inventory`` page,
    so N per-item rows would just be noise. Reuses the same
    ``list_pending_for_person`` the /me/inventory page uses so the count
    can never drift from what the page shows.
    """
    from app.domains.asset.inventory_service import AssetInventoryService

    items = await AssetInventoryService(db).list_pending_for_person(person_id)
    pending = [it for it in items if it.status == "pending"]
    if not pending:
        return []
    # Sort key = the oldest pending item's snapshot time, so a holder who
    # has been sitting on a cycle surfaces above a fresh one.
    oldest = min((it.created_at for it in pending), default=None)
    sort_dt = _naive(oldest) if oldest else datetime.now(UTC).replace(tzinfo=None)
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_INVENTORY_RESPONSE}:{pending[0].cycle_id}",
                type=TYPE_INVENTORY_RESPONSE,
                # Count of items awaiting my response — client localises.
                title=str(len(pending)),
                severity="warning",
                created_at=sort_dt.isoformat(),
                deep_link="/m/inventory",
            ),
            sort_dt=sort_dt,
        )
    ]


async def _provider_onboarding_ack(
    db: AsyncSession, person_id: UUID,
) -> list[_Cand]:
    """入职须知 acknowledgements I still owe, while I'm onboarding.

    Self-scope (new hires). Gated on an OPEN onboarding process so a
    long-tenured employee who never happened to ack a checklist item does
    NOT get a perpetual to-do — the ack obligation only exists inside the
    onboarding window. Collapsed into one aggregate item deep-linking to
    the personal lifecycle page where the checklist lives.
    """
    from app.domains.onboarding.ack_service import OnboardingAckService
    from app.domains.onboarding.models import (
        PROC_OPEN_STATUSES,
        OnboardingProcess,
    )

    proc = (
        await db.execute(
            select(OnboardingProcess.id).where(
                and_(
                    OnboardingProcess.person_id == person_id,
                    OnboardingProcess.status.in_(PROC_OPEN_STATUSES),
                )
            ).limit(1)
        )
    ).scalar_one_or_none()
    if proc is None:
        return []
    statuses = await OnboardingAckService(db).my_ack_status(person_id)
    pending = [s for s in statuses if not s.get("acknowledged")]
    if not pending:
        return []
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_ONBOARDING_ACK}:{proc}",
                type=TYPE_ONBOARDING_ACK,
                title=str(len(pending)),  # count — client localises
                severity="warning",
                created_at=datetime.now(UTC).replace(tzinfo=None).isoformat(),
                deep_link="/m/lifecycle",
            ),
            sort_dt=datetime.now(UTC).replace(tzinfo=None),
        )
    ]


async def _provider_leaver_override(
    db: AsyncSession, person_id: UUID,
) -> list[_Cand]:
    """Leaver dual-sign overrides waiting on MY secondary sign-off.

    Role-gated on ``can_secondary_sign_leaver_override`` (global_admin /
    hr baseline) AND scoped to the NAMED secondary signer
    (``secondary_signer_user_id == me``) — the row carries a specific
    signer chosen at initiation, and ``secondary_approve`` enforces the
    exact identity match. High-stakes + rare; deep-links to the decision
    page under ``/m/admin``.
    """
    rows = (
        await db.execute(
            select(LeaverDualSignOverride)
            .where(
                and_(
                    LeaverDualSignOverride.secondary_signer_user_id == person_id,
                    LeaverDualSignOverride.status == DUAL_SIGN_PENDING_SECONDARY,
                )
            )
            .order_by(LeaverDualSignOverride.created_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_LEAVER_OVERRIDE}:{r.id}",
                type=TYPE_LEAVER_OVERRIDE,
                # Primary signer's reason preview — the client renders the
                # full context on the detail page.
                title=(r.primary_reason or "")[:80],
                severity="urgent",
                deadline=r.expires_at.isoformat() if r.expires_at else None,
                created_at=r.created_at.isoformat(),
                deep_link=f"/m/admin/leaver-override/{r.id}",
            ),
            sort_dt=_naive(r.created_at),
        )
        for r in rows
    ]


async def _provider_cloud_grant_approval(
    db: AsyncSession, person_id: UUID, role_names: list[str],
) -> list[_Cand]:
    """Pending cloud-access GRANTS awaiting MY approval.

    Mirrors ``GET /cmdb/my-pending-grant-approvals`` exactly so the inbox
    and that endpoint never disagree on who owes an approval:
      * ``can_manage_cloud`` (cloud_admin / global_admin) → every pending
        grant org-wide.
      * a plain scope owner → pending grants only on scopes they own,
        excluding self-requested rows (four-eyes; self-approve would 409).
    Runs regardless of perspective because owning a scope is a data-driven
    duty, not a role — an admin "viewing as employee" still sees grants on
    scopes THEY own (their own duty), just not the org-wide set. Decidable
    in-app (no step-up) → deep-links to the mobile grant-approval page.
    """
    q = (
        select(
            CmdbCloudAccessGrant,
            CmdbCloudScope.display_name,
            CmdbCloudScope.external_id,
        )
        .join(CmdbCloudScope, CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id)
        .where(CmdbCloudAccessGrant.status == GRANT_STATUS_PENDING)
    )
    if not can_manage_cloud(role_names):
        q = q.where(
            CmdbCloudScope.owner_person_id == str(person_id),
            CmdbCloudAccessGrant.person_id != str(person_id),
        )
    rows = (
        await db.execute(
            q.order_by(CmdbCloudAccessGrant.created_at.desc()).limit(_PER_SOURCE_CAP)
        )
    ).all()
    out: list[_Cand] = []
    for grant, scope_display, scope_external in rows:
        out.append(
            _Cand(
                item=MobileActionItem(
                    id=f"{TYPE_CLOUD_GRANT_APPROVAL}:{grant.id}",
                    type=TYPE_CLOUD_GRANT_APPROVAL,
                    title=scope_display or scope_external or grant.scope_id,
                    subtitle=grant.role_display_name or grant.role_key,
                    severity="warning",
                    created_at=grant.created_at.isoformat(),
                    deep_link=f"/m/cloud/grant-approval/{grant.id}",
                ),
                sort_dt=_naive(grant.created_at),
            )
        )
    return out


async def _provider_cloud_renewal_approval(
    db: AsyncSession, person_id: UUID, role_names: list[str],
) -> list[_Cand]:
    """Cloud grant RENEWAL requests awaiting MY decision.

    Same data-driven-duty shape as ``_provider_cloud_grant_approval``:
    ``can_manage_cloud`` sees the org-wide queue, a plain scope owner sees
    only their own scopes, and self-submitted rows are excluded because
    self-approval is refused (it would raise, so surfacing it would be a
    to-do nobody can complete).

    Why this exists: renewal became an approval flow on 2026-07-29, and the
    approver's reminder card lands in Feishu — i.e. on a phone. Without an
    inbox item and a mobile decision page the card would be a dead end, and
    the most likely failure of the whole flow is silence: nobody says no,
    they just never open it, and the access lapses.

    One item per SUBMISSION (``request_group_id``), not per role — the
    owner decides the person once.
    """
    from app.domains.cmdb.models import (
        RENEWAL_STATUSES_OPEN,
        CmdbCloudGrantRenewalRequest,
    )

    q = (
        select(
            CmdbCloudGrantRenewalRequest,
            CmdbCloudScope.display_name,
            CmdbCloudScope.external_id,
            IdentityPerson.display_name,
        )
        .join(
            CmdbCloudScope,
            CmdbCloudScope.id == CmdbCloudGrantRenewalRequest.scope_id,
        )
        .join(
            IdentityPerson,
            IdentityPerson.id == CmdbCloudGrantRenewalRequest.person_id,
        )
        .where(
            CmdbCloudGrantRenewalRequest.status.in_(RENEWAL_STATUSES_OPEN),
        )
    )
    if not can_manage_cloud(role_names):
        q = q.where(
            CmdbCloudScope.owner_person_id == str(person_id),
            CmdbCloudGrantRenewalRequest.person_id != str(person_id),
        )
        # Both open statuses are actionable by a plain owner since the
        # cloud-admin second signature was removed (2026-08-26): a legacy
        # ``pending_admin`` row is one this owner already signed, and
        # ``decide_renewal_group`` now finalizes it for them instead of
        # 403ing. Same rule in
        # ``CloudRenewalService.list_pending_for_approver`` — the two must
        # not drift, or the badge and the page disagree.
    rows = (
        await db.execute(
            q.order_by(CmdbCloudGrantRenewalRequest.created_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).all()
    out: list[_Cand] = []
    seen_groups: set[str] = set()
    for req, scope_display, scope_external, holder_name in rows:
        group = str(req.request_group_id or req.id)
        if group in seen_groups:
            continue
        seen_groups.add(group)
        out.append(
            _Cand(
                item=MobileActionItem(
                    id=f"{TYPE_CLOUD_RENEWAL_APPROVAL}:{group}",
                    type=TYPE_CLOUD_RENEWAL_APPROVAL,
                    title=scope_display or scope_external or str(req.scope_id),
                    subtitle=(
                        f"{holder_name or ''} · {int(req.requested_ttl_days)}d"
                    ).strip(" ·"),
                    severity="warning",
                    created_at=req.created_at.isoformat(),
                    deep_link=f"/m/cloud/renewal-approval/{group}",
                ),
                sort_dt=_naive(req.created_at),
            )
        )
    return out


async def _provider_email_group_join_approval(
    db: AsyncSession, person_id: UUID, role_names: list[str],
) -> list[_Cand]:
    """Pending mail-group JOIN requests awaiting MY decision.

    Mirrors ``GET /email-groups/join-requests/pending`` exactly so the
    inbox and that endpoint never disagree on who owes a decision:
      * ``can_delete_any_mail_group`` (HR / IT Admin) → every pending
        join request org-wide.
      * a plain group owner → pending requests only on groups they own,
        excluding self-requested rows (four-eyes).
    Runs regardless of perspective because owning a mail group is a
    data-driven duty, not a role — an admin "viewing as employee" still
    sees requests on groups THEY own. Decidable in-app → deep-links to
    the directory's 待审批 view.
    """
    q = (
        select(
            EmailGroupJoinRequest,
            EmailGroup.email,
            EmailGroup.name,
        )
        .join(EmailGroup, EmailGroup.id == EmailGroupJoinRequest.email_group_id)
        .where(
            EmailGroupJoinRequest.status == "pending",
            EmailGroup.status == "active",
        )
    )
    if not can_delete_any_mail_group(role_names):
        owned_subq = (
            select(EmailGroupOwner.email_group_id)
            .where(EmailGroupOwner.person_id == person_id)
        )
        q = q.where(
            EmailGroupJoinRequest.email_group_id.in_(owned_subq),
            EmailGroupJoinRequest.requester_id != person_id,
        )
    rows = (
        await db.execute(
            q.order_by(EmailGroupJoinRequest.created_at.desc()).limit(
                _PER_SOURCE_CAP
            )
        )
    ).all()
    # Resolve requester display names in one query for readable titles.
    requester_ids = {req.requester_id for req, _e, _n in rows}
    names: dict[UUID, str] = {}
    if requester_ids:
        people = (
            await db.execute(
                select(IdentityPerson.id, IdentityPerson.display_name,
                       IdentityPerson.work_email)
                .where(IdentityPerson.id.in_(requester_ids))
            )
        ).all()
        names = {
            pid: (dn or we or str(pid)) for pid, dn, we in people
        }
    out: list[_Cand] = []
    for req, group_email, group_name in rows:
        who = names.get(req.requester_id, str(req.requester_id))
        out.append(
            _Cand(
                item=MobileActionItem(
                    id=f"{TYPE_EMAIL_GROUP_JOIN_APPROVAL}:{req.id}",
                    type=TYPE_EMAIL_GROUP_JOIN_APPROVAL,
                    title=group_email or group_name or "",
                    subtitle=who,
                    severity="info",
                    created_at=req.created_at.isoformat(),
                    # Mobile-native decision surface (added
                    # 2026-07-28). Was "/email-groups?review=1",
                    # which has never been a route on either shell —
                    # this to-do 404'd on every tap.
                    deep_link="/m/mail-groups/join-approvals",
                ),
                sort_dt=_naive(req.created_at),
            )
        )
    return out


async def _provider_github_id_approval(
    db: AsyncSession, person_id: UUID, role_names: list[str],
) -> list[_Cand]:
    """等我确认的 GitHub ID 修改单。

    与 ``GET /me/github-id/approvals`` 必须给出同一个集合 —— 两边算法一旦
    分叉,就会出现「待办里有、点进去列表空」。所以 WHERE 条件与
    ``GithubIdChangeService.list_pending_for_approver`` 逐条对齐:
      * 兜底角色(HR / global_admin)→ 全公司在途单,含没有指定审批人的;
      * 普通人 → 只看指派给自己的,且不含自己提的(四眼)。

    不看视角(perspective):被指定为某个人的 HR Ops 是**数据**,不是角色
    —— 一个 global_admin「以员工身份查看」时,他自己名下该批的单子照样
    是他的活。与云订阅 owner 审批同一条规则。
    """
    q = select(GithubIdChangeRequest).where(
        GithubIdChangeRequest.status == "pending",
    )
    if not can_decide_github_id_change(role_names):
        q = q.where(
            GithubIdChangeRequest.approver_person_id == str(person_id),
            GithubIdChangeRequest.person_id != str(person_id),
        )
    rows = (
        await db.execute(
            q.order_by(GithubIdChangeRequest.created_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    names: dict[str, str] = {}
    requester_ids = {r.person_id for r in rows}
    if requester_ids:
        people = (
            await db.execute(
                select(IdentityPerson.id, IdentityPerson.display_name,
                       IdentityPerson.work_email)
                .where(IdentityPerson.id.in_(requester_ids))
            )
        ).all()
        names = {str(pid): (dn or we or str(pid)) for pid, dn, we in people}
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_GITHUB_ID_APPROVAL}:{r.id}",
                type=TYPE_GITHUB_ID_APPROVAL,
                title=names.get(r.person_id, r.person_id),
                subtitle=f"{r.old_github_id or '—'} → {r.new_github_id}",
                severity="info",
                created_at=r.created_at.isoformat(),
                deep_link="/m/github-id/approvals",
            ),
            sort_dt=_naive(r.created_at),
        )
        for r in rows
    ]


async def _provider_cloud_scope_approval(
    db: AsyncSession, _person_id: UUID,
) -> list[_Cand]:
    """Pending cloud SCOPE requests awaiting cloud-admin approval.

    Gated on ``can_manage_cloud`` upstream (scope requests are approved by
    the cloud-admin team, not per-scope owners), so this lists the whole
    pending queue. Decidable in-app — the Feishu webview's JWT passes the
    ``require_browser_session`` step-up (it distinguishes PAT from JWT, not
    browser from webview) — so it deep-links to the scope-request detail
    page that now hosts approve/reject inline.
    """
    rows = (
        await db.execute(
            select(CmdbCloudScopeRequest)
            .where(
                CmdbCloudScopeRequest.status.in_(
                    [SCOPE_REQUEST_STATUS_PENDING, SCOPE_REQUEST_STATUS_FEISHU_SENT],
                )
            )
            .order_by(CmdbCloudScopeRequest.created_at.desc())
            .limit(_PER_SOURCE_CAP)
        )
    ).scalars().all()
    return [
        _Cand(
            item=MobileActionItem(
                id=f"{TYPE_CLOUD_SCOPE_APPROVAL}:{r.id}",
                type=TYPE_CLOUD_SCOPE_APPROVAL,
                title=r.name or r.provider,
                subtitle=r.provider,
                severity="warning",
                created_at=r.created_at.isoformat(),
                deep_link=f"/m/cloud/scope-requests/{r.id}",
            ),
            sort_dt=_naive(r.created_at),
        )
        for r in rows
    ]


async def build_action_inbox(
    db: AsyncSession,
    *,
    person_id: UUID,
    role_names: list[str],
    types: list[str] | None = None,
    page: int = 1,
    page_size: int = 20,
) -> MobileActionInboxResponse:
    """Assemble the caller's prioritised cross-domain action inbox.

    ``role_names`` must already reflect any ``X-View-Role`` narrowing
    (``get_current_user`` does this). ``types``, when given, filters the
    surfaced item types (comma-split upstream). Items are sorted by
    severity then newest-first; ``counts`` and ``total`` are computed off
    the full candidate set before pagination.
    """
    # Self-scope providers run for every caller and every perspective —
    # these are personal obligations (my transfers, my tasks, my
    # attestation, my ticket replies), never gated by role.
    providers: list[Callable[[AsyncSession, UUID], Awaitable[list[_Cand]]]] = [
        _provider_asset_transfer,
        _provider_provisioning,
        _provider_close_tasks,
        _provider_access_attestation,
        _provider_support_reply,
        _provider_attendance_confirm,
        _provider_inventory_response,
        _provider_onboarding_ack,
        # Cloud grant approval runs for every caller: being a scope owner
        # is a data-driven duty, not a role. The provider itself widens to
        # the org-wide queue only when the (perspective-narrowed) role set
        # grants cloud-management authority.
        lambda db_, pid: _provider_cloud_grant_approval(db_, pid, role_names),
        # Renewal approval — same data-driven duty as grant approval.
        # Added 2026-07-29 with the renewal approval flow: the approver's
        # card arrives in Feishu (on a phone), so it needs an inbox item
        # and a mobile decision page or the card is a dead end.
        lambda db_, pid: _provider_cloud_renewal_approval(
            db_, pid, role_names,
        ),
        # Mail-group join-request approval — same data-driven-duty shape:
        # every caller who owns a group sees its pending join requests;
        # HR / IT Admin see the org-wide queue.
        lambda db_, pid: _provider_email_group_join_approval(db_, pid, role_names),
        # GitHub ID 修改确认 —— 同样是数据驱动的责任(申请人的 HR Ops /
        # HRBP),所以不挂角色闸;HR / global_admin 在 provider 内部才拓宽
        # 到兜底队列。
        lambda db_, pid: _provider_github_id_approval(db_, pid, role_names),
    ]
    # Security findings are an AUDITOR-perspective concern, not a general
    # admin one. Gating on the auditor role (rather than can_manage_security,
    # which also returns True for global_admin) keeps the global_admin
    # operational action centre focused on operational to-dos — findings
    # only surface when the user is in the auditor perspective. Combined
    # with the as_role narrowing upstream, the employee perspective never
    # sees findings at all.
    if has_role(role_names, AUDITOR_ROLE):
        providers.append(_provider_findings)
    # IT workbench: tickets assigned to me. Gated on global_admin so the
    # employee/other perspectives never see IT-workbench work; scoped to
    # my own assignments inside the provider.
    if has_role(role_names, GLOBAL_ADMIN_ROLE):
        providers.append(_provider_support_assigned)
    if can_manage_leaver_blocked(role_names):
        providers.append(_provider_leaver_blockers)
    # Leaver dual-sign secondary sign-off: baseline global_admin / hr,
    # further narrowed inside the provider to the NAMED secondary signer.
    if can_secondary_sign_leaver_override(role_names):
        providers.append(_provider_leaver_override)
    # Cloud scope-request approval is a cloud-admin duty (not per-scope
    # owner), so it only surfaces to the cloud-management role set.
    if can_manage_cloud(role_names):
        providers.append(_provider_cloud_scope_approval)

    candidates: list[_Cand] = []
    for provider in providers:
        candidates.extend(await provider(db, person_id))

    if types:
        wanted = set(types)
        candidates = [c for c in candidates if c.item.type in wanted]

    counts: dict[str, int] = {}
    for c in candidates:
        counts[c.item.type] = counts.get(c.item.type, 0) + 1
    total = len(candidates)

    # Stable two-pass sort: newest-first within a severity band, urgent
    # bands first. Python's sort is stable, so sorting by recency then by
    # severity rank yields (severity asc, created_at desc).
    candidates.sort(key=lambda c: c.sort_dt, reverse=True)
    candidates.sort(key=lambda c: _SEV_RANK.get(c.item.severity, 3))

    start = (page - 1) * page_size
    page_items = [c.item for c in candidates[start : start + page_size]]

    return MobileActionInboxResponse(
        items=page_items,
        counts=counts,
        total=total,
        page=page,
        page_size=page_size,
    )