"""DAX 授权生命周期(migration 606):授予之后,这条权限什么时候、由谁、凭什么没了。

Two facts the binding table could not answer before this module:

* **When does a personal grant end?** Nothing — a ``google_user`` binding lived
  until the holder left or someone remembered. Now every personal grant carries
  ``expires_at`` (default 180 days, ``data_access.grant_lifecycle.personal_ttl_days``),
  the holder is reminded on a ladder (T-14 / T-7 / T-1), may self-renew a bounded
  number of times, and at expiry the grant is revoked through the SAME path as any
  other revoke (``revoke_data_access_binding`` → apply-absent reconciler — BigQuery
  is still written by exactly one place).
* **Who still vouches for a service-account grant?** Service accounts do not
  expire (a workload does not re-apply), but their ``technical_owner_person_id``
  re-confirms every ``sa_confirmation_interval_days`` (182) that the workload
  still needs the dataset. Overdue past ``sa_confirmation_grace_days`` the grant
  is revoked — again through the ordinary revoke path.

Employee-facing pushes are gated (2026-09-05 decision: ship the class complete,
deliver IT digest first, open per class): ``employee_push_enabled`` controls the
holder reminders, ``owner_confirmation_push_enabled`` controls the owner cards,
and the two ``auto_revoke_*`` switches cannot be turned on while the matching
push is off — a grant must never disappear without the holder having been told.
Until a switch is on, the daily IT digest carries the same rows.

Every transition writes a :class:`DataAccessLifecycleEvent`; the admin console
renders them as the binding's timeline.
"""
from __future__ import annotations

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

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

from app.audit.service import AuditService
from app.auth.models import AuthRole, AuthUserRole
from app.core.exceptions import AppException
from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role, normalize_role_names
from app.core.setting_values import as_bool
from app.core.txn import best_effort
from app.domains.data_access import constants as c
from app.domains.data_access.models import (
    BINDING_STATUS_ACTIVE,
    BINDING_STATUS_REVOKED,
    BINDING_STATUS_REVOKED_PENDING,
    LIFECYCLE_EVENT_EXPIRED_OVERDUE,
    LIFECYCLE_EVENT_EXPIRED_REVOKE,
    LIFECYCLE_EVENT_EXPIRY_REMINDER,
    LIFECYCLE_EVENT_EXPIRY_STAMPED,
    LIFECYCLE_EVENT_EXTENDED_BY_ADMIN,
    LIFECYCLE_EVENT_OWNER_CONFIRMATION_LAPSED_REVOKE,
    LIFECYCLE_EVENT_OWNER_CONFIRMATION_REMINDER,
    LIFECYCLE_EVENT_OWNER_CONFIRMATION_REQUESTED,
    LIFECYCLE_EVENT_OWNER_CONFIRMED,
    LIFECYCLE_EVENT_OWNER_DECLINED,
    LIFECYCLE_EVENT_RELEASED_BY_HOLDER,
    LIFECYCLE_EVENT_RENEWED,
    LIFECYCLE_EVENT_REVOKED_BY_ADMIN,
    LIFECYCLE_EVENT_REVOKED_OFFBOARDING,
    LIFECYCLE_EVENT_REVOKED_SELF_ATTESTATION,
    REVOKE_REASON_ADMIN,
    REVOKE_REASON_EXPIRED,
    REVOKE_REASON_HOLDER_RELEASED,
    REVOKE_REASON_OFFBOARDING,
    REVOKE_REASON_OWNER_CONFIRMATION_LAPSED,
    REVOKE_REASON_OWNER_DECLINED,
    REVOKE_REASON_SELF_ATTESTATION,
    DataAccessBinding,
    DataAccessLifecycleEvent,
    DataAccessResource,
    DataAccessServiceAccountProfile,
)
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 = "data_access.grant_lifecycle"

TEMPLATE_EXPIRY_REMINDER = "data_access.expiry_reminder"
TEMPLATE_EXPIRED_REVOKED = "data_access.expired_revoked"
TEMPLATE_OWNER_CONFIRMATION = "data_access.owner_confirmation"
TEMPLATE_OWNER_LAPSED = "data_access.owner_confirmation_lapsed"
TEMPLATE_IT_DIGEST = "data_access.lifecycle_it_digest"
DIGEST_RESOURCE_TYPE = "data_access_lifecycle_digest"

# Lifecycle states projected onto a binding for the UI / filters.
STATE_OK = "ok"
STATE_EXPIRING = "expiring"
STATE_EXPIRED = "expired"
STATE_CONFIRMATION_DUE = "confirmation_due"
STATE_CONFIRMATION_OVERDUE = "confirmation_overdue"
LIFECYCLE_STATES = (
    STATE_OK, STATE_EXPIRING, STATE_EXPIRED,
    STATE_CONFIRMATION_DUE, STATE_CONFIRMATION_OVERDUE,
)

_REMIND_FLOOR = timedelta(hours=20)
_OWNER_REMINDER_EVERY = timedelta(days=7)
_CONFIRMATION_DUE_WINDOW = timedelta(days=14)
_DIGEST_MAX_ROWS = 12
_BACKFILL_GRACE = timedelta(days=30)
MY_GRANTS_URL = "/my/data-access/grants"
ADMIN_GRANTS_URL = "/admin/data-warehouse/grants"


# ─────────────────────────── settings ───────────────────────────


@dataclass(frozen=True, slots=True)
class LifecycleSettings:
    personal_ttl_days: int = 180
    reminder_ladder_days: tuple[int, ...] = (14, 7, 1)
    renewal_extends_days: int = 180
    self_renewals_before_reapproval: int = 2
    employee_push_enabled: bool = False
    auto_revoke_on_expiry: bool = False
    sa_confirmation_interval_days: int = 182
    sa_confirmation_grace_days: int = 30
    owner_confirmation_push_enabled: bool = False
    auto_revoke_on_confirmation_lapse: bool = False

    @classmethod
    def from_setting(cls, raw: Any) -> LifecycleSettings:
        base = cls()
        if not isinstance(raw, dict):
            return base
        ladder_raw = raw.get("reminder_ladder_days")
        ladder = base.reminder_ladder_days
        if isinstance(ladder_raw, list) and ladder_raw:
            try:
                ladder = tuple(sorted({int(d) for d in ladder_raw if int(d) > 0}, reverse=True))
            except (TypeError, ValueError):
                ladder = base.reminder_ladder_days
            if not ladder:
                ladder = base.reminder_ladder_days

        def _int(key: str, default: int) -> int:
            try:
                return int(raw.get(key, default))
            except (TypeError, ValueError):
                return default

        return cls(
            personal_ttl_days=_int("personal_ttl_days", base.personal_ttl_days),
            reminder_ladder_days=ladder,
            renewal_extends_days=_int("renewal_extends_days", base.renewal_extends_days),
            self_renewals_before_reapproval=_int(
                "self_renewals_before_reapproval", base.self_renewals_before_reapproval,
            ),
            employee_push_enabled=as_bool(raw.get("employee_push_enabled", False)),
            auto_revoke_on_expiry=as_bool(raw.get("auto_revoke_on_expiry", False)),
            sa_confirmation_interval_days=_int(
                "sa_confirmation_interval_days", base.sa_confirmation_interval_days,
            ),
            sa_confirmation_grace_days=_int(
                "sa_confirmation_grace_days", base.sa_confirmation_grace_days,
            ),
            owner_confirmation_push_enabled=as_bool(
                raw.get("owner_confirmation_push_enabled", False),
            ),
            auto_revoke_on_confirmation_lapse=as_bool(
                raw.get("auto_revoke_on_confirmation_lapse", False),
            ),
        )

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

    def validate(self) -> None:
        """Refuse a configuration that could take access away silently."""
        problems: list[str] = []
        if not 30 <= self.personal_ttl_days <= 730:
            problems.append("personal_ttl_days must be between 30 and 730")
        if not 7 <= self.renewal_extends_days <= 730:
            problems.append("renewal_extends_days must be between 7 and 730")
        if not 0 <= self.self_renewals_before_reapproval <= 12:
            problems.append("self_renewals_before_reapproval must be between 0 and 12")
        if not self.reminder_ladder_days or any(d <= 0 for d in self.reminder_ladder_days):
            problems.append("reminder_ladder_days must be positive day counts")
        if self.reminder_ladder_days and max(self.reminder_ladder_days) >= self.personal_ttl_days:
            problems.append("the first reminder must come after the grant, not before")
        if not 30 <= self.sa_confirmation_interval_days <= 730:
            problems.append("sa_confirmation_interval_days must be between 30 and 730")
        if not 0 <= self.sa_confirmation_grace_days <= 180:
            problems.append("sa_confirmation_grace_days must be between 0 and 180")
        if self.auto_revoke_on_expiry and not self.employee_push_enabled:
            problems.append(
                "auto_revoke_on_expiry requires employee_push_enabled — a grant must not "
                "expire without the holder having been reminded",
            )
        if self.auto_revoke_on_confirmation_lapse and not self.owner_confirmation_push_enabled:
            problems.append(
                "auto_revoke_on_confirmation_lapse requires owner_confirmation_push_enabled",
            )
        if problems:
            raise AppException(
                code="DATA_ACCESS_LIFECYCLE_SETTINGS_INVALID",
                message="; ".join(problems),
                status_code=400,
            )


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


async def save_lifecycle_settings(
    db: AsyncSession, settings: LifecycleSettings,
) -> LifecycleSettings:
    settings.validate()
    row = await db.get(PlatformSetting, SETTING_KEY)
    if row is None:
        row = PlatformSetting(
            key=SETTING_KEY, value=settings.to_setting(),
            description="DAX 授权生命周期(个人 TTL / 提醒阶梯 / 续期 / 服务账号负责人确认 / 推送与自动回收开关)",
        )
        db.add(row)
    else:
        row.value = settings.to_setting()
    await db.flush()
    return settings


# ─────────────────────────── projections ───────────────────────────


def _aware(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    return value if value.tzinfo is not None else value.replace(tzinfo=UTC)


def lifecycle_state(
    binding: DataAccessBinding, *, now: datetime, settings: LifecycleSettings,
) -> str | None:
    """Project the lifecycle position of one binding (``None`` for groups)."""
    if binding.status != BINDING_STATUS_ACTIVE:
        return None
    if binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER:
        expires_at = _aware(binding.expires_at)
        if expires_at is None:
            return STATE_OK
        if expires_at <= now:
            return STATE_EXPIRED
        first = settings.reminder_ladder_days[0] if settings.reminder_ladder_days else 14
        if expires_at - now <= timedelta(days=first):
            return STATE_EXPIRING
        return STATE_OK
    if binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT:
        due = _aware(binding.owner_confirmation_due_at)
        if due is None:
            return STATE_OK
        if due <= now:
            return STATE_CONFIRMATION_OVERDUE
        if due - now <= _CONFIRMATION_DUE_WINDOW:
            return STATE_CONFIRMATION_DUE
        return STATE_OK
    return None


def lifecycle_fields(
    binding: DataAccessBinding, *, now: datetime, settings: LifecycleSettings,
) -> dict[str, Any]:
    """The lifecycle slice of a binding as the API exposes it."""
    expires_at = _aware(binding.expires_at)
    due_at = _aware(binding.owner_confirmation_due_at)
    is_personal = binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER
    renewals_remaining = (
        max(settings.self_renewals_before_reapproval - int(binding.renewal_count or 0), 0)
        if is_personal else None
    )
    return {
        "lifecycle_state": lifecycle_state(binding, now=now, settings=settings),
        "expires_at": expires_at.isoformat() if expires_at else None,
        "days_to_expiry": (
            (expires_at - now).days if expires_at is not None else None
        ),
        "renewal_count": int(binding.renewal_count or 0),
        "renewals_remaining": renewals_remaining,
        "renewable": bool(
            is_personal and binding.status == BINDING_STATUS_ACTIVE
            and renewals_remaining is not None and renewals_remaining > 0,
        ),
        "last_renewed_at": (
            _aware(binding.last_renewed_at).isoformat()  # type: ignore[union-attr]
            if binding.last_renewed_at else None
        ),
        "owner_confirmation_due_at": due_at.isoformat() if due_at else None,
        "owner_confirmed_at": (
            _aware(binding.owner_confirmed_at).isoformat()  # type: ignore[union-attr]
            if binding.owner_confirmed_at else None
        ),
        "owner_confirmation_overdue_days": (
            max((now - due_at).days, 0) if due_at is not None and due_at <= now else None
        ),
    }


def stamp_new_binding(
    binding: DataAccessBinding, *, settings: LifecycleSettings, now: datetime,
) -> None:
    """Give a freshly created binding its lifecycle clock (call before flush)."""
    if binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER:
        if binding.expires_at is None:
            binding.expires_at = now + timedelta(days=settings.personal_ttl_days)
    elif binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT:
        if binding.owner_confirmation_due_at is None:
            binding.owner_confirmation_due_at = now + timedelta(
                days=settings.sa_confirmation_interval_days,
            )


def record_event(
    db: AsyncSession,
    binding: DataAccessBinding,
    kind: str,
    *,
    actor_type: str = "system",
    actor_person_id: uuid.UUID | str | None = None,
    detail: dict[str, Any] | None = None,
) -> DataAccessLifecycleEvent:
    row = DataAccessLifecycleEvent(
        binding_id=binding.id,
        event_kind=kind,
        actor_type=actor_type,
        actor_person_id=str(actor_person_id) if actor_person_id else None,
        detail=detail or {},
    )
    db.add(row)
    return row


def _locator(resource: DataAccessResource | None) -> str:
    if resource is None:
        return "-"
    return ".".join(filter(None, (resource.project_id, resource.dataset_id, resource.table_id)))


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,
    )


async def select_personal_active_bindings(
    db: AsyncSession, person_id: uuid.UUID | str,
) -> list[DataAccessBinding]:
    """Personal (google_user) bindings a person currently holds.

    The one predicate every consumer (offboarding scan, mover holdings, T+1
    post-leaver check, self-attestation snapshot) must share — a binding is
    the person's when its principal snapshot IS the person id.
    """
    return list((await db.scalars(
        select(DataAccessBinding).where(
            DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER,
            DataAccessBinding.principal_subject_snapshot == str(person_id),
            DataAccessBinding.status == BINDING_STATUS_ACTIVE,
        ).order_by(DataAccessBinding.granted_at.desc()),
    )).all())


# ─────────────────────────── service ───────────────────────────


class DataAccessLifecycleService:
    """Owns expiry / renewal / owner-confirmation on :class:`DataAccessBinding`."""

    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.audit = AuditService(db)

    # ── helpers ──

    async def _binding(self, binding_id: uuid.UUID, *, lock: bool = True) -> DataAccessBinding:
        stmt = select(DataAccessBinding).where(DataAccessBinding.id == binding_id)
        if lock:
            stmt = stmt.with_for_update()
        row = (await self.db.scalars(stmt)).one_or_none()
        if row is None:
            raise AppException(
                code="DATA_ACCESS_BINDING_NOT_FOUND",
                message="Data-access binding was not found.",
                status_code=404,
                field="binding_id",
            )
        return row

    async def _role_names(self, actor: IdentityPerson) -> list[str]:
        override = actor.__dict__.get("_role_names_override")
        if override is not None:
            return list(override)
        rows = (await self.db.scalars(
            select(AuthRole.name)
            .join(AuthUserRole, AuthUserRole.role_id == AuthRole.id)
            .where(AuthUserRole.user_id == actor.id),
        )).all()
        return normalize_role_names(rows)

    async def _is_admin(self, actor: IdentityPerson) -> bool:
        return has_role(await self._role_names(actor), GLOBAL_ADMIN_ROLE)

    @staticmethod
    def _is_holder(binding: DataAccessBinding, actor: IdentityPerson) -> bool:
        return (
            binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER
            and binding.principal_subject_snapshot == str(actor.id)
        )

    async def _is_technical_owner(
        self, binding: DataAccessBinding, actor: IdentityPerson,
    ) -> bool:
        if (
            binding.principal_type_snapshot != c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT
            or binding.principal_id is None
        ):
            return False
        hit = await self.db.scalar(
            select(DataAccessServiceAccountProfile.principal_id).where(
                DataAccessServiceAccountProfile.principal_id == binding.principal_id,
                DataAccessServiceAccountProfile.technical_owner_person_id == actor.id,
            ).limit(1),
        )
        return hit is not None

    async def _technical_owner_id(self, binding: DataAccessBinding) -> uuid.UUID | None:
        if (
            binding.principal_type_snapshot != c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT
            or binding.principal_id is None
        ):
            return None
        return await self.db.scalar(
            select(DataAccessServiceAccountProfile.technical_owner_person_id).where(
                DataAccessServiceAccountProfile.principal_id == binding.principal_id,
            ).limit(1),
        )

    async def _holder(self, binding: DataAccessBinding) -> IdentityPerson | None:
        if binding.principal_type_snapshot != c.PRINCIPAL_TYPE_GOOGLE_USER:
            return None
        try:
            pid = uuid.UUID(str(binding.principal_subject_snapshot))
        except (TypeError, ValueError):
            return None
        return await self.db.get(IdentityPerson, pid)

    @staticmethod
    def _require_personal_active(binding: DataAccessBinding) -> None:
        if binding.principal_type_snapshot != c.PRINCIPAL_TYPE_GOOGLE_USER:
            raise AppException(
                code="DATA_ACCESS_NOT_PERSONAL_GRANT",
                message="Only personal (google_user) grants carry an expiry.",
                status_code=409,
                field="binding_id",
            )
        if binding.status != BINDING_STATUS_ACTIVE:
            raise AppException(
                code="DATA_ACCESS_BINDING_NOT_ACTIVE",
                message=f"Binding is {binding.status}; only active grants can be renewed.",
                status_code=409,
                field="binding_id",
            )

    # ── holder / admin verbs ──

    async def renew(
        self,
        binding_id: uuid.UUID,
        *,
        actor: IdentityPerson,
        reason: str,
        settings: LifecycleSettings,
        trace_id: str,
        now: datetime | None = None,
    ) -> DataAccessBinding:
        """Holder (or admin on their behalf) extends a personal grant.

        Bounded: after ``self_renewals_before_reapproval`` self-renewals the
        holder must submit a fresh Feishu request — the approver sees the
        access again instead of it rolling forever on the holder's say-so.
        """
        now = now or datetime.now(UTC)
        binding = await self._binding(binding_id)
        self._require_personal_active(binding)
        is_admin = await self._is_admin(actor)
        if not (is_admin or self._is_holder(binding, actor)):
            raise AppException(
                code="DATA_ACCESS_RENEW_NOT_AUTHORIZED",
                message="Only the grant holder or a global admin may renew this grant.",
                status_code=403,
                field="binding_id",
            )
        clean_reason = (reason or "").strip()
        if not clean_reason:
            raise AppException(
                code="DATA_ACCESS_RENEW_REASON_REQUIRED",
                message="A one-line reason is required to renew a data-access grant.",
                status_code=422,
                field="reason",
            )
        if (
            not is_admin
            and int(binding.renewal_count or 0) >= settings.self_renewals_before_reapproval
        ):
            raise AppException(
                code="DATA_ACCESS_RENEWAL_LIMIT_REACHED",
                message=(
                    "This grant has been self-renewed the maximum number of times; "
                    "submit a new data-access request so the dataset owner reviews it."
                ),
                status_code=409,
                field="binding_id",
            )
        previous = _aware(binding.expires_at)
        base = max(now, previous) if previous is not None else now
        binding.expires_at = base + timedelta(days=settings.renewal_extends_days)
        binding.expiry_reminder_stage = 0
        binding.expiry_last_reminded_at = None
        binding.renewal_count = int(binding.renewal_count or 0) + 1
        binding.last_renewed_at = now
        binding.last_renewed_by_person_id = str(actor.id)
        record_event(
            self.db, binding, LIFECYCLE_EVENT_RENEWED,
            actor_type="user", actor_person_id=actor.id,
            detail={
                "previous_expires_at": previous.isoformat() if previous else None,
                "new_expires_at": binding.expires_at.isoformat(),
                "renewal_count": binding.renewal_count,
                "reason": clean_reason[:500],
                "by_admin": is_admin and not self._is_holder(binding, actor),
            },
        )
        await self.audit.log(
            trace_id=trace_id, actor_type="user", actor_user_id=actor.id,
            action="data_access.binding.renewed", status="success",
            resource_type="data_access_binding", resource_id=str(binding.id),
            detail={
                "new_expires_at": binding.expires_at.isoformat(),
                "renewal_count": binding.renewal_count,
                "reason": clean_reason[:500],
            },
        )
        await self.db.flush()
        return binding

    async def extend(
        self,
        binding_id: uuid.UUID,
        *,
        actor: IdentityPerson,
        days: int,
        reason: str,
        trace_id: str,
        now: datetime | None = None,
    ) -> DataAccessBinding:
        """Admin extends a personal grant by ``days`` (does not consume a renewal)."""
        now = now or datetime.now(UTC)
        if not await self._is_admin(actor):
            raise AppException(
                code="DATA_ACCESS_ADMIN_REQUIRED",
                message="Only a global admin may extend a data-access grant.",
                status_code=403,
            )
        if not 1 <= int(days) <= 730:
            raise AppException(
                code="DATA_ACCESS_EXTEND_DAYS_INVALID",
                message="days must be between 1 and 730.",
                status_code=422,
                field="days",
            )
        binding = await self._binding(binding_id)
        self._require_personal_active(binding)
        previous = _aware(binding.expires_at)
        base = max(now, previous) if previous is not None else now
        binding.expires_at = base + timedelta(days=int(days))
        binding.expiry_reminder_stage = 0
        binding.expiry_last_reminded_at = None
        record_event(
            self.db, binding, LIFECYCLE_EVENT_EXTENDED_BY_ADMIN,
            actor_type="user", actor_person_id=actor.id,
            detail={
                "previous_expires_at": previous.isoformat() if previous else None,
                "new_expires_at": binding.expires_at.isoformat(),
                "days": int(days),
                "reason": (reason or "").strip()[:500],
            },
        )
        await self.audit.log(
            trace_id=trace_id, actor_type="user", actor_user_id=actor.id,
            action="data_access.binding.extended", status="success",
            resource_type="data_access_binding", resource_id=str(binding.id),
            detail={"days": int(days), "new_expires_at": binding.expires_at.isoformat()},
        )
        await self.db.flush()
        return binding

    async def release(
        self,
        binding_id: uuid.UUID,
        *,
        actor: IdentityPerson,
        reason_note: str | None,
        trace_id: str,
        revoke_reason: str = REVOKE_REASON_HOLDER_RELEASED,
        event_kind: str = LIFECYCLE_EVENT_RELEASED_BY_HOLDER,
    ) -> DataAccessBinding:
        """Holder / technical owner / admin gives the grant up.

        Authorisation lives in ``DataAccessService.revoke_data_access_binding``
        (holder, technical owner, group owner, dataset owner, global admin); this
        wrapper only adds the lifecycle event.
        """
        from app.domains.data_access.service import DataAccessService  # noqa: PLC0415

        binding = await self._binding(binding_id, lock=False)
        result = await DataAccessService().revoke_data_access_binding(
            self.db, actor=actor, binding_id=binding_id,
            reason=revoke_reason, trace_id=trace_id,
        )
        if result.revoked_pending_count:
            record_event(
                self.db, binding, event_kind,
                actor_type="user", actor_person_id=actor.id,
                detail={
                    "revoke_reason": revoke_reason,
                    "authorized_as": result.authorized_as,
                    "note": (reason_note or "").strip()[:500] or None,
                },
            )
            await self.db.flush()
        await self.db.refresh(binding)
        return binding

    async def revoke_as_admin(
        self,
        binding_id: uuid.UUID,
        *,
        actor: IdentityPerson,
        reason_note: str | None,
        trace_id: str,
    ) -> DataAccessBinding:
        return await self.release(
            binding_id, actor=actor, reason_note=reason_note, trace_id=trace_id,
            revoke_reason=REVOKE_REASON_ADMIN, event_kind=LIFECYCLE_EVENT_REVOKED_BY_ADMIN,
        )

    async def revoke_for_system(
        self,
        binding_id: uuid.UUID,
        *,
        revoke_reason: str,
        event_kind: str,
        trace_id: str,
        actor_person_id: uuid.UUID | None = None,
        detail: dict[str, Any] | None = None,
    ) -> DataAccessBinding:
        """System-context revoke (offboarding T+0, expiry, confirmation lapse).

        Skips actor authorisation — the caller IS the policy. Still the same
        semantic transition + reconciler path as a human revoke.
        """
        from app.domains.data_access.service import DataAccessService  # noqa: PLC0415

        binding = await self._binding(binding_id, lock=False)
        result = await DataAccessService().revoke_data_access_binding(
            self.db, actor=None, binding_id=binding_id, reason=revoke_reason,
            trace_id=trace_id, system_context=True, system_actor_user_id=actor_person_id,
        )
        if result.revoked_pending_count:
            record_event(
                self.db, binding, event_kind,
                actor_type="system", actor_person_id=actor_person_id,
                detail={"revoke_reason": revoke_reason, **(detail or {})},
            )
            await self.db.flush()
        await self.db.refresh(binding)
        return binding

    async def revoke_for_offboarding(
        self,
        binding_id: uuid.UUID,
        *,
        leaver_person_id: uuid.UUID,
        actor_person_id: uuid.UUID | None,
        trace_id: str,
    ) -> bool:
        """T+0 handler body. Returns False when the binding is not the leaver's."""
        binding = await self._binding(binding_id, lock=False)
        if (
            binding.principal_type_snapshot != c.PRINCIPAL_TYPE_GOOGLE_USER
            or binding.principal_subject_snapshot != str(leaver_person_id)
        ):
            return False
        if binding.status != BINDING_STATUS_ACTIVE:
            return True
        await self.revoke_for_system(
            binding_id, revoke_reason=REVOKE_REASON_OFFBOARDING,
            event_kind=LIFECYCLE_EVENT_REVOKED_OFFBOARDING, trace_id=trace_id,
            actor_person_id=actor_person_id,
            detail={"leaver_person_id": str(leaver_person_id)},
        )
        return True

    async def revoke_for_self_attestation(
        self, binding_id: uuid.UUID, *, actor: IdentityPerson, trace_id: str,
    ) -> DataAccessBinding:
        return await self.release(
            binding_id, actor=actor, reason_note="self attestation: no longer needed",
            trace_id=trace_id, revoke_reason=REVOKE_REASON_SELF_ATTESTATION,
            event_kind=LIFECYCLE_EVENT_REVOKED_SELF_ATTESTATION,
        )

    # ── service-account owner confirmation ──

    async def respond_owner_confirmation(
        self,
        binding_id: uuid.UUID,
        *,
        actor: IdentityPerson,
        decision: str,
        note: str | None,
        settings: LifecycleSettings,
        trace_id: str,
        now: datetime | None = None,
    ) -> DataAccessBinding:
        now = now or datetime.now(UTC)
        if decision not in ("confirm", "revoke"):
            raise AppException(
                code="DATA_ACCESS_OWNER_DECISION_INVALID",
                message="decision must be 'confirm' or 'revoke'.",
                status_code=422,
                field="decision",
            )
        binding = await self._binding(binding_id)
        if binding.principal_type_snapshot != c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT:
            raise AppException(
                code="DATA_ACCESS_NOT_SERVICE_ACCOUNT_GRANT",
                message="Owner confirmation applies to service-account grants only.",
                status_code=409,
                field="binding_id",
            )
        if binding.status != BINDING_STATUS_ACTIVE:
            raise AppException(
                code="DATA_ACCESS_BINDING_NOT_ACTIVE",
                message=f"Binding is {binding.status}.",
                status_code=409,
                field="binding_id",
            )
        if not (await self._is_technical_owner(binding, actor) or await self._is_admin(actor)):
            raise AppException(
                code="DATA_ACCESS_OWNER_CONFIRMATION_NOT_AUTHORIZED",
                message=(
                    "Only the service account's technical owner or a global admin may "
                    "answer this confirmation."
                ),
                status_code=403,
                field="binding_id",
            )
        if decision == "revoke":
            await self.db.flush()
            return await self.release(
                binding_id, actor=actor, reason_note=note, trace_id=trace_id,
                revoke_reason=REVOKE_REASON_OWNER_DECLINED,
                event_kind=LIFECYCLE_EVENT_OWNER_DECLINED,
            )
        previous_due = _aware(binding.owner_confirmation_due_at)
        binding.owner_confirmed_at = now
        binding.owner_confirmed_by_person_id = str(actor.id)
        binding.owner_confirmation_due_at = now + timedelta(
            days=settings.sa_confirmation_interval_days,
        )
        binding.owner_confirmation_reminder_count = 0
        binding.owner_confirmation_last_reminded_at = None
        record_event(
            self.db, binding, LIFECYCLE_EVENT_OWNER_CONFIRMED,
            actor_type="user", actor_person_id=actor.id,
            detail={
                "previous_due_at": previous_due.isoformat() if previous_due else None,
                "next_due_at": binding.owner_confirmation_due_at.isoformat(),
                "note": (note or "").strip()[:500] or None,
            },
        )
        await self.audit.log(
            trace_id=trace_id, actor_type="user", actor_user_id=actor.id,
            action="data_access.binding.owner_confirmed", status="success",
            resource_type="data_access_binding", resource_id=str(binding.id),
            detail={"next_due_at": binding.owner_confirmation_due_at.isoformat()},
        )
        await self.db.flush()
        return binding

    async def list_owner_confirmations_for(
        self,
        person_id: uuid.UUID,
        *,
        now: datetime,
        window: timedelta = _CONFIRMATION_DUE_WINDOW,
    ) -> list[tuple[DataAccessBinding, DataAccessResource]]:
        """Service-account grants this person must confirm (due within ``window``)."""
        rows = (await self.db.execute(
            select(DataAccessBinding, DataAccessResource)
            .join(
                DataAccessServiceAccountProfile,
                DataAccessServiceAccountProfile.principal_id == DataAccessBinding.principal_id,
            )
            .join(DataAccessResource, DataAccessResource.id == DataAccessBinding.resource_id)
            .where(
                DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT,
                DataAccessBinding.status == BINDING_STATUS_ACTIVE,
                DataAccessServiceAccountProfile.technical_owner_person_id == person_id,
                DataAccessBinding.owner_confirmation_due_at.is_not(None),
                DataAccessBinding.owner_confirmation_due_at <= now + window,
            )
            .order_by(DataAccessBinding.owner_confirmation_due_at.asc()),
        )).all()
        return [(b, r) for b, r in rows]

    # ── admin read models ──

    async def events_for(self, binding_id: uuid.UUID) -> list[DataAccessLifecycleEvent]:
        return list((await self.db.scalars(
            select(DataAccessLifecycleEvent)
            .where(DataAccessLifecycleEvent.binding_id == binding_id)
            .order_by(
                DataAccessLifecycleEvent.created_at.desc(),
                DataAccessLifecycleEvent.id.desc(),
            ),
        )).all())

    async def summary(self, *, now: datetime, settings: LifecycleSettings) -> dict[str, Any]:
        first = settings.reminder_ladder_days[0] if settings.reminder_ladder_days else 14
        active = DataAccessBinding.status == BINDING_STATUS_ACTIVE
        personal = DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER
        sa = DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT

        async def _count(*conds: Any) -> int:
            return int(await self.db.scalar(
                select(func.count(DataAccessBinding.id)).where(*conds),
            ) or 0)

        by_type_rows = (await self.db.execute(
            select(DataAccessBinding.principal_type_snapshot, func.count(DataAccessBinding.id))
            .where(active).group_by(DataAccessBinding.principal_type_snapshot),
        )).all()
        since_30d = now - timedelta(days=30)
        lifecycle_revoked_30d = await _count(
            DataAccessBinding.status.in_((BINDING_STATUS_REVOKED_PENDING, BINDING_STATUS_REVOKED)),
            DataAccessBinding.revoked_at.is_not(None),
            DataAccessBinding.revoked_at >= since_30d,
            DataAccessBinding.revoke_reason.in_((
                REVOKE_REASON_EXPIRED, REVOKE_REASON_OWNER_CONFIRMATION_LAPSED,
                REVOKE_REASON_OWNER_DECLINED, REVOKE_REASON_HOLDER_RELEASED,
            )),
        )
        renewals_30d = int(await self.db.scalar(
            select(func.count(DataAccessLifecycleEvent.id)).where(
                DataAccessLifecycleEvent.event_kind == LIFECYCLE_EVENT_RENEWED,
                DataAccessLifecycleEvent.created_at >= since_30d,
            ),
        ) or 0)
        return {
            "active_by_principal_type": {pt: int(n) for pt, n in by_type_rows},
            "expiring": await _count(
                active, personal, DataAccessBinding.expires_at.is_not(None),
                DataAccessBinding.expires_at > now,
                DataAccessBinding.expires_at <= now + timedelta(days=first),
            ),
            "expired_unrevoked": await _count(
                active, personal, DataAccessBinding.expires_at.is_not(None),
                DataAccessBinding.expires_at <= now,
            ),
            "confirmation_due": await _count(
                active, sa, DataAccessBinding.owner_confirmation_due_at.is_not(None),
                DataAccessBinding.owner_confirmation_due_at > now,
                DataAccessBinding.owner_confirmation_due_at <= now + _CONFIRMATION_DUE_WINDOW,
            ),
            "confirmation_overdue": await _count(
                active, sa, DataAccessBinding.owner_confirmation_due_at.is_not(None),
                DataAccessBinding.owner_confirmation_due_at <= now,
            ),
            "lifecycle_revoked_30d": lifecycle_revoked_30d,
            "renewals_30d": renewals_30d,
            "settings": settings.to_setting(),
        }

    async def list_admin_grants(
        self,
        *,
        now: datetime,
        settings: LifecycleSettings,
        principal_type: str | None,
        lifecycle: str | None,
        status: str | None,
        search: str | None,
        page: int,
        page_size: int,
    ) -> tuple[list[dict[str, Any]], int]:
        first = settings.reminder_ladder_days[0] if settings.reminder_ladder_days else 14
        conds: list[Any] = [
            DataAccessBinding.principal_type_snapshot.in_((
                c.PRINCIPAL_TYPE_GOOGLE_USER, c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT,
                c.PRINCIPAL_TYPE_GOOGLE_GROUP,
            )),
        ]
        if principal_type:
            conds.append(DataAccessBinding.principal_type_snapshot == principal_type)
        if status:
            conds.append(DataAccessBinding.status == status)
        elif lifecycle in (
            STATE_EXPIRING, STATE_EXPIRED, STATE_CONFIRMATION_DUE, STATE_CONFIRMATION_OVERDUE,
        ):
            conds.append(DataAccessBinding.status == BINDING_STATUS_ACTIVE)
        if lifecycle == STATE_EXPIRING:
            conds += [
                DataAccessBinding.expires_at.is_not(None),
                DataAccessBinding.expires_at > now,
                DataAccessBinding.expires_at <= now + timedelta(days=first),
            ]
        elif lifecycle == STATE_EXPIRED:
            conds += [DataAccessBinding.expires_at.is_not(None), DataAccessBinding.expires_at <= now]
        elif lifecycle == STATE_CONFIRMATION_DUE:
            conds += [
                DataAccessBinding.owner_confirmation_due_at.is_not(None),
                DataAccessBinding.owner_confirmation_due_at > now,
                DataAccessBinding.owner_confirmation_due_at <= now + _CONFIRMATION_DUE_WINDOW,
            ]
        elif lifecycle == STATE_CONFIRMATION_OVERDUE:
            conds += [
                DataAccessBinding.owner_confirmation_due_at.is_not(None),
                DataAccessBinding.owner_confirmation_due_at <= now,
            ]
        elif lifecycle == "lifecycle_revoked":
            conds.append(DataAccessBinding.revoke_reason.in_((
                REVOKE_REASON_EXPIRED, REVOKE_REASON_OWNER_CONFIRMATION_LAPSED,
                REVOKE_REASON_OWNER_DECLINED, REVOKE_REASON_HOLDER_RELEASED,
            )))
        if search and search.strip():
            like = f"%{search.strip()}%"
            conds.append(or_(
                DataAccessResource.project_id.ilike(like),
                DataAccessResource.dataset_id.ilike(like),
                DataAccessResource.table_id.ilike(like),
                DataAccessResource.display_name.ilike(like),
                DataAccessBinding.principal_email_snapshot.ilike(like),
            ))
        base = (
            select(DataAccessBinding, DataAccessResource)
            .join(DataAccessResource, DataAccessResource.id == DataAccessBinding.resource_id)
            .where(*conds)
        )
        total = int(await self.db.scalar(
            select(func.count()).select_from(base.order_by(None).subquery()),
        ) or 0)
        rows = (await self.db.execute(
            base.order_by(
                DataAccessBinding.status.asc(),
                func.coalesce(
                    DataAccessBinding.expires_at, DataAccessBinding.owner_confirmation_due_at,
                ).asc(),
                DataAccessBinding.granted_at.desc(),
            ).offset((page - 1) * page_size).limit(page_size),
        )).all()
        person_ids: set[str] = set()
        sa_principal_ids: list[uuid.UUID] = []
        for binding, resource in rows:
            if binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER and binding.principal_subject_snapshot:
                person_ids.add(binding.principal_subject_snapshot)
            if binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT and binding.principal_id:
                sa_principal_ids.append(binding.principal_id)
            if resource.owner_person_id:
                person_ids.add(str(resource.owner_person_id))
        owner_by_principal: dict[uuid.UUID, uuid.UUID | None] = {}
        if sa_principal_ids:
            for pid, owner in (await self.db.execute(
                select(
                    DataAccessServiceAccountProfile.principal_id,
                    DataAccessServiceAccountProfile.technical_owner_person_id,
                ).where(DataAccessServiceAccountProfile.principal_id.in_(sa_principal_ids)),
            )).all():
                owner_by_principal[pid] = owner
                if owner:
                    person_ids.add(str(owner))
        labels = await self._person_labels(person_ids)
        items: list[dict[str, Any]] = []
        for binding, resource in rows:
            holder_id = (
                binding.principal_subject_snapshot
                if binding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER else None
            )
            tech_owner = owner_by_principal.get(binding.principal_id) if binding.principal_id else None
            items.append({
                "binding_id": str(binding.id),
                "resource_id": str(resource.id),
                "locator": _locator(resource),
                "resource_type": resource.resource_type,
                "display_name": resource.display_name,
                "resource_owner_person_id": resource.owner_person_id,
                "resource_owner_label": labels.get(str(resource.owner_person_id)) if resource.owner_person_id else None,
                "principal_type": binding.principal_type_snapshot,
                "principal_email": binding.principal_email_snapshot,
                "holder_person_id": holder_id,
                "holder_label": labels.get(holder_id) if holder_id else None,
                "technical_owner_person_id": str(tech_owner) if tech_owner else None,
                "technical_owner_label": labels.get(str(tech_owner)) if tech_owner else None,
                "entitlement": binding.entitlement_snapshot,
                "binding_status": binding.status,
                "reconcile_status": binding.reconcile_status,
                "granted_at": _aware(binding.granted_at).isoformat(),  # type: ignore[union-attr]
                "revoked_at": _aware(binding.revoked_at).isoformat() if binding.revoked_at else None,  # type: ignore[union-attr]
                "revoke_reason": binding.revoke_reason,
                "source_request_id": str(binding.source_request_id) if binding.source_request_id else None,
                **lifecycle_fields(binding, now=now, settings=settings),
            })
        return items, total

    async def _person_labels(self, ids: set[str]) -> dict[str, dict[str, Any]]:
        if not ids:
            return {}
        uuids: list[uuid.UUID] = []
        for raw in ids:
            try:
                uuids.append(uuid.UUID(str(raw)))
            except ValueError:
                continue
        rows = (await self.db.execute(
            select(
                IdentityPerson.id, IdentityPerson.display_name, IdentityPerson.work_email,
                IdentityPerson.employee_no, IdentityPerson.employment_status,
            ).where(IdentityPerson.id.in_(uuids)),
        )).all()
        return {
            str(pid): {
                "display_name": name, "email": email,
                "employee_no": emp_no, "employment_status": status,
            }
            for pid, name, email, emp_no, status in rows
        }

    # ── daily sweep ──

    async def run_daily(
        self, *, now: datetime, settings: LifecycleSettings, trace_id: str,
    ) -> dict[str, Any]:
        """The ``data_access.grant_lifecycle`` cron body (no commit here)."""
        summary: dict[str, Any] = {
            "stamped": 0,
            "reminders_sent": 0,
            "reminders_withheld": 0,
            "expired_revoked": 0,
            "expired_overdue": 0,
            "owner_confirmations_requested": 0,
            "owner_confirmation_reminders": 0,
            "owner_confirmations_withheld": 0,
            "confirmation_lapsed_revoked": 0,
            "confirmation_lapsed_overdue": 0,
            "errors": 0,
        }
        digest: dict[str, list[dict[str, Any]]] = {
            "expiring": [], "expired_overdue": [], "expired_revoked": [],
            "confirmation_due": [], "confirmation_lapsed": [], "confirmation_revoked": [],
        }
        await self._stamp_missing(now=now, settings=settings, summary=summary)
        await self._sweep_personal(now=now, settings=settings, trace_id=trace_id, summary=summary, digest=digest)
        await self._sweep_service_accounts(now=now, settings=settings, trace_id=trace_id, summary=summary, digest=digest)
        summary["digest"] = digest
        return summary

    async def _stamp_missing(
        self, *, now: datetime, settings: LifecycleSettings, summary: dict[str, Any],
    ) -> None:
        personal = list((await self.db.scalars(
            select(DataAccessBinding).where(
                DataAccessBinding.status == BINDING_STATUS_ACTIVE,
                DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER,
                DataAccessBinding.expires_at.is_(None),
            ).with_for_update(),
        )).all())
        for b in personal:
            granted = _aware(b.granted_at) or now
            b.expires_at = max(
                granted + timedelta(days=settings.personal_ttl_days), now + _BACKFILL_GRACE,
            )
            record_event(self.db, b, LIFECYCLE_EVENT_EXPIRY_STAMPED, detail={
                "expires_at": b.expires_at.isoformat(), "source": "daily_backfill",
            })
            summary["stamped"] += 1
        sas = list((await self.db.scalars(
            select(DataAccessBinding).where(
                DataAccessBinding.status == BINDING_STATUS_ACTIVE,
                DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT,
                DataAccessBinding.owner_confirmation_due_at.is_(None),
            ).with_for_update(),
        )).all())
        for b in sas:
            granted = _aware(b.granted_at) or now
            b.owner_confirmation_due_at = max(
                granted + timedelta(days=settings.sa_confirmation_interval_days),
                now + _BACKFILL_GRACE,
            )
            summary["stamped"] += 1
        if personal or sas:
            await self.db.flush()

    async def _sweep_personal(
        self, *, now: datetime, settings: LifecycleSettings, trace_id: str,
        summary: dict[str, Any], digest: dict[str, list[dict[str, Any]]],
    ) -> None:
        ladder = settings.reminder_ladder_days
        first = ladder[0] if ladder else 14
        rows = (await self.db.execute(
            select(DataAccessBinding, DataAccessResource)
            .join(DataAccessResource, DataAccessResource.id == DataAccessBinding.resource_id)
            .where(
                DataAccessBinding.status == BINDING_STATUS_ACTIVE,
                DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GOOGLE_USER,
                DataAccessBinding.expires_at.is_not(None),
                DataAccessBinding.expires_at <= now + timedelta(days=first),
            )
            .order_by(DataAccessBinding.expires_at.asc()),
        )).all()
        for binding, resource in rows:
            expires_at = _aware(binding.expires_at)
            assert expires_at is not None
            holder = await self._holder(binding)
            entry = {
                "binding_id": str(binding.id),
                "locator": _locator(resource),
                "holder_label": (holder.display_name or holder.work_email) if holder else binding.principal_email_snapshot,
                "expires_at": expires_at.isoformat(),
                "days": (expires_at - now).days,
                "renewal_count": int(binding.renewal_count or 0),
            }
            try:
                if expires_at <= now:
                    if settings.auto_revoke_on_expiry:
                        await self.revoke_for_system(
                            binding.id, revoke_reason=REVOKE_REASON_EXPIRED,
                            event_kind=LIFECYCLE_EVENT_EXPIRED_REVOKE, trace_id=trace_id,
                            detail={"expires_at": expires_at.isoformat()},
                        )
                        summary["expired_revoked"] += 1
                        digest["expired_revoked"].append(entry)
                        if settings.employee_push_enabled and _deliverable(holder):
                            await self._notify_expired(holder, binding, resource, trace_id=trace_id)  # type: ignore[arg-type]
                    else:
                        summary["expired_overdue"] += 1
                        digest["expired_overdue"].append(entry)
                        if not await self._has_event_since(
                            binding.id, LIFECYCLE_EVENT_EXPIRED_OVERDUE, since=expires_at,
                        ):
                            record_event(self.db, binding, LIFECYCLE_EVENT_EXPIRED_OVERDUE, detail={
                                "expires_at": expires_at.isoformat(),
                                "auto_revoke_on_expiry": False,
                            })
                    continue
                digest["expiring"].append(entry)
                # Which ladder rung is due? Rung i fires when the remaining time
                # is at most ladder[i] days and the binding has not yet passed it.
                remaining = expires_at - now
                due_stage = 0
                for idx, days in enumerate(ladder):
                    if remaining <= timedelta(days=days):
                        due_stage = idx + 1
                current = int(binding.expiry_reminder_stage or 0)
                if due_stage <= current:
                    continue
                last = _aware(binding.expiry_last_reminded_at)
                if last is not None and now - last < _REMIND_FLOOR:
                    continue
                if not settings.employee_push_enabled or not _deliverable(holder):
                    summary["reminders_withheld"] += 1
                    continue
                await self._notify_expiring(
                    holder, binding, resource, stage=due_stage, settings=settings,  # type: ignore[arg-type]
                    trace_id=trace_id,
                )
                binding.expiry_reminder_stage = due_stage
                binding.expiry_last_reminded_at = now
                record_event(self.db, binding, LIFECYCLE_EVENT_EXPIRY_REMINDER, detail={
                    "stage": due_stage, "days_left": remaining.days,
                    "expires_at": expires_at.isoformat(),
                })
                summary["reminders_sent"] += 1
            except Exception as exc:  # noqa: BLE001 — one grant must not end the sweep
                summary["errors"] += 1
                logger.warning("data_access lifecycle: binding %s failed: %s", binding.id, exc)
        await self.db.flush()

    async def _sweep_service_accounts(
        self, *, now: datetime, settings: LifecycleSettings, trace_id: str,
        summary: dict[str, Any], digest: dict[str, list[dict[str, Any]]],
    ) -> None:
        rows = (await self.db.execute(
            select(DataAccessBinding, DataAccessResource)
            .join(DataAccessResource, DataAccessResource.id == DataAccessBinding.resource_id)
            .where(
                DataAccessBinding.status == BINDING_STATUS_ACTIVE,
                DataAccessBinding.principal_type_snapshot == c.PRINCIPAL_TYPE_GCP_SERVICE_ACCOUNT,
                DataAccessBinding.owner_confirmation_due_at.is_not(None),
                DataAccessBinding.owner_confirmation_due_at <= now + _CONFIRMATION_DUE_WINDOW,
            )
            .order_by(DataAccessBinding.owner_confirmation_due_at.asc()),
        )).all()
        grace = timedelta(days=settings.sa_confirmation_grace_days)
        for binding, resource in rows:
            due = _aware(binding.owner_confirmation_due_at)
            assert due is not None
            owner_id = await self._technical_owner_id(binding)
            owner = await self.db.get(IdentityPerson, owner_id) if owner_id else None
            entry = {
                "binding_id": str(binding.id),
                "locator": _locator(resource),
                "principal_email": binding.principal_email_snapshot,
                "owner_label": (owner.display_name or owner.work_email) if owner else None,
                "due_at": due.isoformat(),
                "overdue_days": max((now - due).days, 0),
            }
            try:
                if due > now:
                    digest["confirmation_due"].append(entry)
                    continue
                if now > due + grace:
                    if settings.auto_revoke_on_confirmation_lapse:
                        await self.revoke_for_system(
                            binding.id, revoke_reason=REVOKE_REASON_OWNER_CONFIRMATION_LAPSED,
                            event_kind=LIFECYCLE_EVENT_OWNER_CONFIRMATION_LAPSED_REVOKE,
                            trace_id=trace_id,
                            detail={"due_at": due.isoformat(), "grace_days": settings.sa_confirmation_grace_days},
                        )
                        summary["confirmation_lapsed_revoked"] += 1
                        digest["confirmation_revoked"].append(entry)
                        if settings.owner_confirmation_push_enabled and _deliverable(owner):
                            await self._notify_owner_lapsed(owner, binding, resource, trace_id=trace_id)  # type: ignore[arg-type]
                        continue
                    summary["confirmation_lapsed_overdue"] += 1
                    digest["confirmation_lapsed"].append(entry)
                else:
                    digest["confirmation_due"].append(entry)
                count = int(binding.owner_confirmation_reminder_count or 0)
                last = _aware(binding.owner_confirmation_last_reminded_at)
                if count > 0 and last is not None and now - last < _OWNER_REMINDER_EVERY:
                    continue
                if not settings.owner_confirmation_push_enabled or not _deliverable(owner):
                    summary["owner_confirmations_withheld"] += 1
                    continue
                await self._notify_owner_confirmation(
                    owner, binding, resource, due=due, count=count, settings=settings,  # type: ignore[arg-type]
                    trace_id=trace_id,
                )
                binding.owner_confirmation_reminder_count = count + 1
                binding.owner_confirmation_last_reminded_at = now
                kind = (
                    LIFECYCLE_EVENT_OWNER_CONFIRMATION_REQUESTED if count == 0
                    else LIFECYCLE_EVENT_OWNER_CONFIRMATION_REMINDER
                )
                record_event(self.db, binding, kind, detail={
                    "due_at": due.isoformat(), "reminder_count": count + 1,
                    "owner_person_id": str(owner_id) if owner_id else None,
                })
                if count == 0:
                    summary["owner_confirmations_requested"] += 1
                else:
                    summary["owner_confirmation_reminders"] += 1
            except Exception as exc:  # noqa: BLE001
                summary["errors"] += 1
                logger.warning("data_access lifecycle: SA binding %s failed: %s", binding.id, exc)
        await self.db.flush()

    async def _has_event_since(
        self, binding_id: uuid.UUID, kind: str, *, since: datetime,
    ) -> bool:
        hit = await self.db.scalar(
            select(DataAccessLifecycleEvent.id).where(
                DataAccessLifecycleEvent.binding_id == binding_id,
                DataAccessLifecycleEvent.event_kind == kind,
                DataAccessLifecycleEvent.created_at >= since,
            ).limit(1),
        )
        return hit is not None

    # ── notifications ──

    async def _notify_expiring(
        self, holder: IdentityPerson, binding: DataAccessBinding, resource: DataAccessResource,
        *, stage: int, settings: LifecycleSettings, trace_id: str,
    ) -> None:
        from app.notifications.service import (  # noqa: PLC0415
            CardAction,
            CardStat,
            LocalizedText,
            NotificationEnvelope,
            NotificationService,
        )

        expires_at = _aware(binding.expires_at)
        assert expires_at is not None
        days = max((expires_at - datetime.now(UTC)).days, 0)
        locator = _locator(resource)
        lang = _lang(holder)
        remaining = settings.self_renewals_before_reapproval - int(binding.renewal_count or 0)
        renew_url = f"{MY_GRANTS_URL}?renew={binding.id}"
        title = {
            "zh": f"⏳ 数据集访问权限 {days} 天后到期:{locator}",
            "en": f"⏳ Dataset access expires in {days} day(s): {locator}",
            "ja": f"⏳ データセットのアクセス権が {days} 日後に期限切れ:{locator}",
        }[lang]
        body = {
            "zh": (
                f"你对 BigQuery 数据集 {locator} 的个人读权限将于 "
                f"{expires_at.date().isoformat()} 到期。还需要就点「续期」写一句理由"
                f"(还可自助续期 {max(remaining, 0)} 次,之后需重新申请);不需要了就「释放」。"
            ),
            "en": (
                f"Your personal read access to BigQuery dataset {locator} expires on "
                f"{expires_at.date().isoformat()}. Still need it? Renew with a one-line reason "
                f"({max(remaining, 0)} self-renewal(s) left before a new request is required). "
                "Done with it? Release it."
            ),
            "ja": (
                f"BigQuery データセット {locator} への個人の読み取り権限は "
                f"{expires_at.date().isoformat()} に期限切れになります。まだ必要なら「更新」で理由を一言"
                f"(自己更新は残り {max(remaining, 0)} 回、その後は再申請)、不要なら「解放」してください。"
            ),
        }[lang]
        await NotificationService(self.db).dispatch(
            NotificationEnvelope(
                recipient=holder, title=title, body_text=body,
                channels=["feishu", "email"], template_key=TEMPLATE_EXPIRY_REMINDER,
                context={
                    "locator": locator, "days": str(days),
                    "expires_at": expires_at.date().isoformat(),
                    "renewals_remaining": str(max(remaining, 0)),
                },
                url=renew_url,
                card_subtitle=LocalizedText(
                    zh=f"第 {stage} 次提醒 · {expires_at.date().isoformat()} 到期",
                    en=f"Reminder {stage} · expires {expires_at.date().isoformat()}",
                    ja=f"{stage} 回目のリマインド · {expires_at.date().isoformat()} 期限",
                ),
                card_stats=[
                    CardStat(label=LocalizedText(zh="数据集", en="Dataset", ja="データセット"), value=locator),
                    CardStat(label=LocalizedText(zh="剩余天数", en="Days left", ja="残り日数"), value=str(days)),
                ],
                card_actions=[
                    CardAction(label=LocalizedText(zh="续期(写理由)", en="Renew (with reason)", ja="更新(理由を記入)"), url=renew_url, style="primary"),
                    CardAction(label=LocalizedText(zh="释放", en="Release", ja="解放"), url=f"{MY_GRANTS_URL}?release={binding.id}", style="danger"),
                ],
                related_resource_type="data_access_binding",
                related_resource_id=binding.id,
            ),
            trace_id=trace_id, actor_user_id=None,
        )

    async def _notify_expired(
        self, holder: IdentityPerson, binding: DataAccessBinding, resource: DataAccessResource,
        *, trace_id: str,
    ) -> None:
        from app.notifications.service import (  # noqa: PLC0415
            CardAction,
            LocalizedText,
            NotificationEnvelope,
            NotificationService,
        )

        locator = _locator(resource)
        lang = _lang(holder)
        title = {
            "zh": f"数据集访问权限已到期并回收:{locator}",
            "en": f"Dataset access expired and was revoked: {locator}",
            "ja": f"データセットのアクセス権が期限切れで回収されました:{locator}",
        }[lang]
        body = {
            "zh": f"你对 {locator} 的个人读权限已到期,Helix 已发起回收。如仍需要,请重新提交数据访问申请。",
            "en": f"Your personal read access to {locator} has expired and Helix has started the revoke. Submit a new data-access request if you still need it.",
            "ja": f"{locator} への個人の読み取り権限は期限切れとなり、Helix が回収を開始しました。引き続き必要な場合は再申請してください。",
        }[lang]
        await NotificationService(self.db).dispatch(
            NotificationEnvelope(
                recipient=holder, title=title, body_text=body,
                channels=["feishu", "email"], template_key=TEMPLATE_EXPIRED_REVOKED,
                context={"locator": locator},
                url="/my/data-access/requests",
                card_actions=[CardAction(
                    label=LocalizedText(zh="重新申请", en="Request again", ja="再申請"),
                    url="/my/data-access/requests", style="primary",
                )],
                related_resource_type="data_access_binding",
                related_resource_id=binding.id,
            ),
            trace_id=trace_id, actor_user_id=None,
        )

    async def _notify_owner_confirmation(
        self, owner: IdentityPerson, binding: DataAccessBinding, resource: DataAccessResource,
        *, due: datetime, count: int, settings: LifecycleSettings, trace_id: str,
    ) -> None:
        from app.notifications.service import (  # noqa: PLC0415
            CardAction,
            CardStat,
            LocalizedText,
            NotificationEnvelope,
            NotificationService,
        )

        locator = _locator(resource)
        lang = _lang(owner)
        sa = binding.principal_email_snapshot
        confirm_url = f"{MY_GRANTS_URL}?confirm={binding.id}"
        overdue = max((datetime.now(UTC) - due).days, 0)
        title = {
            "zh": f"请确认服务账号仍需要数据集访问:{sa} → {locator}",
            "en": f"Confirm this service account still needs dataset access: {sa} → {locator}",
            "ja": f"サービスアカウントのデータセットアクセスを確認してください:{sa} → {locator}",
        }[lang]
        body = {
            "zh": (
                f"你是服务账号 {sa} 的技术负责人。它对 {locator} 的读权限到了半年度确认:"
                f"还在用就点「确认仍需要」;不用了就「回收」。"
                + (f"已逾期 {overdue} 天,超过 {settings.sa_confirmation_grace_days} 天宽限后按策略处理。" if overdue else "")
            ),
            "en": (
                f"You are the technical owner of service account {sa}. Its read access to {locator} "
                "is due for its semiannual confirmation: press Confirm if the workload still needs it, "
                "or Revoke if not."
                + (f" Overdue by {overdue} day(s); policy applies after the {settings.sa_confirmation_grace_days}-day grace." if overdue else "")
            ),
            "ja": (
                f"あなたはサービスアカウント {sa} の技術責任者です。{locator} への読み取り権限の半期確認の時期です:"
                "まだ必要なら「確認」、不要なら「回収」を押してください。"
                + (f"{overdue} 日超過しています。{settings.sa_confirmation_grace_days} 日の猶予後にポリシーが適用されます。" if overdue else "")
            ),
        }[lang]
        await NotificationService(self.db).dispatch(
            NotificationEnvelope(
                recipient=owner, title=title, body_text=body,
                channels=["feishu", "email"], template_key=TEMPLATE_OWNER_CONFIRMATION,
                context={
                    "service_account": sa, "locator": locator,
                    "due_at": due.date().isoformat(), "overdue_days": str(overdue),
                    "grace_days": str(settings.sa_confirmation_grace_days),
                },
                url=confirm_url,
                card_subtitle=LocalizedText(
                    zh=f"半年度确认 · 第 {count + 1} 次提醒",
                    en=f"Semiannual confirmation · reminder {count + 1}",
                    ja=f"半期確認 · {count + 1} 回目",
                ),
                card_stats=[
                    CardStat(label=LocalizedText(zh="服务账号", en="Service account", ja="サービスアカウント"), value=sa),
                    CardStat(label=LocalizedText(zh="数据集", en="Dataset", ja="データセット"), value=locator),
                ],
                card_actions=[
                    CardAction(label=LocalizedText(zh="确认仍需要", en="Confirm — still needed", ja="確認(必要)"), url=confirm_url, style="primary"),
                    CardAction(label=LocalizedText(zh="回收", en="Revoke", ja="回収"), url=f"{MY_GRANTS_URL}?confirm={binding.id}&decision=revoke", style="danger"),
                ],
                related_resource_type="data_access_binding",
                related_resource_id=binding.id,
            ),
            trace_id=trace_id, actor_user_id=None,
        )

    async def _notify_owner_lapsed(
        self, owner: IdentityPerson, binding: DataAccessBinding, resource: DataAccessResource,
        *, trace_id: str,
    ) -> None:
        from app.notifications.service import (  # noqa: PLC0415
            NotificationEnvelope,
            NotificationService,
        )

        locator = _locator(resource)
        sa = binding.principal_email_snapshot
        lang = _lang(owner)
        title = {
            "zh": f"服务账号数据集权限因未确认已回收:{sa} → {locator}",
            "en": f"Service-account dataset access revoked — confirmation lapsed: {sa} → {locator}",
            "ja": f"確認がなかったためサービスアカウントのアクセス権を回収しました:{sa} → {locator}",
        }[lang]
        body = {
            "zh": f"{sa} 对 {locator} 的读权限半年度确认逾期未答复,Helix 已按策略发起回收。如工作负载仍需要,请重新提交服务账号授权申请。",
            "en": f"The semiannual confirmation for {sa}'s read access to {locator} was not answered in time; Helix has started the revoke. Submit a new service-account grant request if the workload still needs it.",
            "ja": f"{sa} の {locator} への読み取り権限の半期確認が期限内に回答されなかったため、Helix は回収を開始しました。必要であれば再申請してください。",
        }[lang]
        await NotificationService(self.db).dispatch(
            NotificationEnvelope(
                recipient=owner, title=title, body_text=body,
                channels=["feishu", "email"], template_key=TEMPLATE_OWNER_LAPSED,
                context={"service_account": sa, "locator": locator},
                url="/my/data-access/requests",
                related_resource_type="data_access_binding",
                related_resource_id=binding.id,
            ),
            trace_id=trace_id, actor_user_id=None,
        )


async def send_it_digest(
    db: AsyncSession,
    *,
    day: str,
    summary: dict[str, Any],
    settings: LifecycleSettings,
    trace_id: str,
) -> int:
    """One digest card per active global admin per day (outbox-deduped).

    Sent only when something needs eyes: expiring / expired / confirmations
    due or lapsed / actions taken. Returns the number of admins reached.
    """
    from app.notifications.models import NotificationOutbox  # noqa: PLC0415
    from app.notifications.service import (  # noqa: PLC0415
        CardAction,
        CardHighlight,
        CardStat,
        LocalizedText,
        NotificationEnvelope,
        NotificationService,
    )

    digest = summary.get("digest") or {}
    expiring = digest.get("expiring") or []
    expired_overdue = digest.get("expired_overdue") or []
    expired_revoked = digest.get("expired_revoked") or []
    conf_due = digest.get("confirmation_due") or []
    conf_lapsed = digest.get("confirmation_lapsed") or []
    conf_revoked = digest.get("confirmation_revoked") or []
    if not any((expiring, expired_overdue, expired_revoked, conf_due, conf_lapsed, conf_revoked)):
        return 0
    digest_id = uuid.uuid5(uuid.NAMESPACE_URL, f"helix:data-access-lifecycle-digest:{day}")
    already = set((await db.execute(
        select(NotificationOutbox.recipient_person_id).where(
            NotificationOutbox.related_resource_type == DIGEST_RESOURCE_TYPE,
            # CHAR(36) column: bind the string form, a UUID object never matches.
            NotificationOutbox.related_resource_id == str(digest_id),
        )
    )).scalars().all())
    admin_ids = list((await db.execute(
        select(AuthUserRole.user_id)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(AuthRole.name.in_(("global_admin", "it_admin")))
    )).scalars().all())

    def _rows(items: list[dict[str, Any]], fmt: str) -> list[str | LocalizedText]:
        out: list[str | LocalizedText] = [fmt.format(**i) for i in items[:_DIGEST_MAX_ROWS]]
        if len(items) > _DIGEST_MAX_ROWS:
            out.append(f"… +{len(items) - _DIGEST_MAX_ROWS}")
        return out

    push_state = LocalizedText(
        zh=("已开启" if settings.employee_push_enabled else "未开启(仅 IT 汇总)"),
        en=("ON" if settings.employee_push_enabled else "OFF (IT digest only)"),
        ja=("オン" if settings.employee_push_enabled else "オフ(IT 向け集計のみ)"),
    )
    revoke_state = LocalizedText(
        zh=("已开启" if settings.auto_revoke_on_expiry else "未开启"),
        en=("ON" if settings.auto_revoke_on_expiry else "OFF"),
        ja=("オン" if settings.auto_revoke_on_expiry else "オフ"),
    )
    highlights: list[Any] = []
    if expired_overdue:
        highlights.append(CardHighlight(
            label=LocalizedText(zh="已过期待处理", en="Expired, awaiting action", ja="期限切れ・未処理"),
            items=_rows(expired_overdue, "[b]{locator}[/b] · {holder_label} · {days}d"),
        ))
    if expiring:
        highlights.append(CardHighlight(
            label=LocalizedText(zh="即将到期", en="Expiring soon", ja="期限間近"),
            items=_rows(expiring, "[b]{locator}[/b] · {holder_label} · {days}d"),
        ))
    if conf_lapsed:
        highlights.append(CardHighlight(
            label=LocalizedText(zh="负责人确认已逾期", en="Owner confirmation lapsed", ja="責任者確認が超過"),
            items=_rows(conf_lapsed, "[b]{principal_email}[/b] → {locator} · {owner_label} · +{overdue_days}d"),
        ))
    if conf_due:
        highlights.append(CardHighlight(
            label=LocalizedText(zh="待负责人确认", en="Owner confirmation due", ja="責任者確認待ち"),
            items=_rows(conf_due, "[b]{principal_email}[/b] → {locator} · {owner_label}"),
        ))
    if expired_revoked or conf_revoked:
        highlights.append(CardHighlight(
            label=LocalizedText(zh="今日已回收", en="Revoked today", ja="本日回収"),
            items=_rows(expired_revoked + conf_revoked, "[b]{locator}[/b]"),
        ))

    svc = NotificationService(db)
    sent = 0
    for aid in admin_ids:
        if aid in already:
            continue
        admin = await db.get(IdentityPerson, aid)
        if admin is None or not _deliverable(admin):
            continue
        lang = _lang(admin)
        push_txt = {"zh": push_state.zh, "en": push_state.en, "ja": push_state.ja}[lang]
        revoke_txt = {"zh": revoke_state.zh, "en": revoke_state.en, "ja": revoke_state.ja}[lang]
        title = {
            "zh": f"DAX 授权日报:{len(expiring)} 条即将到期 · {len(expired_overdue)} 条已过期 · {len(conf_lapsed)} 条负责人确认逾期",
            "en": f"DAX grants daily — {len(expiring)} expiring · {len(expired_overdue)} expired · {len(conf_lapsed)} owner confirmations lapsed",
            "ja": f"DAX 権限日報:期限間近 {len(expiring)} · 期限切れ {len(expired_overdue)} · 責任者確認超過 {len(conf_lapsed)}",
        }[lang]
        body = {
            "zh": (
                f"个人授权:{len(expiring)} 条 {settings.reminder_ladder_days[0]} 天内到期,{len(expired_overdue)} 条已过期待处理,"
                f"今日回收 {len(expired_revoked)} 条。服务账号:{len(conf_due)} 条待确认,{len(conf_lapsed)} 条逾期,今日回收 {len(conf_revoked)} 条。"
                f"员工提醒{push_txt};到期自动回收{revoke_txt}。"
            ),
            "en": (
                f"Personal grants: {len(expiring)} expire within {settings.reminder_ladder_days[0]} days, "
                f"{len(expired_overdue)} expired awaiting action, {len(expired_revoked)} revoked today. "
                f"Service accounts: {len(conf_due)} confirmations due, {len(conf_lapsed)} lapsed, {len(conf_revoked)} revoked today. "
                f"Employee reminders {push_txt}; auto-revoke on expiry {revoke_txt}."
            ),
            "ja": (
                f"個人権限:{settings.reminder_ladder_days[0]} 日以内に期限切れ {len(expiring)} 件、期限切れ未処理 {len(expired_overdue)} 件、"
                f"本日回収 {len(expired_revoked)} 件。サービスアカウント:確認待ち {len(conf_due)} 件、超過 {len(conf_lapsed)} 件、本日回収 {len(conf_revoked)} 件。"
                f"従業員リマインドは{push_txt}、期限切れ自動回収は{revoke_txt}。"
            ),
        }[lang]
        # SAVEPOINT + ``step.reason()``, replacing ``except Exception`` +
        # ``str(exc)``. ``dispatch`` self-isolates its own writes, so the
        # savepoint is defence in depth; the fixed defect is the error text
        # (``str(AppException)`` renders as an empty sentinel).
        async with best_effort(
            db, label=f"data-access lifecycle digest to admin {admin.id}",
            log=logger, log_level=logging.DEBUG,
        ) as step:
            await svc.dispatch(
                NotificationEnvelope(
                    recipient=admin, title=title, body_text=body,
                    channels=["feishu", "email"], template_key=TEMPLATE_IT_DIGEST,
                    context={
                        "day": day,
                        "expiring_count": str(len(expiring)),
                        "expired_count": str(len(expired_overdue)),
                        "expired_revoked_count": str(len(expired_revoked)),
                        "confirmation_due_count": str(len(conf_due)),
                        "confirmation_lapsed_count": str(len(conf_lapsed)),
                        "confirmation_revoked_count": str(len(conf_revoked)),
                        "employee_delivery": push_txt,
                        "auto_revoke": revoke_txt,
                    },
                    url=ADMIN_GRANTS_URL,
                    card_stats=[
                        CardStat(label=LocalizedText(zh="即将到期", en="Expiring", ja="期限間近"), value=str(len(expiring))),
                        CardStat(label=LocalizedText(zh="已过期待处理", en="Expired", ja="期限切れ"), value=str(len(expired_overdue))),
                        CardStat(label=LocalizedText(zh="待负责人确认", en="Confirmations due", ja="確認待ち"), value=str(len(conf_due))),
                        CardStat(label=LocalizedText(zh="确认逾期", en="Lapsed", ja="超過"), value=str(len(conf_lapsed))),
                    ],
                    card_highlights=highlights,
                    card_actions=[CardAction(
                        label=LocalizedText(zh="打开 DAX 授权清单", en="Open the DAX grants console", ja="DAX 権限一覧を開く"),
                        url=ADMIN_GRANTS_URL, style="primary",
                    )],
                    related_resource_type=DIGEST_RESOURCE_TYPE,
                    related_resource_id=digest_id,
                ),
                trace_id=trace_id, actor_user_id=None,
            )
        if step.ok:
            sent += 1
        else:
            logger.warning(
                "data_access lifecycle digest to %s failed: %s",
                admin.work_email, step.reason(),  # type: ignore[union-attr]
            )
    return sent


__all__ = [
    "ADMIN_GRANTS_URL",
    "LIFECYCLE_STATES",
    "MY_GRANTS_URL",
    "SETTING_KEY",
    "DataAccessLifecycleService",
    "LifecycleSettings",
    "lifecycle_fields",
    "lifecycle_state",
    "load_lifecycle_settings",
    "record_event",
    "replace",
    "save_lifecycle_settings",
    "select_personal_active_bindings",
    "send_it_digest",
    "stamp_new_binding",
]