"""Permission-catalog in-app approval + validity (migration 606).

Before this module ``requires_approval=True`` was a dead end: the request was
refused with "contact IT" because nothing in Helix could route or decide it,
and ``PERM_EXPIRED`` was a status no code path ever set. The 2026-09-05
decision — **approvals happen in Helix, not Feishu; the software owner / IT
contact (or IT) decides with one click** — is implemented here and in
:class:`~app.domains.permission.service.PermissionService`:

* who decides: ``AssetSoftwareCatalog.owner_person_id`` and
  ``it_contact_person_id``; when the software names nobody, global admins are
  the approvers and the card is routed to them;
* reminders: ``permission.approval.reminder_days`` after submission the
  approver is nudged; past ``escalate_to_it_after_days`` IT is copied in;
* validity: approvers / admins may give a grant ``valid_days``; the daily
  sweep issues the REVOKE provisioning task when ``expires_at`` passes and the
  task's success hook lands the row on ``expired``. Whether the HOLDER hears
  about the expiry is a gated employee push (default off — IT digest first).
"""
from __future__ import annotations

import logging
import uuid
from dataclasses import asdict, dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthRole, AuthUserRole
from app.core.exceptions import AppException
from app.core.txn import best_effort
from app.domains.asset.models import AssetSoftwareCatalog
from app.domains.permission.models import (
    PERM_ACTIVE,
    PERM_EXPIRED,
    PERM_PENDING_APPROVAL,
    SoftwarePermission,
    UserPermission,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import PlatformSetting
from app.integrations.service import get_platform_setting

logger = logging.getLogger(__name__)

SETTING_KEY = "permission.approval"

TEMPLATE_APPROVAL_REQUESTED = "permission.approval_requested"
TEMPLATE_APPROVAL_REMINDER = "permission.approval_reminder"
TEMPLATE_APPROVAL_ESCALATED = "permission.approval_escalated"
TEMPLATE_APPROVAL_DECIDED = "permission.approval_decided"
TEMPLATE_EXPIRED = "permission.expired"

APPROVALS_URL = "/my/todos?tab=permissionApprovals"
ADMIN_URL = "/admin/permissions?tab=grants&status=pending_approval"
MY_URL = "/me"

_REMIND_FLOOR = timedelta(hours=20)


@dataclass(frozen=True, slots=True)
class ApprovalSettings:
    reminder_days: tuple[int, ...] = (3, 7)
    escalate_to_it_after_days: int = 7
    default_validity_days: int | None = None
    notify_holder_on_expiry: bool = False

    @classmethod
    def from_setting(cls, raw: Any) -> ApprovalSettings:
        base = cls()
        if not isinstance(raw, dict):
            return base
        ladder = base.reminder_days
        raw_ladder = raw.get("reminder_days")
        if isinstance(raw_ladder, list) and raw_ladder:
            try:
                ladder = tuple(sorted({int(d) for d in raw_ladder if int(d) > 0}))
            except (TypeError, ValueError):
                ladder = base.reminder_days
            if not ladder:
                ladder = base.reminder_days
        validity = raw.get("default_validity_days")
        validity_int: int | None
        try:
            validity_int = (
                int(str(validity)) if validity is not None and str(validity) not in ("", "0")
                else None
            )
        except (TypeError, ValueError):
            validity_int = None
        try:
            escalate = int(raw.get("escalate_to_it_after_days", base.escalate_to_it_after_days))
        except (TypeError, ValueError):
            escalate = base.escalate_to_it_after_days
        return cls(
            reminder_days=ladder,
            escalate_to_it_after_days=escalate,
            default_validity_days=validity_int,
            notify_holder_on_expiry=bool(raw.get("notify_holder_on_expiry", False)),
        )

    def to_setting(self) -> dict[str, Any]:
        data = asdict(self)
        data["reminder_days"] = list(self.reminder_days)
        return data

    def validate(self) -> None:
        problems: list[str] = []
        if not self.reminder_days or any(d <= 0 for d in self.reminder_days):
            problems.append("reminder_days must be positive day counts")
        if not 1 <= self.escalate_to_it_after_days <= 60:
            problems.append("escalate_to_it_after_days must be between 1 and 60")
        if self.default_validity_days is not None and not 7 <= self.default_validity_days <= 1095:
            problems.append("default_validity_days must be between 7 and 1095 (or empty)")
        if problems:
            raise AppException(
                code="PERMISSION_APPROVAL_SETTINGS_INVALID",
                message="; ".join(problems),
                status_code=400,
            )


async def load_approval_settings(db: AsyncSession) -> ApprovalSettings:
    raw = await get_platform_setting(SETTING_KEY, default=None, session=db)
    return ApprovalSettings.from_setting(raw)


async def save_approval_settings(db: AsyncSession, settings: ApprovalSettings) -> ApprovalSettings:
    settings.validate()
    row = await db.get(PlatformSetting, SETTING_KEY)
    if row is None:
        row = PlatformSetting(
            key=SETTING_KEY, value=settings.to_setting(),
            description="权限目录站内审批(催办阶梯 / 升级 IT 天数 / 默认有效期 / 到期是否通知持有人)",
        )
        db.add(row)
    else:
        row.value = settings.to_setting()
    await db.flush()
    return settings


# ─────────────────────────── approvers ───────────────────────────


def _lang(person: IdentityPerson | None) -> str:
    raw = ((getattr(person, "preferred_language", None) or "") if person else "").lower()
    if raw.startswith("en"):
        return "en"
    if raw.startswith("ja"):
        return "ja"
    return "zh"


def _deliverable(person: IdentityPerson | None) -> bool:
    return bool(
        person is not None and person.deleted_at is None
        and (person.employment_status or "").lower() == "active" and person.work_email,
    )


def software_approver_ids(software: AssetSoftwareCatalog | None) -> list[uuid.UUID]:
    """Owner first, then IT contact — deduplicated, NULLs dropped."""
    if software is None:
        return []
    out: list[uuid.UUID] = []
    for raw in (software.owner_person_id, software.it_contact_person_id):
        if raw is None:
            continue
        pid = raw if isinstance(raw, uuid.UUID) else uuid.UUID(str(raw))
        if pid not in out:
            out.append(pid)
    return out


async def global_admin_ids(db: AsyncSession) -> list[uuid.UUID]:
    return list((await db.scalars(
        select(AuthUserRole.user_id)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(AuthRole.name.in_(("global_admin", "it_admin"))),
    )).all())


async def resolve_approvers(
    db: AsyncSession, software: AssetSoftwareCatalog | None,
) -> tuple[list[IdentityPerson], bool]:
    """Active approver persons for ``software`` and whether IT is the fallback.

    Returns ``(persons, routed_to_it)``. ``routed_to_it`` is True when the
    software names no deliverable owner / IT contact, so the request card
    goes to every global admin instead.
    """
    ids = software_approver_ids(software)
    persons: list[IdentityPerson] = []
    for pid in ids:
        p = await db.get(IdentityPerson, pid)
        if _deliverable(p):
            persons.append(p)  # type: ignore[arg-type]
    if persons:
        return persons, False
    admins: list[IdentityPerson] = []
    for aid in await global_admin_ids(db):
        p = await db.get(IdentityPerson, aid)
        if _deliverable(p):
            admins.append(p)  # type: ignore[arg-type]
    return admins, True


# ─────────────────────────── notifications ───────────────────────────


def _label(person: IdentityPerson | None) -> str:
    if person is None:
        return "-"
    return person.display_name or person.work_email or str(person.id)


async def notify_approval_requested(
    db: AsyncSession,
    *,
    up: UserPermission,
    perm: SoftwarePermission,
    software: AssetSoftwareCatalog | None,
    requester: IdentityPerson,
    approvers: list[IdentityPerson],
    routed_to_it: bool,
    trace_id: str,
) -> int:
    from app.notifications.service import (  # noqa: PLC0415
        CardAction,
        CardFact,
        CardFactBox,
        LocalizedText,
        NotificationEnvelope,
        NotificationService,
    )

    svc = NotificationService(db)
    sent = 0
    sw_name = software.name if software else "-"
    reason = (up.reason or "").strip()[:300] or "-"
    url = ADMIN_URL if routed_to_it else APPROVALS_URL
    for approver in approvers:
        lang = _lang(approver)
        title = {
            "zh": f"待你审批:{_label(requester)} 申请 {sw_name} · {perm.display_name}",
            "en": f"Approval needed: {_label(requester)} requests {sw_name} · {perm.display_name}",
            "ja": f"承認待ち:{_label(requester)} が {sw_name} · {perm.display_name} を申請",
        }[lang]
        body = {
            "zh": (
                f"{_label(requester)} 申请软件权限「{sw_name} · {perm.display_name}」,理由:{reason}\n\n"
                + ("该软件没有登记负责人 / IT 联系人,路由给 IT。" if routed_to_it else "你是该软件的负责人 / IT 联系人,")
                + "请在 Helix 里批准或拒绝(不走飞书)。"
            ),
            "en": (
                f"{_label(requester)} requests the software permission \"{sw_name} · {perm.display_name}\". "
                f"Reason: {reason}\n\n"
                + ("The software names no owner / IT contact, so this is routed to IT. " if routed_to_it else "As the software owner / IT contact, ")
                + "approve or reject it in Helix (no Feishu step)."
            ),
            "ja": (
                f"{_label(requester)} がソフトウェア権限「{sw_name} · {perm.display_name}」を申請しました。理由:{reason}\n\n"
                + ("このソフトウェアには責任者 / IT 担当が登録されていないため IT に回されました。" if routed_to_it else "ソフトウェアの責任者 / IT 担当として、")
                + "Helix で承認または却下してください(飛書は経由しません)。"
            ),
        }[lang]
        # SAVEPOINT + ``step.reason()``, replacing ``except Exception`` +
        # ``str(exc)``. ``dispatch`` self-isolates its own writes, so the
        # savepoint is defence in depth and uniformity with the sweep loops
        # below (which wrap calls that genuinely do not isolate). The fixed
        # defect is the error text: ``str(AppException)`` is an empty
        # sentinel, so the warning named no cause at all.
        async with best_effort(
            db, label=f"permission approval card to approver {approver.id}",
            log=logger, log_level=logging.DEBUG,
        ) as step:
            await svc.dispatch(
                NotificationEnvelope(
                    recipient=approver, title=title, body_text=body,
                    channels=["feishu", "email"], template_key=TEMPLATE_APPROVAL_REQUESTED,
                    context={
                        "requester": _label(requester), "software": sw_name,
                        "permission": perm.display_name, "reason": reason,
                    },
                    url=url,
                    card_subtitle=LocalizedText(
                        zh="软件权限 · 站内审批", en="Software permission · in-app approval",
                        ja="ソフトウェア権限 · アプリ内承認",
                    ),
                    # 事实盒:左边是要批的那个权限,右边留空 —— 这张卡没有一个
                    # 「决定答不答应的数字」(不像续期有天数),硬塞一个反而是噪音。
                    card_fact_box=CardFactBox(
                        subject=perm.display_name,
                        detail=sw_name,
                    ),
                    # 原来这两行走的是 ``card_stats``,而 stat 网格是给**数字**的
                    # (浅底格子 + 大号粗体值)。「申请人 = 张三」被渲染成一个大号
                    # 粗体的名字,读起来像一个指标。事实行才是它们的形状。
                    card_facts=[
                        CardFact(
                            label=LocalizedText(zh="申请人", en="Requester", ja="申請者"),
                            value=_label(requester),
                            icon="person",
                        ),
                        CardFact(
                            label=LocalizedText(zh="原因", en="Reason", ja="理由"),
                            value=reason or "—",
                            icon="reason",
                        ),
                    ],
                    card_actions=[
                        CardAction(label=LocalizedText(zh="去审批", en="Decide", ja="承認へ"), url=url, style="primary"),
                    ],
                    related_resource_type="user_permission",
                    related_resource_id=up.id,
                ),
                trace_id=trace_id, actor_user_id=requester.id,
            )
        if step.ok:
            sent += 1
        else:
            logger.warning(
                "permission approval card to %s failed: %s",
                approver.work_email, step.reason(),
            )
    return sent


async def notify_decision(
    db: AsyncSession,
    *,
    up: UserPermission,
    perm: SoftwarePermission,
    software: AssetSoftwareCatalog | None,
    requester: IdentityPerson,
    approver: IdentityPerson | None,
    approved: bool,
    note: str | None,
    trace_id: str,
) -> bool:
    from app.notifications.service import (  # noqa: PLC0415
        CardAction,
        LocalizedText,
        NotificationEnvelope,
        NotificationService,
    )

    if not _deliverable(requester):
        return False
    lang = _lang(requester)
    sw_name = software.name if software else "-"
    decision = {
        "zh": "已批准" if approved else "已拒绝",
        "en": "approved" if approved else "rejected",
        "ja": "承認されました" if approved else "却下されました",
    }[lang]
    note_txt = f" 备注:{note.strip()}" if note and note.strip() else ""
    note_txt_en = f" Note: {note.strip()}" if note and note.strip() else ""
    note_txt_ja = f" 備考:{note.strip()}" if note and note.strip() else ""
    validity = ""
    validity_en = ""
    validity_ja = ""
    if approved and up.expires_at:
        d = up.expires_at.date().isoformat()
        validity = f" 有效期至 {d}。"
        validity_en = f" Valid until {d}."
        validity_ja = f" 有効期限は {d} までです。"
    title = {
        "zh": f"你的权限申请{decision}:{sw_name} · {perm.display_name}",
        "en": f"Your permission request was {decision}: {sw_name} · {perm.display_name}",
        "ja": f"権限申請が{decision}:{sw_name} · {perm.display_name}",
    }[lang]
    body = {
        "zh": f"{_label(approver)} 已{decision}你对「{sw_name} · {perm.display_name}」的申请。{note_txt}{validity}"
              + (" 云侧开通完成后状态会变为「生效」。" if approved else ""),
        "en": f"{_label(approver)} {decision} your request for \"{sw_name} · {perm.display_name}\".{note_txt_en}{validity_en}"
              + (" The status turns Active once the vendor-side grant lands." if approved else ""),
        "ja": f"{_label(approver)} があなたの「{sw_name} · {perm.display_name}」申請を{decision}。{note_txt_ja}{validity_ja}"
              + (" ベンダー側の付与が完了すると「有効」になります。" if approved else ""),
    }[lang]
    # SAVEPOINT — ``dispatch`` writes; see ``notify_approvers`` above.
    async with best_effort(
        db, label="permission decision card", log=logger,
        log_level=logging.DEBUG,
    ) as step:
        await NotificationService(db).dispatch(
            NotificationEnvelope(
                recipient=requester, title=title, body_text=body,
                channels=["feishu", "email"], template_key=TEMPLATE_APPROVAL_DECIDED,
                context={
                    "approver": _label(approver), "decision_label": decision,
                    "software": sw_name, "permission": perm.display_name,
                    "note": {"zh": note_txt, "en": note_txt_en, "ja": note_txt_ja}[lang],
                    "validity": {"zh": validity, "en": validity_en, "ja": validity_ja}[lang],
                },
                url=MY_URL,
                card_actions=[CardAction(
                    label=LocalizedText(zh="查看我的权限", en="View my permissions", ja="自分の権限を見る"),
                    url=MY_URL, style="primary",
                )],
                related_resource_type="user_permission",
                related_resource_id=up.id,
            ),
            trace_id=trace_id, actor_user_id=approver.id if approver else None,
        )
    if step.ok:
        return True
    logger.warning(
        "permission decision card to %s failed: %s",
        requester.work_email, step.reason(),
    )
    return False


async def notify_expired(
    db: AsyncSession,
    *,
    up: UserPermission,
    perm: SoftwarePermission,
    software: AssetSoftwareCatalog | None,
    holder: IdentityPerson,
    trace_id: str,
) -> bool:
    from app.notifications.service import NotificationEnvelope, NotificationService  # noqa: PLC0415

    if not _deliverable(holder):
        return False
    lang = _lang(holder)
    sw_name = software.name if software else "-"
    expires = up.expires_at.date().isoformat() if up.expires_at else "-"
    title = {
        "zh": f"软件权限已到期并回收:{sw_name} · {perm.display_name}",
        "en": f"Software permission expired and revoked: {sw_name} · {perm.display_name}",
        "ja": f"ソフトウェア権限が期限切れで回収されました:{sw_name} · {perm.display_name}",
    }[lang]
    body = {
        "zh": f"你持有的「{sw_name} · {perm.display_name}」有效期已到({expires}),Helix 已发起回收。如仍需要,请重新申请。",
        "en": f"Your \"{sw_name} · {perm.display_name}\" permission reached its validity end ({expires}); Helix has started the revoke. Request it again if you still need it.",
        "ja": f"保持していた「{sw_name} · {perm.display_name}」の有効期限({expires})が終了し、Helix が回収を開始しました。必要であれば再申請してください。",
    }[lang]
    # SAVEPOINT — ``dispatch`` writes; see ``notify_approvers`` above.
    async with best_effort(
        db, label="permission expiry card", log=logger,
        log_level=logging.DEBUG,
    ) as step:
        await NotificationService(db).dispatch(
            NotificationEnvelope(
                recipient=holder, title=title, body_text=body,
                channels=["feishu", "email"], template_key=TEMPLATE_EXPIRED,
                context={"software": sw_name, "permission": perm.display_name, "expires_at": expires},
                url=MY_URL,
                related_resource_type="user_permission",
                related_resource_id=up.id,
            ),
            trace_id=trace_id, actor_user_id=None,
        )
    if step.ok:
        return True
    logger.warning(
        "permission expiry card to %s failed: %s",
        holder.work_email, step.reason(),
    )
    return False


# ─────────────────────────── daily sweep ───────────────────────────


async def sweep_lifecycle(
    db: AsyncSession, *, now: datetime, settings: ApprovalSettings, trace_id: str,
) -> dict[str, Any]:
    """``permission.lifecycle_sweep`` body: approval reminders / escalation and expiry."""
    from app.domains.permission.service import PermissionService  # noqa: PLC0415
    from app.notifications.service import (  # noqa: PLC0415
        CardAction,
        LocalizedText,
        NotificationEnvelope,
        NotificationService,
    )

    summary: dict[str, Any] = {
        "reminders_sent": 0, "escalated": 0, "expired_revoke_tasks": 0,
        "expired_holder_notified": 0, "errors": 0, "expired_items": [],
    }
    notify = NotificationService(db)

    # ── 1. pending approvals ──
    pending = list((await db.execute(
        select(UserPermission, SoftwarePermission, AssetSoftwareCatalog)
        .join(SoftwarePermission, SoftwarePermission.id == UserPermission.permission_id)
        .outerjoin(AssetSoftwareCatalog, AssetSoftwareCatalog.id == SoftwarePermission.software_id)
        .where(UserPermission.status == PERM_PENDING_APPROVAL),
    )).all())
    admins_cache: list[IdentityPerson] | None = None
    for up, perm, software in pending:
        # Read the id BEFORE the savepoint: a rollback expires every object
        # the body dirtied, and an expired attribute read outside an await
        # raises ``MissingGreenlet``.
        up_id = up.id
        # SAVEPOINT, not a bare ``except``: the body dispatches (outbox
        # insert + flush) and writes back the reminder counters. A swallowed
        # flush failure deactivates the transaction, so "one request must not
        # stop the sweep" was untrue — every remaining row then failed with
        # ``PendingRollbackError`` and the trailing ``db.flush()`` took the
        # whole cron down.
        async with best_effort(
            db, label=f"permission approval reminder for {up_id}",
            log=logger, log_level=logging.DEBUG,
        ) as step:
            created = up.created_at if up.created_at.tzinfo else up.created_at.replace(tzinfo=UTC)
            waiting = (now - created).days
            due_stage = sum(1 for d in settings.reminder_days if waiting >= d)
            escalate = waiting >= settings.escalate_to_it_after_days
            stage_done = int(up.approval_reminder_count or 0)
            if due_stage <= stage_done and not (escalate and stage_done < len(settings.reminder_days) + 1):
                continue
            last = up.approval_last_reminded_at
            if last is not None:
                last = last if last.tzinfo else last.replace(tzinfo=UTC)
                if now - last < _REMIND_FLOOR:
                    continue
            requester = await db.get(IdentityPerson, up.user_id)
            approvers, routed_to_it = await resolve_approvers(db, software)
            sw_name = software.name if software else "-"
            recipients: list[tuple[IdentityPerson, bool]] = [(a, False) for a in approvers]
            if escalate and not routed_to_it:
                if admins_cache is None:
                    admins_cache = []
                    for aid in await global_admin_ids(db):
                        p = await db.get(IdentityPerson, aid)
                        if _deliverable(p):
                            admins_cache.append(p)  # type: ignore[arg-type]
                recipients += [(a, True) for a in admins_cache if a.id not in {x.id for x in approvers}]
            for person, is_escalation in recipients:
                lang = _lang(person)
                key = TEMPLATE_APPROVAL_ESCALATED if is_escalation else TEMPLATE_APPROVAL_REMINDER
                url = ADMIN_URL if (is_escalation or routed_to_it) else APPROVALS_URL
                title = {
                    "zh": (f"权限申请超过 {waiting} 天未批,请 IT 接手:" if is_escalation else f"⏳ 权限申请待审批已 {waiting} 天:")
                          + f"{_label(requester)} · {sw_name} · {perm.display_name}",
                    "en": (f"Permission request unanswered for {waiting} days — IT please take over: " if is_escalation else f"⏳ Permission request waiting {waiting} day(s): ")
                          + f"{_label(requester)} · {sw_name} · {perm.display_name}",
                    "ja": (f"権限申請が {waiting} 日間未処理 — IT が引き継いでください:" if is_escalation else f"⏳ 権限申請が {waiting} 日間承認待ち:")
                          + f"{_label(requester)} · {sw_name} · {perm.display_name}",
                }[lang]
                body = {
                    "zh": f"{_label(requester)} 对「{sw_name} · {perm.display_name}」的申请已等待 {waiting} 天。"
                          + ("负责人未答复,IT 可直接在权限管理台批准或拒绝。" if is_escalation else "请在 Helix 待办里处理。"),
                    "en": f"{_label(requester)}'s request for \"{sw_name} · {perm.display_name}\" has waited {waiting} day(s). "
                          + ("The owner has not decided; IT can approve or reject it in the permission console." if is_escalation else "Please decide it in your Helix to-dos."),
                    "ja": f"{_label(requester)} の「{sw_name} · {perm.display_name}」申請は {waiting} 日間承認待ちです。"
                          + ("責任者から回答がないため、IT は権限管理台で直接処理できます。" if is_escalation else "Helix の To-Do で処理してください。"),
                }[lang]
                await notify.dispatch(
                    NotificationEnvelope(
                        recipient=person, title=title, body_text=body,
                        channels=["feishu", "email"], template_key=key,
                        context={
                            "requester": _label(requester), "software": sw_name,
                            "permission": perm.display_name, "days": str(waiting),
                        },
                        url=url,
                        card_actions=[CardAction(
                            label=LocalizedText(zh="去审批", en="Decide", ja="承認へ"), url=url, style="primary",
                        )],
                        related_resource_type="user_permission",
                        related_resource_id=up.id,
                    ),
                    trace_id=trace_id, actor_user_id=None,
                )
                if is_escalation:
                    summary["escalated"] += 1
                else:
                    summary["reminders_sent"] += 1
            up.approval_reminder_count = max(due_stage, stage_done) + (1 if escalate else 0)
            up.approval_last_reminded_at = now
        if not step.ok:
            summary["errors"] += 1
            logger.warning(
                "permission approval reminder for %s failed: %s",
                up_id, step.reason(),
            )
    await db.flush()

    # ── 2. expiry ──
    expired = list((await db.execute(
        select(UserPermission, SoftwarePermission, AssetSoftwareCatalog)
        .join(SoftwarePermission, SoftwarePermission.id == UserPermission.permission_id)
        .outerjoin(AssetSoftwareCatalog, AssetSoftwareCatalog.id == SoftwarePermission.software_id)
        .where(
            UserPermission.status == PERM_ACTIVE,
            UserPermission.expires_at.is_not(None),
            UserPermission.expires_at <= now,
        ),
    )).all())
    svc = PermissionService(db)
    for up, perm, software in expired:
        up_id = up.id
        # SAVEPOINT: ``revoke_user_permission`` queues a provisioning task
        # and flips the row — the single most consequential write in this
        # module. Swallowing its failed flush meant every LATER expired
        # permission in the same tick silently failed to be revoked, which
        # is access staying live past its own expiry date.
        async with best_effort(
            db, label=f"permission expiry revoke for {up_id}",
            log=logger, log_level=logging.DEBUG,
        ) as step:
            if up.provisioning_task_id is not None:
                # A revoke task from a previous tick may still be in flight;
                # do not stack a second one.
                from app.domains.provisioning.models import (  # noqa: PLC0415
                    ACTION_REVOKE,
                    ProvisioningTask,
                )
                task = await db.get(ProvisioningTask, up.provisioning_task_id)
                if (
                    task is not None and task.action == ACTION_REVOKE
                    and task.status not in ("completed", "failed", "cancelled")
                ):
                    continue
            await svc.revoke_user_permission(
                up.id, reason="expired", trace_id=trace_id, actor_user_id=None,
                terminal_status=PERM_EXPIRED,
            )
            summary["expired_revoke_tasks"] += 1
            holder = await db.get(IdentityPerson, up.user_id)
            summary["expired_items"].append({
                "user_permission_id": str(up.id),
                "holder": _label(holder),
                "software": software.name if software else "-",
                "permission": perm.display_name,
                "expires_at": up.expires_at.isoformat() if up.expires_at else None,
            })
            if settings.notify_holder_on_expiry and holder is not None:
                if await notify_expired(db, up=up, perm=perm, software=software, holder=holder, trace_id=trace_id):
                    summary["expired_holder_notified"] += 1
        if not step.ok:
            summary["errors"] += 1
            logger.warning(
                "permission expiry for %s failed: %s", up_id, step.reason(),
            )
    await db.flush()
    return summary


__all__ = [
    "ADMIN_URL",
    "APPROVALS_URL",
    "SETTING_KEY",
    "ApprovalSettings",
    "global_admin_ids",
    "load_approval_settings",
    "notify_approval_requested",
    "notify_decision",
    "notify_expired",
    "resolve_approvers",
    "save_approval_settings",
    "software_approver_ids",
    "sweep_lifecycle",
]