"""SSO 撤销落地 —— 把「Helix 说撤了」和「IdP 侧真的没了」接起来(migration 603)。

Why this module exists (2026-09-05 audit). ``SsoGrantService.revoke`` flips
``status='revoked'`` and pages the system owner with a card whose literal
body was "the third-party SSO/SAML access has NOT been automatically removed
— please disable it". No task, no due date, no re-nag, no verification. After
the "revoke" the grant also drops out of the semiannual self-attestation
snapshot, so the residual access became *less* visible than before.

The second half of a revoke is now a first-class lifecycle on the grant row
(``SsoGrant.removal_*`` — see the ``REMOVAL_*`` constants in ``models.py``):

    revoke() ──► open_task()  ── manual_notify ProvisioningTask, evidence
                                  required to complete (ticket / console URL)
              ──► reconcile()  ── daily: re-open a missing task, remind on a
                                  widening ladder, open a compliance incident
                                  at day 14, and — for Google-IdP grants —
                                  read Admin Reports' SAML login audit:
                                    login after revoke  → contradicted (task
                                                          re-opened, owner+IT
                                                          paged with the time)
                                    no login N days     → verified
                                    after confirmation
                                    account gone at IdP → verified

Everything below lives inside the SSO domain. The two cross-domain seams
are (a) the provisioning queue — the sanctioned work-queue primitive — whose
post-success hook calls :meth:`SsoRemovalService.on_task_completed`, and
(b) the Google driver, resolved from the org-level ``google_workspace``
integration the same way ``email_group`` does.

Design rules honoured here:
* **Notify ≠ handled.** Every notification is paired with a row someone owns
  (the task) and a clock (the ladder). Ignoring a card does not silence it.
* **No speculative backfill.** Rows revoked before 603 are
  ``unverified_legacy``; admins open a review task explicitly.
* **Ladders widen, never tighten** (DECISIONS 2026-08-30): 3 / 7 / 14 days,
  then every 7 days for as long as the item is open.
* **Verification is evidence, not inference.** A "verified" needs either the
  account to be gone at the IdP or a name-aligned SAML login history with no
  entry after the revoke; an app whose IdP name we cannot align to is
  reported as *unverifiable*, not verified.
"""
from __future__ import annotations

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

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

from app.audit.service import AuditService
from app.compliance.models import (
    INCIDENT_ACCEPTED,
    INCIDENT_OPEN,
    SEVERITY_WARNING,
    ComplianceIncident,
)
from app.compliance.service import ComplianceIncidentService
from app.core.exceptions import AppException
from app.core.txn import best_effort
from app.domains.provisioning.models import (
    ACTION_REVOKE,
    EXEC_MANUAL_NOTIFY,
    SOURCE_AUDIT,
    SOURCE_MANUAL,
    STATUS_NOTIFIED,
    TARGET_SSO_GRANT,
    TERMINAL_TASK_STATUSES,
    ProvisioningTask,
)
from app.domains.sso.models import (
    REMOVAL_CONFIRMED,
    REMOVAL_CONTRADICTED,
    REMOVAL_NOT_APPLICABLE,
    REMOVAL_OPEN_STATUSES,
    REMOVAL_PENDING,
    REMOVAL_UNVERIFIED_LEGACY,
    REMOVAL_VERIFIED,
    SSO_GRANT_STATUS_REVOKED,
    SsoApp,
    SsoGrant,
)
from app.identity.person_models import IdentityPerson
from app.integrations.service import get_platform_setting

logger = logging.getLogger(__name__)

SETTING_KEY = "sso.removal"
INCIDENT_TYPE_SSO_REMOVAL_OVERDUE = "sso_removal_overdue"

# Notification template keys — registered in
# ``app.notifications.service._DEFAULT_TEMPLATES`` (the locale-parity guard
# fails the build on an unregistered key).
TEMPLATE_REMINDER = "sso.removal_reminder"
TEMPLATE_CONTRADICTED = "sso.removal_contradicted"

# Re-sending the same reminder inside this window is a bug, not a nudge — a
# manual cron re-run must not double-page the owner.
_REMIND_FLOOR = timedelta(hours=20)
# Don't hammer Reports API: one check per grant per day is plenty.
_VERIFY_FLOOR = timedelta(hours=20)
# Pre-revoke look-back used to establish that our catalog app name and the
# IdP's ``application_name`` agree (see ``_google_evidence``).
_ALIGNMENT_LOOKBACK = timedelta(days=180)
# Hard cap on Reports calls per reconcile tick — the daily cron is not a
# backfill engine; a big legacy queue drains over a few days.
_MAX_VERIFY_PER_RUN = 200

# Google's SAML application name is the value the admin typed when creating
# the app in Google admin. Ours is what the requester / admin typed in Helix.
# Normalise both before comparing so "Datadog " ≠ "datadog" doesn't produce a
# false "verified".
def _norm(name: str | None) -> str:
    return " ".join((name or "").strip().lower().split())


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 _pick(lang: str, *, zh: str, en: str, ja: str) -> str:
    return {"zh": zh, "en": en, "ja": ja}.get(lang, zh)


@dataclass(frozen=True)
class RemovalSettings:
    """``platform_settings['sso.removal']`` — the cadence knobs.

    ``reminder_ladder_days`` is validated to be strictly increasing and
    non-empty; a malformed setting falls back to the default rather than
    silencing the ladder (an empty ladder would mean "never remind").
    """

    verify_via_google_reports: bool = True
    reminder_ladder_days: tuple[int, ...] = (3, 7, 14)
    verified_after_days: int = 7
    incident_after_days: int = 14

    @classmethod
    def from_setting(cls, value: Any) -> RemovalSettings:
        if not isinstance(value, dict):
            return cls()
        ladder_raw = value.get("reminder_ladder_days")
        ladder: tuple[int, ...] = cls.reminder_ladder_days
        if isinstance(ladder_raw, list) and ladder_raw:
            try:
                cand = tuple(int(x) for x in ladder_raw)
            except (TypeError, ValueError):
                cand = ()
            if cand and all(x > 0 for x in cand) and list(cand) == sorted(set(cand)):
                ladder = cand
        try:
            verified_after = max(1, int(value.get("verified_after_days") or 7))
        except (TypeError, ValueError):
            verified_after = 7
        try:
            incident_after = max(1, int(value.get("incident_after_days") or 14))
        except (TypeError, ValueError):
            incident_after = 14
        return cls(
            verify_via_google_reports=bool(
                value.get("verify_via_google_reports", True),
            ),
            reminder_ladder_days=ladder,
            verified_after_days=verified_after,
            incident_after_days=incident_after,
        )

    def to_setting(self) -> dict[str, Any]:
        return {
            "verify_via_google_reports": self.verify_via_google_reports,
            "reminder_ladder_days": list(self.reminder_ladder_days),
            "verified_after_days": self.verified_after_days,
            "incident_after_days": self.incident_after_days,
        }


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


def reminder_due(
    *,
    revoked_at: datetime,
    reminder_count: int,
    last_reminded_at: datetime | None,
    now: datetime,
    ladder: tuple[int, ...],
) -> bool:
    """Pure ladder arithmetic, unit-testable.

    The ``n``-th reminder (0-based) is due on ``ladder[n]`` days after the
    revoke; once the ladder is exhausted the tail repeats at the ladder's
    last interval (never shorter than 7 days). A reminder already sent within
    ``_REMIND_FLOOR`` blocks re-sending regardless of the ladder.
    """
    if not ladder:
        return False
    age_days = (now - revoked_at).days
    if reminder_count < len(ladder):
        due_day = ladder[reminder_count]
    else:
        tail = ladder[-1] - ladder[-2] if len(ladder) >= 2 else ladder[-1]
        tail = max(tail, 7)
        due_day = ladder[-1] + (reminder_count - len(ladder) + 1) * tail
    if age_days < due_day:
        return False
    if last_reminded_at is not None and now - last_reminded_at < _REMIND_FLOOR:
        return False
    return True


class SsoRemovalService:
    """Owns the ``removal_*`` half of :class:`SsoGrant`."""

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

    # ── Task lifecycle ─────────────────────────────────────────────────

    async def open_task(
        self,
        grant: SsoGrant,
        *,
        source: str,
        actor_user_id: uuid.UUID | None,
        trace_id: str,
        reason: str | None = None,
        notify: bool = True,
    ) -> ProvisioningTask:
        """Ensure a revoked grant has an OPEN manual_notify task and return it.

        Idempotent: an existing non-terminal task is returned untouched (the
        offboarding T+0 path and the reconcile cron can both call this).

        ``notify=True`` runs the task through ``execute_task`` so the
        assignee's Feishu/email card goes out now and the task lands in
        ``notified``. ``notify=False`` marks it ``notified`` WITHOUT a DM —
        used by the offboarding bulk path when the grant has no owner: the
        fallback recipients would be the IT admins who already receive the
        leaver's coalesced summary (Round-59 R59F-C forbids N×M cards). The
        dispatcher never picks a ``notified`` task, so nothing double-pages.
        """
        if grant.status != SSO_GRANT_STATUS_REVOKED:
            raise AppException(
                code="SSO_GRANT_NOT_REVOKED",
                message="A removal task can only be opened for a revoked grant.",
                status_code=409,
            )
        existing = await self._open_task_for(grant)
        if existing is not None:
            return existing

        from app.domains.provisioning.schemas import (  # noqa: PLC0415
            ProvisioningTaskCreate,
        )
        from app.domains.provisioning.service import (  # noqa: PLC0415
            ProvisioningService,
        )

        holder_id = uuid.UUID(str(grant.person_id))
        assignee_id = await self._active_owner_id(grant)
        app_name = await self._idp_app_name(grant)
        prov = ProvisioningService(self.db)
        task = await prov.create_task(ProvisioningTaskCreate(
            user_id=holder_id,
            software_id=None,
            action=ACTION_REVOKE,
            target_type=TARGET_SSO_GRANT,
            target_id=uuid.UUID(str(grant.id)),
            source=source,
            execution_mode=EXEC_MANUAL_NOTIFY,
            assignee_id=assignee_id,
            context_data={
                "sso_grant_id": str(grant.id),
                "system_name": grant.system_name,
                "sso_provider": grant.sso_provider,
                "idp_app_name": app_name,
                "revoke_reason": grant.revoke_reason or reason,
                # The completion path reads this to insist on evidence even
                # on the one-click token page.
                "requires_evidence": True,
            },
        ))
        grant.removal_task_id = task.id
        if grant.removal_status in (REMOVAL_NOT_APPLICABLE, REMOVAL_UNVERIFIED_LEGACY):
            grant.removal_status = REMOVAL_PENDING
        grant.updated_at = datetime.now(UTC)
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user" if actor_user_id else "system",
            actor_user_id=actor_user_id,
            action="sso.grant.removal_task_opened",
            status="success",
            resource_type="sso_grant",
            resource_id=str(grant.id),
            detail={
                "task_id": str(task.id),
                "assignee_person_id": str(assignee_id) if assignee_id else None,
                "source": source,
                "removal_status": grant.removal_status,
            },
        )
        if notify:
            # execute_task → manual_notify branch → DM + status=notified. DM
            # delivery failure is caught inside (goes to the outbox retry);
            # a raise here would be a real DB problem and must surface.
            await prov.execute_task(task, trace_id=trace_id, actor_user_id=actor_user_id)
        else:
            task.status = STATUS_NOTIFIED
            await self.db.flush()
            await self.audit.log(
                trace_id=trace_id,
                actor_type="system",
                actor_user_id=None,
                action="provisioning.manual_notified",
                status="success",
                resource_type="provisioning_task",
                resource_id=str(task.id),
                detail={
                    "action": task.action,
                    "target_type": task.target_type,
                    "user_id": str(task.user_id),
                    "coalesced": True,
                    "note": "no active system owner; IT reached via the leaver summary card",
                },
            )
        return task

    async def on_task_completed(
        self, task: ProvisioningTask, *, trace_id: str | None = None,
    ) -> SsoGrant | None:
        """Provisioning post-success hook: a human said "removed", with proof."""
        grant = await self._grant_for_task(task)
        if grant is None:
            return None
        now = task.completed_at or datetime.now(UTC)
        ctx = task.context_data or {}
        confirmed_by: uuid.UUID | None = None
        raw_by = ctx.get("completed_by_person_id")
        if raw_by:
            try:
                confirmed_by = uuid.UUID(str(raw_by))
            except (TypeError, ValueError):
                confirmed_by = None
        grant.removal_status = REMOVAL_CONFIRMED
        grant.removal_confirmed_at = now
        grant.removal_confirmed_by_person_id = confirmed_by
        grant.removal_external_ref = (task.external_ref or "")[:512] or None
        grant.removal_check_error = None
        grant.updated_at = datetime.now(UTC)
        await self.db.flush()
        await self._resolve_incident(
            grant, note="removal confirmed by task completion", trace_id=trace_id or "",
        )
        await self.audit.log(
            trace_id=trace_id or f"provisioning_task:{task.id}",
            actor_type="user" if confirmed_by else "system",
            actor_user_id=confirmed_by,
            action="sso.grant.removal_confirmed",
            status="success",
            resource_type="sso_grant",
            resource_id=str(grant.id),
            detail={
                "task_id": str(task.id),
                "external_ref": grant.removal_external_ref,
            },
        )
        return grant

    async def on_task_not_completed(
        self, task: ProvisioningTask, *, outcome: str, trace_id: str | None = None,
    ) -> SsoGrant | None:
        """Failed / cancelled task: the removal is still outstanding.

        We detach the task so the next reconcile tick opens a fresh one — a
        cancelled task must never read as "nothing to do". ``removal_status``
        is left alone (``pending`` stays pending, ``contradicted`` stays
        contradicted).
        """
        grant = await self._grant_for_task(task)
        if grant is None:
            return None
        if grant.removal_task_id == task.id:
            grant.removal_task_id = None
            grant.updated_at = datetime.now(UTC)
            await self.db.flush()
        await self.audit.log(
            trace_id=trace_id or f"provisioning_task:{task.id}",
            actor_type="system",
            actor_user_id=None,
            action="sso.grant.removal_task_detached",
            status="success",
            resource_type="sso_grant",
            resource_id=str(grant.id),
            detail={"task_id": str(task.id), "task_outcome": outcome},
        )
        return grant

    async def open_review_for_legacy(
        self,
        grant_ids: list[str],
        *,
        actor_user_id: uuid.UUID,
        trace_id: str,
    ) -> dict[str, Any]:
        """Admin action: turn ``unverified_legacy`` rows into real work."""
        opened: list[str] = []
        skipped: list[dict[str, str]] = []
        for gid in grant_ids:
            grant = await self.db.get(SsoGrant, str(gid))
            if grant is None:
                skipped.append({"grant_id": str(gid), "reason": "not_found"})
                continue
            if grant.status != SSO_GRANT_STATUS_REVOKED:
                skipped.append({"grant_id": str(gid), "reason": "not_revoked"})
                continue
            if grant.removal_status not in (REMOVAL_UNVERIFIED_LEGACY, REMOVAL_PENDING):
                skipped.append({"grant_id": str(gid), "reason": grant.removal_status})
                continue
            if await self._open_task_for(grant) is not None:
                skipped.append({"grant_id": str(gid), "reason": "task_already_open"})
                continue
            await self.open_task(
                grant, source=SOURCE_MANUAL, actor_user_id=actor_user_id,
                trace_id=trace_id, reason="legacy review",
            )
            opened.append(str(gid))
        return {"opened": opened, "skipped": skipped}

    # ── Verification (Google Admin Reports · SAML audit) ───────────────

    async def verify_grant(
        self,
        grant: SsoGrant,
        *,
        settings: RemovalSettings,
        now: datetime,
        driver: Any | None,
        trace_id: str,
        force: bool = False,
    ) -> dict[str, Any]:
        """One evidence check. Returns a summary dict; mutates the grant.

        Outcomes (``result``):
          contradicted     a SAML login for this app after ``revoked_at``
          verified         account gone at IdP, or name-aligned history with
                           no post-revoke login ≥ verified_after_days after
                           confirmation
          still_open       aligned, no login, but not yet long enough since
                           confirmation (or not confirmed yet)
          unverifiable     app name never matched any IdP login → we cannot
                           prove absence (``removal_check_error`` set)
          skipped_*        not a Google grant / no driver / checked recently
          error            Reports API failed (scope, quota); error recorded
        """
        if grant.status != SSO_GRANT_STATUS_REVOKED or grant.revoked_at is None:
            return {"result": "skipped_not_revoked"}
        if grant.removal_status in (REMOVAL_VERIFIED,):
            return {"result": "skipped_terminal"}
        if not _is_google(grant):
            return {"result": "skipped_provider"}
        if driver is None:
            grant.removal_check_error = "google_not_configured"
            grant.removal_last_checked_at = now
            return {"result": "skipped_no_driver"}
        if (
            not force
            and grant.removal_last_checked_at is not None
            # ``_aware`` is not decoration. MySQL hands this column back
            # NAIVE, and ``now`` is aware, so the bare subtraction raises
            # ``TypeError`` — but only from the SECOND check onwards, once
            # the value has made a round trip through the database. The
            # first pass writes an aware value in memory and compares
            # fine, which is exactly why this survived review and the
            # tests: every other timestamp in this module is already
            # coerced, this one was not. Effect: for every grant that had
            # been checked once, the whole Google verification half
            # raised inside the per-grant savepoint and was counted as a
            # skipped error, forever.
            and now - _aware(grant.removal_last_checked_at) < _VERIFY_FLOOR
        ):
            return {"result": "skipped_recent"}

        holder = await self.db.get(IdentityPerson, uuid.UUID(str(grant.person_id)))
        holder_email = (holder.work_email or "").strip().lower() if holder else ""
        if not holder_email:
            grant.removal_check_error = "holder_email_missing"
            grant.removal_last_checked_at = now
            return {"result": "unverifiable", "reason": "holder_email_missing"}
        app_name = await self._idp_app_name(grant)

        from app.integrations.drivers.google_workspace import (  # noqa: PLC0415
            GoogleApiError,
        )

        revoked_at = _aware(grant.revoked_at)
        try:
            post = await driver.list_saml_login_events(
                user_email=holder_email, start_time=revoked_at,
            )
        except GoogleApiError as exc:
            grant.removal_last_checked_at = now
            if exc.reason == "user_not_found":
                # The IdP account itself is gone — no SAML app can be used.
                return await self._mark_verified(
                    grant, now=now, evidence="idp_account_absent", trace_id=trace_id,
                )
            grant.removal_check_error = (exc.reason or f"http_{exc.status}")[:255]
            return {"result": "error", "reason": grant.removal_check_error}

        matching = [e for e in post if _norm(e.get("application_name")) == _norm(app_name)]
        grant.removal_last_checked_at = now
        if matching:
            last_login = max(_aware(e["time"]) for e in matching)
            return await self._mark_contradicted(
                grant, last_login=last_login, now=now, trace_id=trace_id,
            )

        # No post-revoke login. Before we call that evidence, make sure our
        # app name is one the IdP has ever used for this person — otherwise
        # "no login" is what a misnamed app looks like too.
        try:
            pre = await driver.list_saml_login_events(
                user_email=holder_email,
                start_time=revoked_at - _ALIGNMENT_LOOKBACK,
                end_time=revoked_at,
            )
        except GoogleApiError as exc:
            grant.removal_check_error = (exc.reason or f"http_{exc.status}")[:255]
            return {"result": "error", "reason": grant.removal_check_error}
        aligned = any(_norm(e.get("application_name")) == _norm(app_name) for e in pre)
        if not aligned:
            grant.removal_check_error = "app_name_unmatched"
            return {"result": "unverifiable", "reason": "app_name_unmatched"}

        grant.removal_check_error = None
        if (
            grant.removal_status == REMOVAL_CONFIRMED
            and grant.removal_confirmed_at is not None
            and now - _aware(grant.removal_confirmed_at)
            >= timedelta(days=settings.verified_after_days)
        ):
            return await self._mark_verified(
                grant, now=now, evidence="no_saml_login_since_revoke", trace_id=trace_id,
            )
        return {"result": "still_open"}

    # ── Daily reconcile (cron body) ────────────────────────────────────

    async def reconcile(
        self,
        *,
        now: datetime,
        settings: RemovalSettings,
        driver: Any | None,
        trace_id: str,
    ) -> dict[str, Any]:
        summary: dict[str, Any] = {
            "scanned": 0, "tasks_opened": 0, "reminded": 0,
            "incidents_opened": 0, "verified": 0, "contradicted": 0,
            "unverifiable": 0, "verify_errors": 0, "verify_skipped": 0,
            "errors": [],
        }
        rows = list((await self.db.execute(
            select(SsoGrant).where(
                SsoGrant.status == SSO_GRANT_STATUS_REVOKED,
                SsoGrant.removal_status.in_(
                    (REMOVAL_PENDING, REMOVAL_CONTRADICTED, REMOVAL_CONFIRMED),
                ),
            ).order_by(SsoGrant.revoked_at.asc())
        )).scalars().all())
        verify_budget = _MAX_VERIFY_PER_RUN
        for grant in rows:
            summary["scanned"] += 1
            # Read the id BEFORE the savepoint: a rollback expires every
            # object the body dirtied, and an expired attribute read outside
            # an await raises ``MissingGreenlet``.
            grant_id = grant.id
            # SAVEPOINT, not a bare ``except``: ``open_task`` /
            # ``_send_reminder`` / ``_open_incident`` / ``verify_grant`` all
            # write. A swallowed flush failure deactivates the whole
            # transaction, so "one grant must not stop the sweep" was a lie —
            # every REMAINING grant then failed with ``PendingRollbackError``
            # and the reconcile silently stopped opening removal tasks.
            async with best_effort(
                self.db,
                label=f"removal reconcile for sso grant {grant_id}",
                log=logger,
                log_level=logging.DEBUG,  # logged with a traceback below
            ) as step:
                if grant.removal_status in (REMOVAL_PENDING, REMOVAL_CONTRADICTED):
                    task = await self._open_task_for(grant)
                    if task is None:
                        task = await self.open_task(
                            grant, source=SOURCE_AUDIT, actor_user_id=None,
                            trace_id=trace_id, reason="reconcile: no open task",
                        )
                        summary["tasks_opened"] += 1
                    if grant.revoked_at is not None and reminder_due(
                        revoked_at=_aware(grant.revoked_at),
                        reminder_count=grant.removal_reminder_count,
                        last_reminded_at=(
                            _aware(grant.removal_last_reminded_at)
                            if grant.removal_last_reminded_at else None
                        ),
                        now=now,
                        ladder=settings.reminder_ladder_days,
                    ):
                        await self._send_reminder(grant, task, now=now, trace_id=trace_id)
                        summary["reminded"] += 1
                    if (
                        grant.revoked_at is not None
                        and (now - _aware(grant.revoked_at)).days
                        >= settings.incident_after_days
                    ):
                        created = await self._open_incident(grant, now=now)
                        if created:
                            summary["incidents_opened"] += 1
                if settings.verify_via_google_reports and verify_budget > 0 and _is_google(grant):
                    verify_budget -= 1
                    outcome = await self.verify_grant(
                        grant, settings=settings, now=now, driver=driver,
                        trace_id=trace_id,
                    )
                    res = outcome.get("result", "")
                    if res == "verified":
                        summary["verified"] += 1
                    elif res == "contradicted":
                        summary["contradicted"] += 1
                    elif res == "unverifiable":
                        summary["unverifiable"] += 1
                    elif res == "error":
                        summary["verify_errors"] += 1
                    elif res.startswith("skipped"):
                        summary["verify_skipped"] += 1
                await self.db.flush()
            if not step.ok:
                logger.warning(
                    "sso.removal_reconcile: grant %s failed: %s",
                    grant_id, step.reason(), exc_info=step.error,
                )
                summary["errors"].append({
                    "grant_id": str(grant_id),
                    "error": step.reason(max_len=200) or "",
                })
        return summary

    # ── Read side ──────────────────────────────────────────────────────

    async def summary(self, *, now: datetime | None = None) -> dict[str, Any]:
        """Counts for the admin ledger's status strip."""
        now = now or datetime.now(UTC)
        rows = (await self.db.execute(
            select(SsoGrant.removal_status, func.count())
            .where(SsoGrant.status == SSO_GRANT_STATUS_REVOKED)
            .group_by(SsoGrant.removal_status)
        )).all()
        counts = {status: 0 for status in (
            REMOVAL_PENDING, REMOVAL_CONFIRMED, REMOVAL_VERIFIED,
            REMOVAL_CONTRADICTED, REMOVAL_UNVERIFIED_LEGACY,
        )}
        for status, n in rows:
            if status in counts:
                counts[status] = int(n)
        settings = await load_removal_settings(self.db)
        overdue_cutoff = now - timedelta(days=settings.incident_after_days)
        overdue = (await self.db.execute(
            select(func.count()).select_from(SsoGrant).where(
                SsoGrant.status == SSO_GRANT_STATUS_REVOKED,
                SsoGrant.removal_status.in_((REMOVAL_PENDING, REMOVAL_CONTRADICTED)),
                SsoGrant.revoked_at <= overdue_cutoff,
            )
        )).scalar() or 0
        check_errors = (await self.db.execute(
            select(func.count()).select_from(SsoGrant).where(
                SsoGrant.status == SSO_GRANT_STATUS_REVOKED,
                SsoGrant.removal_status.in_(REMOVAL_OPEN_STATUSES + (REMOVAL_CONFIRMED,)),
                SsoGrant.removal_check_error.isnot(None),
            )
        )).scalar() or 0
        return {
            "counts": counts,
            "open": sum(counts[s] for s in REMOVAL_OPEN_STATUSES),
            "overdue": int(overdue),
            "check_errors": int(check_errors),
            "settings": settings.to_setting(),
        }

    @staticmethod
    def serialize(grant: SsoGrant) -> dict[str, Any]:
        def _iso(dt: datetime | None) -> str | None:
            return dt.isoformat() if dt else None
        return {
            "removal_status": grant.removal_status,
            "removal_task_id": str(grant.removal_task_id) if grant.removal_task_id else None,
            "removal_confirmed_at": _iso(grant.removal_confirmed_at),
            "removal_confirmed_by_person_id": (
                str(grant.removal_confirmed_by_person_id)
                if grant.removal_confirmed_by_person_id else None
            ),
            "removal_external_ref": grant.removal_external_ref,
            "removal_verified_at": _iso(grant.removal_verified_at),
            "removal_last_checked_at": _iso(grant.removal_last_checked_at),
            "removal_last_login_at": _iso(grant.removal_last_login_at),
            "removal_check_error": grant.removal_check_error,
            "removal_reminder_count": grant.removal_reminder_count,
            "removal_last_reminded_at": _iso(grant.removal_last_reminded_at),
        }

    # ── Internals ──────────────────────────────────────────────────────

    async def _open_task_for(self, grant: SsoGrant) -> ProvisioningTask | None:
        """The grant's linked task if it is still open; otherwise any other
        open sso_grant revoke task for the same target (a task opened before
        the link column existed, or after a cancel detached it)."""
        if grant.removal_task_id is not None:
            task = await self.db.get(ProvisioningTask, grant.removal_task_id)
            if task is not None and task.status not in TERMINAL_TASK_STATUSES:
                return task
        target = uuid.UUID(str(grant.id))
        task = (await self.db.execute(
            select(ProvisioningTask).where(
                ProvisioningTask.target_type == TARGET_SSO_GRANT,
                ProvisioningTask.target_id == target,
                ProvisioningTask.action == ACTION_REVOKE,
                ProvisioningTask.status.notin_(list(TERMINAL_TASK_STATUSES)),
            ).order_by(ProvisioningTask.created_at.desc())
        )).scalars().first()
        if task is not None and grant.removal_task_id != task.id:
            grant.removal_task_id = task.id
        return task

    async def _grant_for_task(self, task: ProvisioningTask) -> SsoGrant | None:
        if task.target_type != TARGET_SSO_GRANT or task.target_id is None:
            return None
        return await self.db.get(SsoGrant, str(task.target_id))

    async def _active_owner_id(self, grant: SsoGrant) -> uuid.UUID | None:
        if not grant.system_owner_person_id:
            return None
        try:
            owner_id = uuid.UUID(str(grant.system_owner_person_id))
        except (TypeError, ValueError):
            return None
        owner = await self.db.get(IdentityPerson, owner_id)
        if (
            owner is None
            or owner.deleted_at is not None
            or (owner.employment_status or "").lower() != "active"
            or not owner.work_email
        ):
            return None
        return owner.id

    async def _idp_app_name(self, grant: SsoGrant) -> str:
        """What the IdP calls this app: the catalog row's display name when
        the grant is linked to one, else the free-text system name."""
        if grant.created_app_id:
            app = await self.db.get(SsoApp, grant.created_app_id)
            if app is not None and app.display_name:
                return app.display_name
        return grant.system_name

    async def _mark_verified(
        self, grant: SsoGrant, *, now: datetime, evidence: str, trace_id: str,
    ) -> dict[str, Any]:
        grant.removal_status = REMOVAL_VERIFIED
        grant.removal_verified_at = now
        grant.removal_check_error = None
        grant.updated_at = now
        await self.db.flush()
        await self._resolve_incident(grant, note=f"verified: {evidence}", trace_id=trace_id)
        await self.audit.log(
            trace_id=trace_id,
            actor_type="system",
            actor_user_id=None,
            action="sso.grant.removal_verified",
            status="success",
            resource_type="sso_grant",
            resource_id=str(grant.id),
            detail={"evidence": evidence},
        )
        return {"result": "verified", "evidence": evidence}

    async def _mark_contradicted(
        self, grant: SsoGrant, *, last_login: datetime, now: datetime, trace_id: str,
    ) -> dict[str, Any]:
        newly = (
            grant.removal_status != REMOVAL_CONTRADICTED
            or grant.removal_last_login_at is None
            or _aware(grant.removal_last_login_at) < last_login
        )
        grant.removal_status = REMOVAL_CONTRADICTED
        grant.removal_last_login_at = last_login
        grant.removal_check_error = None
        grant.updated_at = now
        # A confirmation that the IdP disproves is void: clear it so the
        # ledger does not show "confirmed by X" next to a live login.
        grant.removal_confirmed_at = None
        grant.removal_confirmed_by_person_id = None
        grant.removal_external_ref = None
        await self.db.flush()
        task = await self._open_task_for(grant)
        if task is None:
            task = await self.open_task(
                grant, source=SOURCE_AUDIT, actor_user_id=None, trace_id=trace_id,
                reason="contradicted by IdP login audit", notify=False,
            )
        await self.audit.log(
            trace_id=trace_id,
            actor_type="system",
            actor_user_id=None,
            action="sso.grant.removal_contradicted",
            status="failed",
            resource_type="sso_grant",
            resource_id=str(grant.id),
            detail={"last_login_at": last_login.isoformat(), "task_id": str(task.id)},
        )
        if newly:
            await self._send_contradiction(grant, task, last_login=last_login, trace_id=trace_id)
        return {"result": "contradicted", "last_login_at": last_login.isoformat()}

    async def _open_incident(self, grant: SsoGrant, *, now: datetime) -> bool:
        holder = await self.db.get(IdentityPerson, uuid.UUID(str(grant.person_id)))
        label = (holder.display_name or holder.work_email) if holder else str(grant.person_id)
        age_days = (now - _aware(grant.revoked_at)).days if grant.revoked_at else 0
        _row, created = await ComplianceIncidentService(self.db).create_or_upsert_incident(
            incident_type=INCIDENT_TYPE_SSO_REMOVAL_OVERDUE,
            severity=SEVERITY_WARNING,
            person_id=holder.id if holder else None,
            resource_type="sso_grant",
            resource_id=str(grant.id),
            resource_label=grant.system_name,
            title=(
                f"SSO 撤销未在 IdP 侧落地 {age_days} 天:{grant.system_name} · {label}"
            ),
            detail={
                "grant_id": str(grant.id),
                "system_name": grant.system_name,
                "sso_provider": grant.sso_provider,
                "revoked_at": grant.revoked_at.isoformat() if grant.revoked_at else None,
                "removal_status": grant.removal_status,
                "removal_task_id": (
                    str(grant.removal_task_id) if grant.removal_task_id else None
                ),
                "reminder_count": grant.removal_reminder_count,
            },
        )
        return bool(created)

    async def _resolve_incident(self, grant: SsoGrant, *, note: str, trace_id: str) -> None:
        rows = (await self.db.execute(
            select(ComplianceIncident).where(
                ComplianceIncident.incident_type == INCIDENT_TYPE_SSO_REMOVAL_OVERDUE,
                ComplianceIncident.resource_type == "sso_grant",
                ComplianceIncident.resource_id == uuid.UUID(str(grant.id)),
                ComplianceIncident.status.in_([INCIDENT_OPEN, INCIDENT_ACCEPTED]),
            )
        )).scalars().all()
        svc = ComplianceIncidentService(self.db)
        for row in rows:
            if row.status != INCIDENT_OPEN:
                continue
            await svc.resolve_incident(
                row.id, resolved_by_person_id=None, resolution_note=note,
                trace_id=trace_id,
            )

    async def _it_admins(self, *, exclude: set[uuid.UUID]) -> list[IdentityPerson]:
        from app.auth.models import AuthRole, AuthUserRole  # noqa: PLC0415

        ids = list((await self.db.execute(
            select(AuthUserRole.user_id)
            .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
            .where(AuthRole.name.in_(("global_admin", "it_admin")))
        )).scalars().all())
        out: list[IdentityPerson] = []
        for pid in ids:
            if pid in exclude:
                continue
            p = await self.db.get(IdentityPerson, pid)
            if (
                p is not None
                and p.deleted_at is None
                and (p.employment_status or "").lower() == "active"
                and p.work_email
            ):
                out.append(p)
                exclude.add(pid)
        return out

    async def _recipients(
        self, grant: SsoGrant, task: ProvisioningTask, *, include_it: bool,
    ) -> list[IdentityPerson]:
        seen: set[uuid.UUID] = set()
        out: list[IdentityPerson] = []
        if task.assignee_id is not None:
            assignee = await self.db.get(IdentityPerson, task.assignee_id)
            if (
                assignee is not None
                and assignee.deleted_at is None
                and (assignee.employment_status or "").lower() == "active"
                and assignee.work_email
            ):
                out.append(assignee)
                seen.add(assignee.id)
        if include_it or not out:
            out.extend(await self._it_admins(exclude=seen))
        return out

    async def _send_reminder(
        self, grant: SsoGrant, task: ProvisioningTask, *, now: datetime, trace_id: str,
    ) -> None:
        # The first reminder goes to the assignee alone; from the second on
        # IT is copied (the 7-day rung) — "escalate by widening, not by
        # shouting louder at the same person".
        include_it = grant.removal_reminder_count >= 1
        recipients = await self._recipients(grant, task, include_it=include_it)
        holder = await self.db.get(IdentityPerson, uuid.UUID(str(grant.person_id)))
        holder_label = (holder.display_name or holder.work_email) if holder else str(grant.person_id)
        age_days = (now - _aware(grant.revoked_at)).days if grant.revoked_at else 0
        rung = grant.removal_reminder_count + 1
        async with best_effort(
            self.db, label="sso.removal reminder fan-out", log=logger,
        ) as step:
            await self._dispatch_cards(
                recipients, grant=grant, task=task, kind="reminder",
                holder_label=holder_label, age_days=age_days, rung=rung,
                last_login=None, trace_id=trace_id,
            )
        if step.ok:
            grant.removal_reminder_count = rung
            grant.removal_last_reminded_at = now
            grant.updated_at = now
            await self.db.flush()
        else:
            logger.warning(
                "sso.removal reminder for grant %s not sent: %s", grant.id, step.reason(),
            )

    async def _send_contradiction(
        self, grant: SsoGrant, task: ProvisioningTask, *, last_login: datetime, trace_id: str,
    ) -> None:
        recipients = await self._recipients(grant, task, include_it=True)
        holder = await self.db.get(IdentityPerson, uuid.UUID(str(grant.person_id)))
        holder_label = (holder.display_name or holder.work_email) if holder else str(grant.person_id)
        age_days = (datetime.now(UTC) - _aware(grant.revoked_at)).days if grant.revoked_at else 0
        async with best_effort(
            self.db, label="sso.removal contradiction fan-out", log=logger,
        ) as step:
            await self._dispatch_cards(
                recipients, grant=grant, task=task, kind="contradicted",
                holder_label=holder_label, age_days=age_days, rung=0,
                last_login=last_login, trace_id=trace_id,
            )
        if not step.ok:
            logger.warning(
                "sso.removal contradiction card for grant %s not sent: %s",
                grant.id, step.reason(),
            )

    async def _dispatch_cards(
        self,
        recipients: list[IdentityPerson],
        *,
        grant: SsoGrant,
        task: ProvisioningTask,
        kind: str,
        holder_label: str,
        age_days: int,
        rung: int,
        last_login: datetime | None,
        trace_id: str,
    ) -> None:
        from app.core.security import create_action_token  # noqa: PLC0415
        from app.domains.provisioning.models import (  # noqa: PLC0415
            ACTION_TOKEN_PURPOSE_MANUAL_COMPLETE,
        )
        from app.notifications.service import (  # noqa: PLC0415
            CardAction,
            CardStat,
            LocalizedText,
            NotificationEnvelope,
            NotificationService,
        )

        portal_url = str(await get_platform_setting(
            "app.frontend_base_url", default="", session=self.db,
        ) or "").rstrip("/")
        ledger_url = "/admin/sso-apps?tab=grants&removal=open"
        svc = NotificationService(self.db)
        last_login_iso = last_login.isoformat(timespec="minutes") if last_login else ""
        for recipient in recipients:
            lang = _lang(recipient)
            if kind == "reminder":
                title = _pick(
                    lang,
                    zh=f"⏳ SSO 撤销待落地(第 {rung} 次提醒):{grant.system_name} · {holder_label}",
                    en=f"⏳ SSO removal still outstanding (reminder #{rung}): {grant.system_name} · {holder_label}",
                    ja=f"⏳ SSO 解除が未完了(リマインド {rung} 回目):{grant.system_name} · {holder_label}",
                )
                subtitle = LocalizedText(
                    zh=f"Helix 已于 {age_days} 天前撤销该授权,但目标系统里还没有人确认停用。",
                    en=f"Helix revoked this grant {age_days} days ago; nobody has confirmed it was disabled in the target system.",
                    ja=f"Helix は {age_days} 日前にこの権限を取り消しましたが、対象システム側での無効化はまだ確認されていません。",
                )
                body = LocalizedText(
                    zh="请到 IdP / 目标系统控制台停用该账号的 SSO 访问,然后回到 Helix 填写凭证(工单号、控制台链接或说明)完成任务。",
                    en="Disable this person's SSO access in the IdP / target console, then come back and complete the task with evidence (ticket id, console URL or a note).",
                    ja="IdP / 対象システムのコンソールでこの人の SSO アクセスを無効化し、Helix で根拠(チケット番号・コンソール URL・メモ)を添えてタスクを完了してください。",
                )
                template_key = TEMPLATE_REMINDER
            else:
                title = _pick(
                    lang,
                    zh=f"🚨 撤销后仍在登录:{grant.system_name} · {holder_label}",
                    en=f"🚨 Login observed after revoke: {grant.system_name} · {holder_label}",
                    ja=f"🚨 取り消し後もログインあり:{grant.system_name} · {holder_label}",
                )
                subtitle = LocalizedText(
                    zh=f"Google 登录审计显示该账号在撤销之后仍于 {last_login_iso} 通过 SAML 登录了此系统。",
                    en=f"Google's login audit shows a SAML login to this app at {last_login_iso}, after the Helix revoke.",
                    ja=f"Google のログイン監査では、取り消し後の {last_login_iso} にこのアプリへ SAML ログインが記録されています。",
                )
                body = LocalizedText(
                    zh="之前的「已停用」确认已作废。请立即在 IdP / 目标系统停用,并重新完成任务。",
                    en="The earlier 'disabled' confirmation has been voided. Disable the access in the IdP / target system now and complete the task again.",
                    ja="以前の「無効化済み」確認は無効になりました。至急 IdP / 対象システムで無効化し、タスクを再度完了してください。",
                )
                template_key = TEMPLATE_CONTRADICTED

            actions: list[CardAction] = []
            if portal_url:
                token = create_action_token(
                    purpose=ACTION_TOKEN_PURPOSE_MANUAL_COMPLETE,
                    subject_id=str(recipient.id),
                    resource_id=str(task.id),
                )
                actions.append(CardAction(
                    label=LocalizedText(
                        zh="✅ 已在目标系统停用,填写凭证",
                        en="✅ Disabled it — record evidence",
                        ja="✅ 無効化済み・根拠を記録",
                    ),
                    url=f"{portal_url}/api/v1/provisioning/quick-complete?token={token}",
                    style="primary" if kind == "reminder" else "danger",
                ))
            actions.append(CardAction(
                label=LocalizedText(zh="打开 SSO 台账", en="Open the SSO ledger", ja="SSO 台帳を開く"),
                url=f"{portal_url}{ledger_url}" if portal_url else ledger_url,
                style="default",
            ))
            stats = [
                CardStat(
                    label=LocalizedText(zh="系统", en="System", ja="システム"),
                    value=grant.system_name,
                ),
                CardStat(
                    label=LocalizedText(zh="持有人", en="Holder", ja="保持者"),
                    value=holder_label,
                ),
                CardStat(
                    label=LocalizedText(zh="撤销已过", en="Revoked", ja="取り消しから"),
                    value=LocalizedText(zh=f"{age_days} 天", en=f"{age_days} d", ja=f"{age_days} 日"),
                ),
                CardStat(
                    label=LocalizedText(zh="IdP", en="IdP", ja="IdP"),
                    value=grant.sso_provider or "—",
                ),
            ]
            await svc.dispatch(
                NotificationEnvelope(
                    recipient=recipient,
                    title=title,
                    body_text=resolve_plain(body, lang),
                    channels=["feishu", "email"],
                    template_key=template_key,
                    context={
                        "system_name": grant.system_name,
                        "holder": holder_label,
                        "age_days": str(age_days),
                        "rung": str(rung),
                        "last_login_at": last_login_iso,
                        "task_id": str(task.id),
                    },
                    url=ledger_url,
                    card_subtitle=subtitle,
                    card_stats=stats,
                    card_actions=actions,
                    related_resource_type="sso_grant",
                    related_resource_id=uuid.UUID(str(grant.id)),
                ),
                trace_id=trace_id,
                actor_user_id=None,
            )


def resolve_plain(text: Any, lang: str) -> str:
    """LocalizedText → the recipient's variant, for the plain ``body_text``."""
    zh = getattr(text, "zh", None)
    if zh is None:
        return str(text)
    return _pick(lang, zh=text.zh, en=text.en, ja=text.ja)


def _is_google(grant: SsoGrant) -> bool:
    return (grant.sso_provider or "").strip().lower() in ("google", "google_workspace")


def _aware(dt: datetime) -> datetime:
    """DB timestamps may come back naive (MySQL DATETIME); treat as UTC."""
    return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC)


async def google_driver_for_org(db: AsyncSession) -> Any | None:
    """The org-level Google Workspace driver, or ``None`` when not configured.

    Mirrors ``email_group.service._get_driver`` without raising: the reconcile
    cron records "not configured" on the grant instead of failing the sweep.
    """
    from app.integrations.drivers.google_workspace import (  # noqa: PLC0415
        GoogleWorkspaceDriver,
    )
    from app.integrations.models import Integration  # noqa: PLC0415
    from app.integrations.service import _resolve  # noqa: PLC0415

    row = (await db.execute(
        select(Integration).where(
            Integration.kind == "software_provisioning",
            Integration.provider == "google_workspace",
            Integration.enabled.is_(True),
        ).order_by(Integration.is_primary.desc(), Integration.created_at)
    )).scalars().first()
    if row is None:
        return None
    resolved = await _resolve(db, row)
    return GoogleWorkspaceDriver(integration_id=str(row.id), config=resolved.config)


__all__ = [
    "INCIDENT_TYPE_SSO_REMOVAL_OVERDUE",
    "RemovalSettings",
    "SETTING_KEY",
    "SsoRemovalService",
    "TEMPLATE_CONTRADICTED",
    "TEMPLATE_REMINDER",
    "google_driver_for_org",
    "load_removal_settings",
    "reminder_due",
]