"""Software permission service.

Catalog CRUD is straightforward. Request flow:

1. Employee picks a permission → submits a PermissionRequest.
2. ``requires_approval=False``: create a UserPermission in ``pending`` and
   queue the provisioning task. The reconcile path flips it to ``active``
   only when the cloud-side grant actually completes.
3. ``requires_approval=True``: self-service is REFUSED (see
   ``request_permission``). Unlike the sso / apikey / ai_tool_application
   domains, the permission domain has NO approval-callback wiring — there
   is no Feishu instance, no admin inbox item, and no activation path for
   an approval-gated permission request. It is granted only via
   ``admin_grant`` (POST /permissions/grants, IT-only).

Historical note: the old code created a ``pending`` UserPermission on a
``requires_approval=True`` request that then sat forever with no grant
path — a silent dead-end. batch-6 (2026-07-07) closed it with the
fail-closed guard in ``request_permission``. Prod at the time had an empty
``software_permissions`` catalog and zero permission requests/grants ever,
so no live rows were affected. If a real Feishu-routed permission-approval
flow is built later, replace the guard with create-instance + callback
wiring (mirror the sso/apikey pattern) and record it in DECISIONS.
"""
from __future__ import annotations

from datetime import UTC, datetime, timedelta
from uuid import UUID

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

from app.audit.service import AuditService
from app.core.exceptions import AppException
from app.core.grant_eligibility import is_blocked_from_new_grant
from app.domains.asset.models import AssetSoftwareCatalog
from app.domains.permission.models import (
    PERM_ACTIVE,
    PERM_CANCELLED,
    PERM_PENDING,
    PERM_PENDING_APPROVAL,
    PERM_REJECTED,
    SoftwarePermission,
    UserPermission,
)
from app.domains.permission.schemas import (
    PermissionCreate,
    PermissionResponse,
    PermissionUpdate,
    UserPermissionResponse,
)
from app.domains.provisioning.models import (
    ACTION_GRANT,
    ACTION_REVOKE,
    SOURCE_REQUEST,
    ProvisioningTask,
)
from app.domains.provisioning.models import (
    STATUS_COMPLETED as TASK_COMPLETED,
)
from app.domains.provisioning.schemas import ProvisioningTaskCreate
from app.domains.provisioning.service import ProvisioningService
from app.identity.person_models import IdentityPerson


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

    # ── Catalog CRUD ──

    async def list_permissions(
        self, *, software_id: UUID | None = None, active_only: bool = False,
    ) -> list[PermissionResponse]:
        q = select(SoftwarePermission)
        if software_id:
            q = q.where(SoftwarePermission.software_id == software_id)
        if active_only:
            q = q.where(SoftwarePermission.is_active.is_(True))
        # Hide soft-deleted rows from the admin catalog list. Historical
        # user_permissions rows still reference them via FK; they only
        # disappear from the list / dropdown, not from audit reads.
        if hasattr(SoftwarePermission, "deleted_at"):
            q = q.where(SoftwarePermission.deleted_at.is_(None))
        rows = list(
            (await self.db.execute(q.order_by(SoftwarePermission.sort_order))).scalars().all()
        )
        return await self._enrich_permissions(rows)

    async def create_permission(
        self, body: PermissionCreate, *, trace_id: str, actor_user_id: UUID,
    ) -> PermissionResponse:
        # Uniqueness (software_id, permission_key). There is NO DB unique
        # constraint on this pair — this app-level guard is the only one —
        # so it must consider soft-deleted rows too. Two cases:
        #   * an ACTIVE row exists  -> reject (PERMISSION_EXISTS)
        #   * only a SOFT-DELETED row exists -> resurrect it, rather than
        #     insert a duplicate key. Pre-fix the guard matched the
        #     soft-deleted row and rejected, leaving the key permanently
        #     un-recreatable (the row is hidden from the catalog UI, so the
        #     operator sees PERMISSION_EXISTS for a row they can't find and
        #     has no re-enable path).
        existing = (
            await self.db.execute(
                select(SoftwarePermission).where(
                    and_(
                        SoftwarePermission.software_id == body.software_id,
                        SoftwarePermission.permission_key == body.permission_key,
                    )
                )
            )
        ).scalar_one_or_none()
        if existing is not None and existing.deleted_at is None:
            raise AppException(
                code="PERMISSION_EXISTS",
                message="A permission with this key already exists for this software",
                status_code=400,
            )

        if existing is not None:
            # Resurrect the soft-deleted catalog entry in place so the
            # stable permission id (referenced by historical UserPermission
            # rows) is preserved.
            row = existing
            row.deleted_at = None
            row.display_name = body.display_name
            row.description = body.description
            row.requires_approval = body.requires_approval
            row.is_active = body.is_active
            row.sort_order = body.sort_order
            action = "permission.catalog_restored"
        else:
            row = SoftwarePermission(
                software_id=body.software_id,
                permission_key=body.permission_key,
                display_name=body.display_name,
                description=body.description,
                requires_approval=body.requires_approval,
                is_active=body.is_active,
                sort_order=body.sort_order,
            )
            self.db.add(row)
            action = "permission.catalog_created"
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action=action,
            status="success",
            resource_type="software_permission",
            resource_id=str(row.id),
            detail={"permission_key": body.permission_key},
        )
        return (await self._enrich_permissions([row]))[0]

    async def update_permission(
        self, permission_id: UUID, body: PermissionUpdate,
        *, trace_id: str, actor_user_id: UUID,
    ) -> PermissionResponse:
        row = await self._get_permission(permission_id)
        changes: dict = {}
        if body.display_name is not None:
            row.display_name = body.display_name
            changes["display_name"] = body.display_name
        if body.description is not None:
            row.description = body.description
            changes["description_updated"] = True
        if body.requires_approval is not None:
            row.requires_approval = body.requires_approval
            changes["requires_approval"] = body.requires_approval
        if body.is_active is not None:
            row.is_active = body.is_active
            changes["is_active"] = body.is_active
        if body.sort_order is not None:
            row.sort_order = body.sort_order
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="permission.catalog_updated",
            status="success",
            resource_type="software_permission",
            resource_id=str(row.id),
            detail=changes,
        )
        return (await self._enrich_permissions([row]))[0]

    async def delete_permission(
        self, permission_id: UUID, *, trace_id: str, actor_user_id: UUID,
    ) -> None:
        """Soft-delete the catalog row.

        Hard delete would orphan FK references on every historical
        ``UserPermission`` row whose status is ``revoked`` — the
        catalog id is the only handle for "what permission did Jane
        hold in 2026-Q1?" Soft delete (``is_active=False`` + the
        existing pre-launch ``deleted_at`` column) preserves the
        history while removing the row from the catalog UI and
        blocking new grants. Reactivation flips ``is_active`` back
        on without losing the ledger.

        Refuses delete when active/pending grants exist — caller must
        revoke them first so the audit story is "every grant was
        explicitly revoked, then the catalog row was retired", not
        "the catalog vanished and the grants dangled".
        """
        row = await self._get_permission(permission_id)
        # Check: reject if there are active user_permissions
        active = (
            await self.db.execute(
                select(func.count()).select_from(UserPermission).where(
                    and_(
                        UserPermission.permission_id == row.id,
                        UserPermission.status.in_([PERM_ACTIVE, PERM_PENDING]),
                    )
                )
            )
        ).scalar() or 0
        if active > 0:
            raise AppException(
                code="PERMISSION_IN_USE",
                message=f"Cannot delete: {active} active grants exist. Revoke them first.",
                status_code=400,
            )
        # Soft delete: preserves FK references from historical
        # user_permissions while hiding the row from the catalog UI.
        # ``is_active=False`` is the existing "hidden but referenced"
        # signal already honoured by list_permissions(active_only=True)
        # and request_permission's ``PERMISSION_INACTIVE`` guard.
        row.is_active = False
        if hasattr(row, "deleted_at"):
            row.deleted_at = datetime.now(UTC)
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="permission.catalog_deleted",
            status="success",
            resource_type="software_permission",
            resource_id=str(permission_id),
            detail={"soft_delete": True},
        )

    # ── User grants ──

    async def request_permission(
        self,
        requesting_user_id: UUID,
        permission_id: UUID,
        *,
        reason: str,
        trace_id: str,
    ) -> UserPermission:
        """Request a permission for the current user.

        Wave 1D pre-launch hardening (2026-05-05):
        - Race-safe under double-click: a UNIQUE(user_id, permission_id)
          constraint backs this; we SELECT existing across ALL statuses,
          UPDATE in place if the prior row is in a terminal state, and
          catch IntegrityError as a 409 in case the unique-index rejects
          a concurrent INSERT.
        - For ``requires_approval=False`` paths the row is left in
          ``PERM_PENDING`` rather than flipped to ``PERM_ACTIVE`` up
          front; the provisioning reconcile path
          (``ProvisioningService._on_task_complete``) flips PENDING →
          ACTIVE only when the cloud-side grant actually completes. Pre-
          fix UX showed "active" while the downstream task was still
          pending and could later fail, leaking a phantom-active status
          to the user.

        In-app approval (migration 606, 2026-09-05 decision — "不走飞书,
        管理员站内批"): a ``requires_approval=True`` permission creates a
        ``pending_approval`` row and notifies the software owner / IT
        contact (global admins when the software names nobody). The
        historical fail-closed refusal (batch-6, 2026-07-07) existed only
        because no approval consumer existed; :meth:`approve_request` /
        :meth:`reject_request` are that consumer now.
        """
        perm = await self._get_permission(permission_id)
        if not perm.is_active:
            raise AppException(
                code="PERMISSION_INACTIVE",
                message="This permission is no longer available",
                status_code=400,
            )
        awaiting_approval = bool(perm.requires_approval)

        # Look up by (user, permission) without status filter — UQ
        # uq_user_perm guarantees at most one such row.
        existing = (
            await self.db.execute(
                select(UserPermission).where(
                    and_(
                        UserPermission.user_id == requesting_user_id,
                        UserPermission.permission_id == permission_id,
                    )
                ).with_for_update()
            )
        ).scalar_one_or_none()

        if existing and existing.status in (PERM_PENDING_APPROVAL, PERM_PENDING, PERM_ACTIVE):
            raise AppException(
                code="PERMISSION_ALREADY_HELD",
                message=f"You already have a {existing.status} request for this permission",
                status_code=400,
            )

        async def _attach_provisioning(target_row: UserPermission) -> None:
            """Queue the cloud-side grant. Only reached for
            ``requires_approval=False`` permissions (approval-gated ones get
            their task from :meth:`approve_request`). ``target_row.status``
            is intentionally LEFT at PERM_PENDING so the reconcile path is
            the source of truth for ACTIVE."""
            task = await self.provisioning.create_task(ProvisioningTaskCreate(
                user_id=requesting_user_id,
                software_id=perm.software_id,
                action=ACTION_GRANT,
                target_type="permission",
                target_id=perm.id,
                source=SOURCE_REQUEST,
                context_data={
                    "permission_key": perm.permission_key,
                    "reason": reason,
                },
            ))
            target_row.provisioning_task_id = task.id

        initial_status = PERM_PENDING_APPROVAL if awaiting_approval else PERM_PENDING
        if existing:
            # In a terminal state (revoked / rejected / expired) — reuse
            # the row by RESETTING it to a fresh request rather than
            # INSERTing a duplicate that would trip the UQ.
            existing.status = initial_status
            existing.reason = reason
            existing.granted_at = None
            existing.revoked_at = None
            existing.external_ref = None
            existing.approval_request_id = None
            existing.provisioning_task_id = None
            existing.expires_at = None
            existing.approver_person_id = None
            existing.decided_at = None
            existing.decision_note = None
            existing.approval_reminder_count = 0
            existing.approval_last_reminded_at = None
            if not awaiting_approval:
                await _attach_provisioning(existing)
            await self.db.flush()
            row = existing
        else:
            row = UserPermission(
                user_id=requesting_user_id,
                permission_id=permission_id,
                status=initial_status,
                reason=reason,
            )
            if not awaiting_approval:
                await _attach_provisioning(row)
            self.db.add(row)
            try:
                await self.db.flush()
            except IntegrityError as exc:
                # Concurrent caller beat us to the INSERT. The UQ on
                # (user_id, permission_id) closes the residual race the
                # FOR UPDATE row-lock above can't cover (no row to lock
                # in the not-exists path).
                await self.db.rollback()
                raise AppException(
                    code="PERMISSION_ALREADY_HELD",
                    message=(
                        "A concurrent request created the same grant. "
                        "Refresh and check your existing requests."
                    ),
                    status_code=409,
                ) from exc

        approvers_notified = 0
        routed_to_it = False
        if awaiting_approval:
            from app.domains.permission.lifecycle import (  # noqa: PLC0415
                notify_approval_requested,
                resolve_approvers,
            )
            software = await self.db.get(AssetSoftwareCatalog, perm.software_id)
            requester = await self.db.get(IdentityPerson, requesting_user_id)
            approvers, routed_to_it = await resolve_approvers(self.db, software)
            if requester is not None:
                approvers_notified = await notify_approval_requested(
                    self.db, up=row, perm=perm, software=software, requester=requester,
                    approvers=approvers, routed_to_it=routed_to_it, trace_id=trace_id,
                )

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=requesting_user_id,
            action="permission.requested",
            status="success",
            resource_type="user_permission",
            resource_id=str(row.id),
            detail={
                "permission_id": str(permission_id),
                "requires_approval": perm.requires_approval,
                "initial_status": initial_status,
                "approvers_notified": approvers_notified,
                "routed_to_it": routed_to_it,
                "reason": reason[:500],
            },
        )
        return row

    # ── In-app approval (migration 606) ──

    async def _actor_is_admin(self, actor_user_id: UUID) -> bool:
        from app.auth.models import AuthRole, AuthUserRole  # noqa: PLC0415
        from app.core.rbac import GLOBAL_ADMIN_ROLE, has_role, normalize_role_names  # noqa: PLC0415

        rows = (await self.db.scalars(
            select(AuthRole.name)
            .join(AuthUserRole, AuthUserRole.role_id == AuthRole.id)
            .where(AuthUserRole.user_id == actor_user_id),
        )).all()
        return has_role(normalize_role_names(rows), GLOBAL_ADMIN_ROLE)

    async def can_decide(self, up: UserPermission, *, actor_user_id: UUID) -> bool:
        """Software owner / IT contact of the permission's software, or IT."""
        from app.domains.permission.lifecycle import software_approver_ids  # noqa: PLC0415

        perm = await self._get_permission(up.permission_id)
        software = await self.db.get(AssetSoftwareCatalog, perm.software_id)
        if actor_user_id in software_approver_ids(software):
            return True
        return await self._actor_is_admin(actor_user_id)

    async def _load_decision_context(
        self, up: UserPermission,
    ) -> tuple[SoftwarePermission, AssetSoftwareCatalog | None, IdentityPerson | None]:
        perm = await self._get_permission(up.permission_id)
        software = await self.db.get(AssetSoftwareCatalog, perm.software_id)
        requester = await self.db.get(IdentityPerson, up.user_id)
        return perm, software, requester

    async def approve_request(
        self,
        user_permission_id: UUID,
        *,
        actor_user_id: UUID,
        note: str | None,
        valid_days: int | None,
        trace_id: str,
    ) -> UserPermission:
        """Approver flips ``pending_approval`` → ``pending`` and queues the grant.

        The row is NOT ``active`` yet — the vendor-side grant runs from the
        provisioning task and its success hook is the one place that says
        ACTIVE (same invariant as self-service and admin_grant).
        """
        from app.domains.permission.lifecycle import (  # noqa: PLC0415
            load_approval_settings,
            notify_decision,
        )

        up = await self._get_user_permission(user_permission_id, lock=True)
        if up.status != PERM_PENDING_APPROVAL:
            raise AppException(
                code="PERMISSION_NOT_PENDING_APPROVAL",
                message=f"Request is {up.status!r}; only pending_approval requests can be decided.",
                status_code=409,
            )
        if not await self.can_decide(up, actor_user_id=actor_user_id):
            raise AppException(
                code="PERMISSION_APPROVER_REQUIRED",
                message=(
                    "Only the software owner, its IT contact or a global admin may "
                    "decide this permission request."
                ),
                status_code=403,
            )
        perm, software, requester = await self._load_decision_context(up)
        if not perm.is_active:
            raise AppException(
                code="PERMISSION_INACTIVE",
                message="This permission is retired and can no longer be granted",
                status_code=400,
            )
        if requester is None or is_blocked_from_new_grant(requester):
            raise AppException(
                code="TARGET_NOT_ACTIVE",
                message="The requester is no longer eligible for new permissions (leaver / purged).",
                status_code=403,
            )
        settings = await load_approval_settings(self.db)
        days = valid_days if valid_days is not None else settings.default_validity_days
        now = datetime.now(UTC)
        task = await self.provisioning.create_task(ProvisioningTaskCreate(
            user_id=up.user_id,
            software_id=perm.software_id,
            action=ACTION_GRANT,
            target_type="permission",
            target_id=perm.id,
            source=SOURCE_REQUEST,
            context_data={
                "permission_key": perm.permission_key,
                "reason": up.reason,
                "approved_by": str(actor_user_id),
                "valid_days": days,
            },
        ))
        up.status = PERM_PENDING
        up.provisioning_task_id = task.id
        up.approver_person_id = actor_user_id
        up.decided_at = now
        up.decision_note = (note or "").strip() or None
        up.expires_at = now + timedelta(days=int(days)) if days else None
        await self.db.flush()
        approver = await self.db.get(IdentityPerson, actor_user_id)
        await notify_decision(
            self.db, up=up, perm=perm, software=software, requester=requester,
            approver=approver, approved=True, note=up.decision_note, trace_id=trace_id,
        )
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="permission.request_approved",
            status="success",
            resource_type="user_permission",
            resource_id=str(up.id),
            detail={
                "permission_id": str(up.permission_id),
                "requester": str(up.user_id),
                "valid_days": days,
                "expires_at": up.expires_at.isoformat() if up.expires_at else None,
                "provisioning_task_id": str(task.id),
                "note": up.decision_note,
            },
        )
        return up

    async def reject_request(
        self,
        user_permission_id: UUID,
        *,
        actor_user_id: UUID,
        note: str | None,
        trace_id: str,
    ) -> UserPermission:
        from app.domains.permission.lifecycle import notify_decision  # noqa: PLC0415

        up = await self._get_user_permission(user_permission_id, lock=True)
        if up.status != PERM_PENDING_APPROVAL:
            raise AppException(
                code="PERMISSION_NOT_PENDING_APPROVAL",
                message=f"Request is {up.status!r}; only pending_approval requests can be decided.",
                status_code=409,
            )
        if not await self.can_decide(up, actor_user_id=actor_user_id):
            raise AppException(
                code="PERMISSION_APPROVER_REQUIRED",
                message=(
                    "Only the software owner, its IT contact or a global admin may "
                    "decide this permission request."
                ),
                status_code=403,
            )
        perm, software, requester = await self._load_decision_context(up)
        now = datetime.now(UTC)
        up.status = PERM_REJECTED
        up.approver_person_id = actor_user_id
        up.decided_at = now
        up.revoked_at = now
        up.decision_note = (note or "").strip() or None
        await self.db.flush()
        approver = await self.db.get(IdentityPerson, actor_user_id)
        if requester is not None:
            await notify_decision(
                self.db, up=up, perm=perm, software=software, requester=requester,
                approver=approver, approved=False, note=up.decision_note, trace_id=trace_id,
            )
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="permission.request_rejected",
            status="success",
            resource_type="user_permission",
            resource_id=str(up.id),
            detail={
                "permission_id": str(up.permission_id),
                "requester": str(up.user_id),
                "note": up.decision_note,
            },
        )
        return up

    async def list_pending_approvals(
        self,
        *,
        approver_id: UUID,
        include_all: bool,
        page: int = 1,
        page_size: int = 50,
    ) -> tuple[list[UserPermissionResponse], int]:
        """``pending_approval`` rows the caller may decide.

        ``include_all`` (global admin) lists every pending request; otherwise
        only those whose software names the caller as owner / IT contact.
        """
        conditions = [UserPermission.status == PERM_PENDING_APPROVAL]
        q = (
            select(UserPermission)
            .join(SoftwarePermission, SoftwarePermission.id == UserPermission.permission_id)
            .join(AssetSoftwareCatalog, AssetSoftwareCatalog.id == SoftwarePermission.software_id)
        )
        count_q = (
            select(func.count()).select_from(UserPermission)
            .join(SoftwarePermission, SoftwarePermission.id == UserPermission.permission_id)
            .join(AssetSoftwareCatalog, AssetSoftwareCatalog.id == SoftwarePermission.software_id)
        )
        if not include_all:
            from sqlalchemy import or_  # noqa: PLC0415
            conditions.append(or_(
                AssetSoftwareCatalog.owner_person_id == approver_id,
                AssetSoftwareCatalog.it_contact_person_id == approver_id,
            ))
        q = q.where(and_(*conditions))
        count_q = count_q.where(and_(*conditions))
        total = (await self.db.execute(count_q)).scalar() or 0
        rows = list((await self.db.execute(
            q.order_by(UserPermission.created_at.asc())
            .offset((page - 1) * page_size).limit(page_size)
        )).scalars().all())
        return await self._enrich_user_permissions(
            rows, viewer_id=approver_id, viewer_is_admin=include_all,
        ), total

    async def admin_grant(
        self,
        *,
        user_id: UUID,
        permission_id: UUID,
        reason: str | None,
        trace_id: str,
        actor_user_id: UUID,
        valid_days: int | None = None,
    ) -> UserPermission:
        """Admin force-grants without approval (e.g. as part of role change)."""
        perm = await self._get_permission(permission_id)
        # A retired / soft-deleted permission must not receive NEW grants —
        # mirror the ``request_permission`` guard (self-service already
        # blocks it via PERMISSION_INACTIVE). Without this, the admin path
        # could resurrect a permission IT deliberately retired.
        if not perm.is_active:
            raise AppException(
                code="PERMISSION_INACTIVE",
                message="This permission is retired and can no longer be granted",
                status_code=400,
            )

        # Leaver invariant: every other admin-grant path
        # (admin_grant_seat, admin_approve_seat, _activate_software_seat)
        # rejects a target whose employment_status has flipped to
        # ``left`` or ``leaver_pending``. The permission domain was
        # the only hole — re-opening the documented "no paid privilege
        # to a leaver" rule. Match the seat-path wording so admins
        # see consistent error messages.
        from app.identity.person_models import IdentityPerson  # noqa: PLC0415
        target = await self.db.get(IdentityPerson, user_id)
        if target is None:
            raise AppException(
                code="TARGET_NOT_FOUND",
                message=f"Target user {user_id} not found.",
                status_code=404,
            )
        if is_blocked_from_new_grant(target):
            raise AppException(
                code="TARGET_NOT_ACTIVE",
                message=(
                    f"Cannot grant permission to {target.work_email or target.id} — "
                    f"employment_status={target.employment_status!r}, "
                    f"hris_record_state={target.hris_record_state!r}. "
                    "Leavers and HRIS-purged records cannot receive new permissions."
                ),
                status_code=403,
            )

        # Look up by (user, permission) across ALL statuses. uq_user_perm
        # (UNIQUE user_id+permission_id) means at most one row — so a
        # terminal-state row (revoked / rejected / expired / cancelled) must
        # be REUSED, not INSERTed over: a bare INSERT trips the UNIQUE index
        # with an uncaught IntegrityError → 500, and an admin could never
        # re-grant a previously-revoked permission by any path. Mirror
        # ``request_permission``'s terminal-row reuse.
        existing = (
            await self.db.execute(
                select(UserPermission).where(
                    and_(
                        UserPermission.user_id == user_id,
                        UserPermission.permission_id == permission_id,
                    )
                ).with_for_update()
            )
        ).scalar_one_or_none()
        if existing and existing.status in (PERM_PENDING, PERM_ACTIVE):
            return existing
        # A request still waiting for the software owner: the admin grant IS
        # the decision — fall through and convert the row instead of
        # leaving a stale ``pending_approval`` next to a granted one.

        # Round-49: insert/reset at PERM_PENDING (not PERM_ACTIVE). The
        # cloud-side grant fires from the provisioning task at
        # ``_post_success_side_effects`` (provisioning/service.py),
        # which is the canonical PENDING → ACTIVE flip. Inserting at
        # ACTIVE here lied to the UI when the downstream driver call
        # failed — Helix said the user had the permission while the
        # cloud side never granted it. Mirrors the documented invariant
        # in ``request_permission`` (lines 232-239) and
        # ``revoke_user_permission`` (lines 457-470).
        task = await self.provisioning.create_task(ProvisioningTaskCreate(
            user_id=user_id,
            software_id=perm.software_id,
            action=ACTION_GRANT,
            target_type="permission",
            target_id=perm.id,
            source="role_change",
            context_data={
                "permission_key": perm.permission_key, "admin_granted": True,
                "valid_days": valid_days,
            },
        ))
        now = datetime.now(UTC)
        expires_at = now + timedelta(days=int(valid_days)) if valid_days else None
        if existing:
            was_pending_approval = existing.status == PERM_PENDING_APPROVAL
            # Terminal-state row → reset in place to a fresh pending grant.
            existing.status = PERM_PENDING
            existing.reason = reason if not was_pending_approval else (existing.reason or reason)
            existing.granted_at = None
            existing.revoked_at = None
            existing.external_ref = None
            existing.approval_request_id = None
            existing.provisioning_task_id = task.id
            existing.expires_at = expires_at
            existing.approver_person_id = actor_user_id
            existing.decided_at = now
            existing.decision_note = reason if was_pending_approval else None
            row = existing
        else:
            row = UserPermission(
                user_id=user_id,
                permission_id=permission_id,
                status=PERM_PENDING,
                reason=reason,
            )
            row.provisioning_task_id = task.id
            row.expires_at = expires_at
            row.approver_person_id = actor_user_id
            row.decided_at = now
            self.db.add(row)
        await self.db.flush()

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="permission.admin_granted",
            status="success",
            resource_type="user_permission",
            resource_id=str(row.id),
            detail={
                "target_user": str(user_id), "permission_id": str(permission_id),
                "valid_days": valid_days,
                "expires_at": expires_at.isoformat() if expires_at else None,
            },
        )
        return row

    async def _cancel_queued_grant(
        self, up: UserPermission, *, trace_id: str, actor_user_id: UUID | None,
    ) -> None:
        """#1775: cancel a still-queued auto-authorize GRANT task so a
        withdraw/revoke can't leave the vendor-side grant landing while
        Helix shows the permission cancelled/revoked — a permanent, silent
        cloud-side access leak (the GRANT task's success hook then no-ops
        because the row is CANCELLED/REVOKED, and role-change / offboarding
        diffs only enumerate ACTIVE/PENDING rows, so nothing ever reclaims
        it). Only non-terminal GRANT tasks are cancelled; a GRANT that
        already COMPLETED is left for the normal revoke path to unwind.
        """
        if not up.provisioning_task_id:
            return
        if actor_user_id is None:
            # System callers (the expiry sweep) only ever revoke ACTIVE rows,
            # whose GRANT task has already completed — nothing queued to kill.
            return
        task = await self.db.get(ProvisioningTask, up.provisioning_task_id)
        if task is None or task.action != ACTION_GRANT:
            return
        try:
            await self.provisioning.cancel_task(
                task.id,
                reason=(
                    "permission request withdrawn/revoked before the "
                    "auto-authorize grant executed (#1775)"
                ),
                trace_id=trace_id,
                actor_user_id=actor_user_id,
            )
        except AppException:
            # Task already terminal (raced to COMPLETED/CANCELLED). If it
            # COMPLETED, the vendor grant landed — the revoke path (for
            # revoke_user_permission) or reconcile unwinds it; for a bare
            # withdraw the standard drift/offboarding sweeps catch it.
            pass

    async def cancel_request(
        self,
        user_permission_id: UUID,
        *,
        actor_user_id: UUID,
        trace_id: str,
    ) -> UserPermission:
        """Requester withdraws their own still-PENDING permission request.

        Lifecycle-verb parity: sso / vpn / domain / cloud-scope / asset
        requests all let the requester withdraw a pending row; permission
        requests previously could not (you had to wait for an admin
        decision or escalate). Permission requests are NOT Feishu-routed
        (they're approved in-Helix via the approval inbox), so there is no
        external instance to cancel — this is a clean local state flip.

        Only the requester, only PENDING. If the request was attached to a
        local ``ApprovalRequest`` it is marked cancelled in lock-step so
        the admin inbox stops showing it.
        """
        # Lock the row: the async provisioning-completion path also
        # reads-then-flips PENDING (→ ACTIVE) unlocked, so a concurrent
        # withdraw + grant-completion could otherwise race (last writer
        # wins → status disagrees with the real cloud-side outcome).
        up = await self._get_user_permission(user_permission_id, lock=True)
        if up.user_id != actor_user_id:
            raise AppException(
                code="PERMISSION_NOT_OWNER",
                message="Only the original requester can withdraw this request.",
                status_code=403,
            )
        if up.status not in (PERM_PENDING, PERM_PENDING_APPROVAL):
            raise AppException(
                code="PERMISSION_NOT_PENDING",
                message=f"Cannot withdraw a request in {up.status!r} state.",
                status_code=409,
            )
        up.status = PERM_CANCELLED
        up.revoked_at = datetime.now(UTC)
        # #1775: kill any queued (not-yet-executed) auto-authorize GRANT
        # task so the vendor grant can't land after this withdraw.
        await self._cancel_queued_grant(
            up, trace_id=trace_id, actor_user_id=actor_user_id,
        )

        if up.approval_request_id is not None:
            from app.approval.models import ApprovalRequest
            appr = await self.db.get(ApprovalRequest, up.approval_request_id)
            if appr is not None and appr.status not in ("approved", "rejected", "cancelled"):
                appr.status = "cancelled"

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="permission.request_cancelled",
            status="success",
            resource_type="user_permission",
            resource_id=str(up.id),
            detail={"permission_id": str(up.permission_id)},
        )
        return up

    async def revoke_user_permission(
        self,
        user_permission_id: UUID,
        *,
        reason: str | None,
        trace_id: str,
        actor_user_id: UUID | None,
        terminal_status: str | None = None,
    ) -> UserPermission:
        """Revoke an active / in-flight grant through a REVOKE provisioning task.

        ``terminal_status`` (migration 606) is stamped into the task context;
        the success hook lands the row on it (``expired`` for the validity
        sweep) instead of the default ``revoked``. An admin revoking a row
        that is still ``pending_approval`` is a rejection — no vendor state
        exists yet, so no task is created.
        """
        up = await self._get_user_permission(user_permission_id, lock=True)
        if up.status == PERM_PENDING_APPROVAL:
            if actor_user_id is None:
                raise AppException(
                    code="PERMISSION_NOT_ACTIVE",
                    message="A pending approval can only be rejected by a person.",
                    status_code=400,
                )
            return await self.reject_request(
                user_permission_id, actor_user_id=actor_user_id,
                note=reason or "revoked by admin before approval", trace_id=trace_id,
            )
        if up.status not in {PERM_ACTIVE, PERM_PENDING}:
            raise AppException(
                code="PERMISSION_NOT_ACTIVE",
                message=f"Permission is {up.status}, cannot revoke",
                status_code=400,
            )
        perm = await self._get_permission(up.permission_id)

        # #1775: if this permission is still PENDING with a queued (not-yet-
        # executed) GRANT task, cancel that GRANT first — otherwise the
        # worker would GRANT on the vendor and then the REVOKE task races
        # behind it (or the grant lands after we've marked things revoked).
        await self._cancel_queued_grant(
            up, trace_id=trace_id, actor_user_id=actor_user_id,
        )

        # Create revoke provisioning task
        context: dict[str, object] = {"permission_key": perm.permission_key, "reason": reason}
        if terminal_status:
            context["terminal_status"] = terminal_status
        task = await self.provisioning.create_task(ProvisioningTaskCreate(
            user_id=up.user_id,
            software_id=perm.software_id,
            action=ACTION_REVOKE,
            target_type="permission",
            target_id=perm.id,
            source=SOURCE_REQUEST,
            context_data=context,
        ))
        # Pre-launch hardening 2026-05-06: do NOT flip ``status =
        # PERM_REVOKED`` synchronously here. The driver call hasn't run
        # yet — it fires from the provisioning task. If the driver
        # call fails, ``_post_success_side_effects`` never runs (it's
        # only invoked on task success), and the previous code left
        # Helix saying REVOKED while the cloud side still has the
        # grant — i.e. a leaver retains real cloud-side access while
        # Helix's UI claims they don't. Keeping status as PERM_ACTIVE
        # until ``_post_success_side_effects`` runs (line 2293-2299
        # of provisioning/service.py) means: success → flip to
        # REVOKED with audit, failure → status stays ACTIVE and the
        # task surfaces as FAILED for admin retry. The provisioning
        # task is the single source of truth for "did the cloud side
        # actually move."
        up.provisioning_task_id = task.id

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user" if actor_user_id is not None else "system",
            actor_user_id=actor_user_id,
            action="permission.revocation_requested",
            status="success",
            resource_type="user_permission",
            resource_id=str(up.id),
            detail={
                "reason": reason,
                "terminal_status": terminal_status,
                "provisioning_task_id": str(task.id),
                "note": (
                    "status will flip to revoked when the driver call "
                    "succeeds; failures surface in the provisioning task"
                ),
            },
        )
        return up

    async def reconcile_user_permission(self, user_permission_id: UUID) -> UserPermission:
        """Sync user_permission.status with its underlying provisioning task
        status. Called when admin wants fresh status, or by a future
        scheduler."""
        up = await self._get_user_permission(user_permission_id)
        if not up.provisioning_task_id:
            return up
        task = (
            await self.db.execute(
                select(ProvisioningTask).where(ProvisioningTask.id == up.provisioning_task_id)
            )
        ).scalar_one_or_none()
        if not task:
            return up
        if task.status == TASK_COMPLETED:
            up.external_ref = task.external_ref
            if task.action == ACTION_GRANT and up.status == PERM_PENDING:
                up.status = PERM_ACTIVE
                up.granted_at = datetime.now(UTC)
        return up

    # ── Queries ──

    async def list_user_permissions(
        self,
        *,
        user_id: UUID | None = None,
        permission_id: UUID | None = None,
        status: str | None = None,
        page: int = 1,
        page_size: int = 20,
        viewer_id: UUID | None = None,
        viewer_is_admin: bool = False,
    ) -> tuple[list[UserPermissionResponse], int]:
        conditions = []
        if user_id:
            conditions.append(UserPermission.user_id == user_id)
        if permission_id:
            conditions.append(UserPermission.permission_id == permission_id)
        if status:
            conditions.append(UserPermission.status == status)

        q = select(UserPermission)
        count_q = select(func.count()).select_from(UserPermission)
        if conditions:
            q = q.where(and_(*conditions))
            count_q = count_q.where(and_(*conditions))

        total = (await self.db.execute(count_q)).scalar() or 0
        rows = list(
            (await self.db.execute(
                q.order_by(UserPermission.created_at.desc())
                .offset((page - 1) * page_size).limit(page_size)
            )).scalars().all()
        )
        return await self._enrich_user_permissions(
            rows, viewer_id=viewer_id, viewer_is_admin=viewer_is_admin,
        ), total

    # ── Internal ──

    async def _get_permission(self, permission_id: UUID) -> SoftwarePermission:
        row = (
            await self.db.execute(
                select(SoftwarePermission).where(SoftwarePermission.id == permission_id)
            )
        ).scalar_one_or_none()
        if not row:
            raise AppException(
                code="PERMISSION_NOT_FOUND",
                message="Permission not found",
                status_code=404,
            )
        return row

    async def _get_user_permission(self, up_id: UUID, *, lock: bool = False) -> UserPermission:
        stmt = select(UserPermission).where(UserPermission.id == up_id)
        if lock:
            stmt = stmt.with_for_update()
        row = (await self.db.execute(stmt)).scalar_one_or_none()
        if not row:
            raise AppException(
                code="USER_PERMISSION_NOT_FOUND",
                message="User permission record not found",
                status_code=404,
            )
        return row

    async def _enrich_permissions(
        self, rows: list[SoftwarePermission],
    ) -> list[PermissionResponse]:
        sw_ids = {r.software_id for r in rows}
        sw_rows = (
            await self.db.execute(
                select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.id.in_(sw_ids))
            )
        ).scalars().all() if sw_ids else []
        sw_map = {s.id: s for s in sw_rows}
        return [
            PermissionResponse(
                id=r.id,
                software_id=r.software_id,
                software_name=(sw_map.get(r.software_id).name if sw_map.get(r.software_id) else None),
                permission_key=r.permission_key,
                display_name=r.display_name,
                description=r.description,
                requires_approval=r.requires_approval,
                is_active=r.is_active,
                sort_order=r.sort_order,
                created_at=r.created_at,
                updated_at=r.updated_at,
            )
            for r in rows
        ]

    async def _enrich_user_permissions(
        self,
        rows: list[UserPermission],
        *,
        viewer_id: UUID | None = None,
        viewer_is_admin: bool = False,
    ) -> list[UserPermissionResponse]:
        if not rows:
            return []
        from app.domains.permission.lifecycle import software_approver_ids  # noqa: PLC0415

        perm_ids = {r.permission_id for r in rows}
        perms = (
            await self.db.execute(
                select(SoftwarePermission).where(SoftwarePermission.id.in_(perm_ids))
            )
        ).scalars().all()
        perm_map = {p.id: p for p in perms}

        sw_ids = {p.software_id for p in perms}
        swrows = (
            await self.db.execute(
                select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.id.in_(sw_ids))
            )
        ).scalars().all() if sw_ids else []
        sw_map = {s.id: s for s in swrows}

        # Holders + approvers + software owners / IT contacts in one read.
        user_ids = {r.user_id for r in rows}
        user_ids |= {r.approver_person_id for r in rows if r.approver_person_id}
        for s in swrows:
            user_ids |= set(software_approver_ids(s))
        users = (
            await self.db.execute(
                select(IdentityPerson).where(IdentityPerson.id.in_(user_ids))
            )
        ).scalars().all()
        user_map = {u.id: u for u in users}
        now = datetime.now(UTC)

        out: list[UserPermissionResponse] = []
        for r in rows:
            perm = perm_map.get(r.permission_id)
            sw = sw_map.get(perm.software_id) if perm else None
            user = user_map.get(r.user_id)
            approver = user_map.get(r.approver_person_id) if r.approver_person_id else None
            approver_ids = software_approver_ids(sw)
            pending_labels = [
                (user_map[a].display_name or user_map[a].work_email or str(a))
                for a in approver_ids if a in user_map
            ]
            can_decide = bool(
                r.status == PERM_PENDING_APPROVAL
                and viewer_id is not None
                and (viewer_is_admin or viewer_id in approver_ids)
            )
            waiting_days = None
            if r.status == PERM_PENDING_APPROVAL and r.created_at is not None:
                created = r.created_at if r.created_at.tzinfo else r.created_at.replace(tzinfo=UTC)
                waiting_days = max((now - created).days, 0)
            out.append(UserPermissionResponse(
                requires_approval=bool(perm.requires_approval) if perm else False,
                expires_at=r.expires_at,
                approver_person_id=r.approver_person_id,
                approver_name=(approver.display_name or approver.work_email) if approver else None,
                decided_at=r.decided_at,
                decision_note=r.decision_note,
                pending_approver_labels=pending_labels if r.status == PERM_PENDING_APPROVAL else [],
                can_decide=can_decide,
                approval_waiting_days=waiting_days,
                id=r.id,
                user_id=r.user_id,
                user_name=user.display_name if user else None,
                user_email=user.work_email if user else None,
                user_employee_no=user.employee_no if user else None,
                user_employment_status=user.employment_status if user else None,
                permission_id=r.permission_id,
                permission_name=perm.display_name if perm else "-",
                permission_key=perm.permission_key if perm else "-",
                software_id=perm.software_id if perm else None,
                software_name=sw.name if sw else None,
                status=r.status,
                reason=r.reason,
                approval_request_id=r.approval_request_id,
                provisioning_task_id=r.provisioning_task_id,
                external_ref=r.external_ref,
                granted_at=r.granted_at,
                revoked_at=r.revoked_at,
                created_at=r.created_at,
                updated_at=r.updated_at,
            ))
        return out