"""Integration driver base + kind-specific mixins.

Drivers are pluggable: each provider is one subclass, registered via
``@register_driver`` at import time. Kind and provider are both
free-form strings so adding a new cloud (Azure, GCP), a new chat
system (Slack, Teams), or a new HRIS (Workday, BambooHR) is just a
new driver file — no schema migration.

The framework deliberately keeps the driver's public surface small:

  * class-level **metadata** — kind, provider, display_name,
    config_schema — drives the admin form and the kind directory.
  * **test_connection** — a cheap health probe every driver implements.
  * **kind-specific mixins** — narrow the operations interface for a
    given *kind*. HRIS drivers implement ``list_active_users``, cloud
    drivers implement ``list_servers``, chat_approval drivers implement
    ``submit_approval``, data_access drivers implement
    ``get_dataset_iam`` / ``set_dataset_iam``, etc. A caller looking up
    "the HRIS integration" knows exactly what methods it can call
    without downcasting.

Concrete drivers never touch the database directly; they only know
about their plaintext config dict and the external system.
"""
from __future__ import annotations

import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import TYPE_CHECKING, Any, ClassVar

if TYPE_CHECKING:
    from app.integrations.cloud_role_governance import RoleGovernance

from app.integrations.schemas import (
    IntegrationDriverMetadata,
    IntegrationFieldInfo,
    IntegrationTestResult,
)

logger = logging.getLogger(__name__)


# ─── Framework base ────────────────────────────────────────────────


class IntegrationDriver(ABC):
    """Abstract base every driver inherits from.

    Subclasses MUST set the class attributes listed below before
    ``@register_driver`` is applied; subclasses SHOULD implement
    ``test_connection`` (defaults to a "not implemented" failure).

    Instance lifetime is per-request: the service constructs a driver,
    uses it, throws it away. No long-lived state beyond what the
    concrete driver caches internally (e.g. OAuth access tokens).
    """

    # ─── Required metadata ─
    kind: str = ""
    provider: str = ""
    display_name: str = ""
    description: str = ""

    # ─── Optional metadata ─
    supports_multi_instance: bool = True
    supports_test_connection: bool = True
    capabilities: ClassVar[list[str]] = []  # for kinds like cloud

    # Archival tenant — the driver exists only so a decommissioned
    # provider's migrated historical rows render under a real tenant
    # name in the UI. There is no live sync, no credentials, no
    # capabilities. Active-management surfaces (the half-yearly server
    # owner confirmation, sync schedulers, …) MUST skip these so they
    # don't nag owners about machines the company no longer runs.
    archival: ClassVar[bool] = False

    # Tenant-level config fields (credentials + global settings) rendered
    # on the integration edit form.
    config_schema: list[IntegrationFieldInfo] = []
    # Per-binding config fields — only meaningful for
    # ``software_provisioning`` drivers where each software has its own
    # binding row with provider-specific params (group_email,
    # group_chat_id, operation...). Other kinds leave this empty.
    binding_schema: list[IntegrationFieldInfo] = []

    def __init__(
        self, *,
        integration_id: str,
        config: dict[str, Any],
        enabled_capabilities: list[str] | None = None,
    ) -> None:
        self.integration_id = integration_id
        self.config = config
        # Per-instance enabled capability set. ``test_connection`` and
        # other diagnostic paths consult this to probe exactly what the
        # admin opted into — not the driver-class defaults. The latter
        # only matters as a fallback when an admin has not customised
        # the column (legacy rows, factory defaults). Kept off
        # ``self.config`` so the provider-config payload that gets fed
        # to SDKs stays surface-of-credentials only.
        self.enabled_capabilities: list[str] = (
            list(enabled_capabilities)
            if enabled_capabilities is not None
            else list(self.__class__.capabilities)
        )

    @classmethod
    def metadata(cls) -> IntegrationDriverMetadata:
        return IntegrationDriverMetadata(
            kind=cls.kind,
            provider=cls.provider,
            display_name=cls.display_name or cls.provider,
            description=cls.description,
            fields=list(cls.config_schema),
            binding_fields=list(cls.binding_schema),
            supports_multi_instance=cls.supports_multi_instance,
            supports_test_connection=cls.supports_test_connection,
            capabilities=list(cls.capabilities),
        )

    async def test_connection(self) -> IntegrationTestResult:
        """Health probe. Override in subclass.

        Return ``ok=True`` when credentials are valid and at least one
        minimal read-only call succeeded against the external system.
        """
        return IntegrationTestResult(
            ok=False,
            message=f"Driver {self.kind}/{self.provider} has no test_connection impl",
        )


# ─── HRIS ──────────────────────────────────────────────────────────


class HrisDriver(IntegrationDriver):
    """Drivers that source employee master data."""

    kind = "hris"

    @abstractmethod
    async def list_active_users(self) -> list[dict[str, Any]]:
        """Every active employee in the source system.

        Used to seed identity_persons in a full-sync — no date window.
        Return shape mirrors the SF User object:
        ``{userId, displayName, email, empId, status, hireDate,
           lastModifiedDateTime, ...}``.
        """

    @abstractmethod
    async def list_hires_in_range(
        self, start_date: date, end_date: date,
    ) -> list[dict[str, Any]]:
        """Hire events whose startDate falls in the window."""

    @abstractmethod
    async def list_terminations_in_range(
        self, start_date: date, end_date: date,
    ) -> list[dict[str, Any]]:
        """Termination events whose endDate falls in the window."""

    async def get_users(self, user_ids: list[str]) -> list[dict[str, Any]]:
        """Bulk fetch user records by the source system's user_id."""
        return []

    async def get_jobs_for_users(self, user_ids: list[str]) -> list[dict[str, Any]]:
        """Optional: richer job/org data per user. Default to empty."""
        return []


# ─── Chat + approval (Feishu / Lark, future: Slack, Teams) ─────────


@dataclass(slots=True)
class ApprovalSubmission:
    """Input to ``submit_approval`` — platform-agnostic payload."""

    flow_key: str
    # Canonical field map built from the caller's domain (e.g. asset
    # request reason, user, software name). The driver is responsible
    # for mapping these to the provider's widget-id representation
    # using the IntegrationApprovalFlow.payload_schema.
    fields: dict[str, Any]
    # Feishu-style "on behalf of" user id or email, if applicable.
    open_user_id: str | None = None
    # Stable provider-native idempotency identity, durably persisted by the
    # caller before invoking the driver.  A transport retry MUST reuse it.
    # Optional at construction only for source compatibility with external
    # drivers; submitters fail closed when it is absent.
    provider_request_uuid: str | None = None


@dataclass(slots=True)
class ApprovalSubmissionResult:
    ok: bool
    instance_code: str | None = None
    raw_response: dict[str, Any] | None = None
    error_message: str | None = None


@dataclass(slots=True)
class WebhookVerificationResult:
    ok: bool
    # Decrypted/validated event payload ready for the domain to consume.
    event: dict[str, Any] | None = None
    challenge: str | None = None  # for initial URL verification
    error_message: str | None = None


class ChatApprovalDriver(IntegrationDriver):
    """Chat + approval systems (Feishu today, Slack/Teams later)."""

    kind = "chat_approval"

    @abstractmethod
    async def submit_approval(
        self, submission: ApprovalSubmission,
    ) -> ApprovalSubmissionResult:
        """Create an approval instance in the external system."""

    @abstractmethod
    async def verify_webhook(
        self,
        *,
        body: bytes,
        headers: dict[str, str],
    ) -> WebhookVerificationResult:
        """Validate signature + timestamp, return parsed event."""

    async def cancel_approval(self, instance_code: str) -> bool:
        """Optional: cancel an in-flight approval."""
        return False

    async def resolve_user_id_by_email(self, email: str) -> str | None:
        """Optional: map a work email to the provider's user id."""
        return None


# ─── Data access providers (dataset IAM adapters) ──────────────────


@dataclass(slots=True, frozen=True)
class DataAccessDatasetIamBinding:
    """One role binding in a BigQuery dataset IAM policy."""

    role: str
    members: tuple[str, ...] = ()


@dataclass(slots=True, frozen=True)
class DataAccessDatasetIamPolicy:
    """Provider IAM snapshot for one dataset locator."""

    project_id: str
    dataset_id: str
    etag: str | None
    bindings: tuple[DataAccessDatasetIamBinding, ...] = ()
    raw_policy: dict[str, Any] | None = None
    table_id: str | None = None


@dataclass(slots=True, frozen=True)
class DataAccessDatasetTableSummary:
    """One table or view inside an exact dataset locator."""

    table_id: str
    table_type: str


@dataclass(slots=True, frozen=True)
class DataAccessDatasetTablePage:
    """One bounded provider page of coarse table metadata.

    ``next_page_token`` is provider-opaque and must only be passed back to the
    same exact dataset list operation.  A page intentionally carries no schema
    fields: callers fetch those only after a user opens one table.
    """

    items: tuple[DataAccessDatasetTableSummary, ...]
    next_page_token: str | None = None
    total_rows: int | None = None


@dataclass(slots=True, frozen=True)
class DataAccessDatasetFieldSchema:
    """Flattened BigQuery field metadata, including nested RECORD paths."""

    name: str
    field_path: str
    field_type: str
    mode: str
    description: str | None
    depth: int
    ordinal_position: int


@dataclass(slots=True, frozen=True)
class DataAccessDatasetTableSchema:
    """Current metadata for one table within an exact dataset locator."""

    project_id: str
    dataset_id: str
    table_id: str
    table_type: str
    description: str | None
    fields: tuple[DataAccessDatasetFieldSchema, ...] = ()


@dataclass(slots=True, frozen=True)
class DataAccessDatasetIamPolicyPatch:
    """Narrow IAM patch for one dataset locator.

    The reconciler builds this from compiled effects. V1 callers may only
    encode Google Group members and the canonical read role; broader roles,
    user principals, project IAM, and table access remain out of scope.
    """

    add_bindings: tuple[DataAccessDatasetIamBinding, ...] = ()
    remove_bindings: tuple[DataAccessDatasetIamBinding, ...] = ()


@dataclass(slots=True, frozen=True)
class DataAccessDatasetIamUpdateResult:
    """Result of applying a dataset IAM patch."""

    ok: bool
    changed: bool
    etag: str | None = None
    conflict: bool = False
    already_applied: bool = False
    policy: DataAccessDatasetIamPolicy | None = None
    error_message: str | None = None
    raw_response: dict[str, Any] | None = None


class DataAccessDriver(IntegrationDriver):
    """Dedicated integrations for data-access management.

    Dataset IAM is deliberately locator-scoped: callers must name the exact
    ``(project_id, dataset_id)`` they are reconciling. There is no broad
    dataset inventory/manage operation on this interface.
    """

    kind = "data_access"

    async def list_dataset_tables(
        self,
        project_id: str,
        dataset_id: str,
    ) -> tuple[DataAccessDatasetTableSummary, ...]:
        """List tables within one exact dataset locator, never across projects."""
        raise NotImplementedError(
            f"Data-access driver {self.provider!r} does not implement "
            "list_dataset_tables",
        )

    async def list_dataset_tables_page(
        self,
        project_id: str,
        dataset_id: str,
        *,
        page_size: int,
        page_token: str | None = None,
    ) -> DataAccessDatasetTablePage:
        """List one bounded provider page without enumerating the dataset."""
        raise NotImplementedError(
            f"Data-access driver {self.provider!r} does not implement "
            "list_dataset_tables_page",
        )

    async def get_dataset_table_schema(
        self,
        project_id: str,
        dataset_id: str,
        table_id: str,
    ) -> DataAccessDatasetTableSchema:
        """Read schema metadata for one table within one exact dataset locator."""
        raise NotImplementedError(
            f"Data-access driver {self.provider!r} does not implement "
            "get_dataset_table_schema",
        )

    async def get_dataset_iam(
        self,
        project_id: str,
        dataset_id: str,
    ) -> DataAccessDatasetIamPolicy:
        """Read IAM for one BigQuery dataset locator."""
        raise NotImplementedError(
            f"Data-access driver {self.provider!r} does not implement "
            "get_dataset_iam",
        )

    async def set_dataset_iam(
        self,
        project_id: str,
        dataset_id: str,
        expected_etag: str | None,
        policy_patch: DataAccessDatasetIamPolicyPatch,
    ) -> DataAccessDatasetIamUpdateResult:
        """Apply a patch to one BigQuery dataset IAM policy.

        Implementations must honor ``expected_etag`` for optimistic
        concurrency and must not translate this into project-level IAM.
        """
        raise NotImplementedError(
            f"Data-access driver {self.provider!r} does not implement "
            "set_dataset_iam",
        )

    async def get_table_iam(
        self,
        project_id: str,
        dataset_id: str,
        table_id: str,
    ) -> DataAccessDatasetIamPolicy:
        """Read IAM for one exact BigQuery table or view locator."""
        raise NotImplementedError(
            f"Data-access driver {self.provider!r} does not implement get_table_iam",
        )

    async def set_table_iam(
        self,
        project_id: str,
        dataset_id: str,
        table_id: str,
        expected_etag: str | None,
        policy_patch: DataAccessDatasetIamPolicyPatch,
    ) -> DataAccessDatasetIamUpdateResult:
        """Apply a bounded IAM patch to one exact BigQuery table or view."""
        raise NotImplementedError(
            f"Data-access driver {self.provider!r} does not implement set_table_iam",
        )


# ─── Cloud providers (CMDB data sources) ────────────────────────────


class CloudDriver(IntegrationDriver):
    """Public-cloud providers. CMDB pulls instance / domain / DNS data.

    Also the home of per-user account revocation (``revoke_account``) —
    a cloud driver that supports the ``user_revoke`` capability can
    delete the IAM user / RAM user / service account that a leaver
    owned. Drivers that don't implement IAM ops raise NotImplementedError.

    **Account lifecycle extensions** (PR 2026-04-23): cloud drivers
    participate in the tenant → scope → grant model. Each driver declares
    its mode via ``account_mode`` and implements a subset of the
    ``list_scopes`` / ``grant_access`` / ``revoke_access`` /
    ``list_billing_daily`` surface. Drivers that don't (yet) can raise
    ``NotImplementedError`` and the lifecycle service falls back to a
    manual path (admin marks grant as completed after doing the change
    in the cloud console).
    """

    kind = "cloud"

    # Declared per-provider: which capabilities this provider supports.
    # Admin can enable/disable a subset per-instance via ``capabilities``.
    supported_capabilities: frozenset[str] = frozenset()

    # Whether this driver's ``list_cost_facts`` requires a CNY→USD
    # rate. CNY-native providers (Aliyun, Tencent) bill in CNY and
    # need it to compute ``amount_usd``; USD-native providers (AWS,
    # Azure, GCP) leave this False so cost ingestion doesn't fail
    # spuriously when the platform FX setting is unavailable. The
    # cost-ingestion service inspects this flag and only resolves a
    # rate when needed (passes None otherwise).
    requires_cny_fx: bool = False

    # ``sso`` = federated identity per-sub-account (AWS Org + Identity
    # Center, Azure subscriptions + AAD, GCP projects, Alicloud RD).
    # Grants are role assignments on already-existing principals.
    # ``iam`` = single-tenant IAM user mode (AWS SG, Tencent). A grant
    # means create/locate an IAM user for the person and attach a policy.
    account_mode: str = "sso"

    # Role keys that the driver considers "owner-equivalent" — having
    # any of these on a scope means the principal effectively owns it.
    # The ``cmdb.grants_sync`` handler uses this for auto-populating
    # ``CmdbCloudScope.owner_person_id`` when the scope has no owner
    # and exactly one person holds an owner role. Drivers MUST include
    # the provider's highest-privilege role so scope auto-adoption
    # works (Owner, AdministratorAccess, roles/owner, etc.).
    owner_role_keys: frozenset[str] = frozenset()

    # ─── Role governance (PR cloud-role-governance) ─────────────
    #
    # Provider-specific knowledge for the central tier classifier
    # (``app/integrations/cloud_role_governance.py``). Drivers declare
    # the role keys they KNOW to be restricted (owner-equivalent /
    # full-IAM-admin / billing-admin) or elevated (broad write / IAM-
    # write that isn't named "admin"). ``owner_role_keys`` is folded
    # into restricted automatically. Roles not listed fall back to the
    # classifier's keyword sieve. Ungrantable (wrong-scope / wrong-shape)
    # is per-scope-type, so it's a method (``ungrantable_role_keys``)
    # rather than a flat attribute.
    restricted_role_keys: frozenset[str] = frozenset()
    elevated_role_keys: frozenset[str] = frozenset()

    @classmethod
    def ungrantable_role_keys(cls, scope_type: str) -> frozenset[str]:  # noqa: ARG003
        """Role keys that cannot be bound at ``scope_type`` (provider
        rejects them — e.g. GCP billing/org roles on a project). Default
        empty; drivers override per scope type.

        A ``classmethod`` (uses only class-declared sets, no instance
        config) so the request/approve service layer can classify a role
        straight from the driver class — ``get_driver_class(...)`` — with
        no secrets and independent of whether a role sync has run."""
        return frozenset()

    @classmethod
    def classify_role(
        cls, role_key: str | None, display_name: str | None,
        scope_type: str,
    ) -> RoleGovernance:
        """Governance verdict for one role on ``scope_type`` — composes
        this driver's declared sets with the central keyword backstop.
        Used by role sync (to persist the tier) and by the request /
        approve service layer (to re-check live). ``classmethod`` — see
        :meth:`ungrantable_role_keys`."""
        from app.integrations.cloud_role_governance import classify_role
        return classify_role(
            role_key, display_name,
            owner_role_keys=cls.owner_role_keys,
            ungrantable_keys=cls.ungrantable_role_keys(scope_type),
            restricted_keys=cls.restricted_role_keys,
            elevated_keys=cls.elevated_role_keys,
        )

    # Whether grants are scoped per-scope (True: Azure subscription /
    # GCP project / AWS Identity Center per-account) or tenant-wide
    # (False: Aliyun RAM, Tencent CAM — one user can have policies
    # that apply across every account).
    #
    # The grants-sync service calls ``list_scope_grants`` **once per
    # scope** when True, and **once per integration** (on the first
    # scope) when False. Drivers with tenant-wide semantics must still
    # accept a ``scope`` argument to tag the discovered grants, but
    # they should return the full tenant-wide enumeration regardless.
    grants_scope_wide: bool = True

    @abstractmethod
    async def list_servers(self) -> list[dict[str, Any]]: ...

    async def list_domains(self) -> list[dict[str, Any]]:
        return []

    async def list_dns_records(self) -> list[dict[str, Any]]:
        return []

    async def list_load_balancers(self) -> list[dict[str, Any]]:
        return []

    async def list_cdn_domains(self) -> list[dict[str, Any]]:
        return []

    async def list_object_storage(self) -> list[dict[str, Any]]:
        return []

    async def list_applications(self) -> list[dict[str, Any]]:
        return []

    # ─── Scope + grant lifecycle (PR 2026-04-23) ────────────────
    #
    # Drivers implementing these methods must honor idempotency: calling
    # ``grant_access`` for a (scope, principal, role) that's already
    # granted must succeed with ``already_exists=True`` (don't double-
    # assign). ``revoke_access`` on a missing assignment must succeed
    # with ``already_absent=True``. The lifecycle service relies on
    # these semantics for its retry paths.

    async def list_scopes(self) -> list[CloudScopeInfo]:
        """Enumerate every sub-account / subscription / project / RAM
        member / standalone container under this tenant.

        Default raises NotImplementedError — drivers that don't ship
        scope sync yet route through an IT-admin-created scope row."""
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement list_scopes"
        )

    def scope_snapshot_partitions(self) -> dict[str, CloudScopePartitionStatus]:
        """Completeness metadata for the most recent :meth:`list_scopes`.

        Most providers expose one all-or-nothing scope enumeration: if
        ``list_scopes`` returns, its list is authoritative, so the default is
        an empty mapping.  Drivers that merge independent provider APIs (GCP
        projects + billing accounts) override this and name every partition.
        The lifecycle service uses an incomplete partition as a sweep fence;
        a successful sibling partition can never make it authoritative.
        """
        return {}

    async def create_scope(
        self,
        *,
        name: str,
        description: str | None,
        owner_email: str,
        budget_usd_monthly: float,
        metadata: dict[str, Any] | None = None,
        idempotency_key: str | None = None,
    ) -> CloudScopeInfo:
        """Create a new sub-account / subscription / project on the
        cloud side.

        Powers the self-service "申请新订阅" flow: an employee submits
        a request, Feishu approves, and the worker calls this method to
        actually allocate the cloud-side container. Returns a
        :class:`CloudScopeInfo` whose ``external_id`` will be inserted
        into ``cmdb_cloud_scopes`` with the requester as owner +
        platform-default budget / alert rules pre-applied.

        ``idempotency_key`` (typically the scope-request id) is
        accepted for signature compatibility but support is
        driver-specific. AWS Organizations exposes no first-class
        idempotency token on CreateAccount (the historically-assumed
        ``ClientRequestToken`` parameter does not exist on that API);
        the AWS driver instead achieves redrive safety by sweeping
        ``list_create_account_status`` for an IN_PROGRESS / SUCCEEDED
        entry matching AccountName before each call. Drivers that
        DO support a native token (e.g. some GCP / Azure paths) MAY
        propagate the kwarg; drivers that don't MUST still accept it
        and may ignore. The caller MUST pass a stable key — same
        request id across retries — so future drivers that add native
        dedupe support get a usable handle.

        Drivers that can't safely automate this (most of them — Azure
        needs a billing-scope agreement, GCP needs an org-id quota
        check, Aliyun resource-directory needs RAM enrollment) MUST
        raise ``NotImplementedError`` here so the request lands in
        ``failed`` state with a clear "manual fulfillment required"
        message rather than partially-creating a half-billed account.
        ``cmdb_cloud_scope_requests.failure_reason`` carries the message
        verbatim.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} cannot auto-create scopes — "
            "an IT admin must create the sub-account in the cloud "
            "console and bind it to Helix manually."
        )

    async def list_scope_roles(
        self, scope: CloudScopeRef,
    ) -> list[CloudRoleInfo]:
        """Role / permission-set catalog for one scope.

        Returns both predefined and custom roles. The lifecycle service
        caches the result in ``cmdb_cloud_roles``; this method is called
        daily by ``cmdb.roles_sync`` and on-demand from the "refresh"
        button."""
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement list_scope_roles"
        )

    async def grant_access(self, ctx: CloudAccessContext) -> CloudGrantResult:
        """Grant role on scope to principal (email-keyed).

        Drivers are responsible for resolving ``principal_email`` to a
        tenant-side principal (AAD user object id, Identity Store user
        id, GCP Google identity, RAM user login name). SSO drivers
        expect the principal to already exist; they must raise
        ``CloudPrincipalNotFoundError`` (subclass of ValueError) if not.
        IAM-mode drivers create the IAM user on the fly if needed."""
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement grant_access"
        )

    async def revoke_access(self, ctx: CloudAccessContext) -> CloudGrantResult:
        """Revoke an existing grant. Must be idempotent (404 = already
        absent = success)."""
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement revoke_access"
        )

    async def validate_request_role(
        self, scope: CloudScopeRef, role_key: str,
    ) -> str | None:
        """Pre-flight check at request/add-member time: is ``role_key`` a
        shape the eventual :meth:`grant_access` could actually bind on
        ``scope``? Returns ``None`` when grantable, or a human-readable
        error string when not.

        Why this exists: the governance gate (:meth:`classify_role`) is a
        static heuristic and cannot tell that a RAM-shape role_key
        (``Type:Name``) is un-bindable on a CloudSSO-managed AliCloud
        tenant — so a stale catalog row used to pass submit, route through
        the owner's approval, and only fail at grant time with a confusing
        toast (the approver, not the requester, sees it). This hook lets a
        driver reject the bad shape up front so the requester gets the
        error immediately and the request never enters the queue.

        Contract for overrides: be CHEAP and FAIL-OPEN. A transient API
        error (e.g. ListDirectories 5xx) must return ``None`` — the
        grant-time path remains the authoritative gate and will retry /
        reject there. Never raise. Default permits everything (no driver
        knows of a shape it must reject pre-flight)."""
        return None

    async def list_scope_grants(
        self, scope: CloudScopeRef,
    ) -> list[DiscoveredGrant]:
        """Every role/policy assignment currently active on this scope.

        Powers the reverse-direction sync (cloud → Helix) run daily by
        ``cmdb.grants_sync``. Each returned row represents one principal
        with one role; a single person with Owner + Contributor yields
        two rows. The service layer filters to ``principal_type='user'``
        before persisting a ``CmdbCloudAccessGrant``; groups and
        service principals flow through for drift reporting but never
        become grant rows.

        Drivers MUST:
          * filter out deleted / terminated principals at the provider
            level when possible (saves Graph/Identity Store lookups);
          * populate ``external_ref`` with the same shape used by
            :meth:`grant_access` so a discovered grant can later be
            revoked through the normal path;
          * tolerate transient per-principal resolution failures (e.g.
            Graph 404 on a just-deleted AAD user) by skipping the row
            and logging, not raising — the sweep should continue.

        Default raises ``NotImplementedError`` so drivers that don't
        yet ship grants sync are visibly opted out in the scheduled
        jobs history. The handler marks the per-scope attempt as
        ``skipped`` when this fires, not as an error."""
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement list_scope_grants"
        )

    async def list_scope_orphan_assignments(
        self, scope: CloudScopeRef,
    ) -> list[OrphanAssignment]:
        """Enumerate role/policy assignments on this scope whose
        principal is missing from the tenant directory.

        SSO clouds: AWS Identity Center keeps the assignment row
        forever even after the SCIM-synced user is deleted from the
        Identity Store; the console displays "User not found (UUID)".
        :meth:`list_scope_grants` silently drops these because email
        resolution returns None — they're invisible to the normal sync
        path. This method surfaces them so an admin can clean up.

        IAM clouds: typically not applicable (the IAM user ITSELF is
        the principal — when the user is deleted, the assignment is
        gone too). Drivers may either return ``[]`` or raise
        ``NotImplementedError``; the service treats both as "this
        provider does not have orphan assignments".

        Drivers that implement this MUST also implement
        :meth:`delete_orphan_assignment` so the admin's "确认清理"
        action can actually undo each row. The two methods together
        form the orphan-cleanup contract.

        Default raises ``NotImplementedError`` so providers that
        haven't shipped orphan support are visibly opted out and the
        admin endpoint surfaces a clear "this provider doesn't
        support orphan scan" instead of a silent empty list.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement "
            "list_scope_orphan_assignments",
        )

    async def delete_orphan_assignment(
        self, scope: CloudScopeRef, orphan: OrphanAssignment,
    ) -> CloudGrantResult:
        """Delete one orphan assignment from the cloud side.

        Idempotent: a second call after the assignment is gone must
        return ``CloudGrantResult(ok=True, already_absent=True)``.

        Default raises ``NotImplementedError``. Drivers implementing
        :meth:`list_scope_orphan_assignments` MUST implement this.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement "
            "delete_orphan_assignment",
        )

    async def list_principal_assignments(
        self, email: str,
    ) -> list[PrincipalAssignment]:
        """Enumerate every account/permission-set assignment a single
        live principal holds, keyed by their email.

        Used by the offboarding leaver-sweep: when a person leaves, we
        want to delete ALL their cloud-side assignments — including ones
        Helix never tracked as a grant row (created directly in the AWS
        console, pre-Helix, or by another tool) — BEFORE the SCIM
        deprovision deletes the directory user. Once the user object is
        gone, those untracked assignments become un-attributable
        blank-name orphans that only :meth:`list_scope_orphan_assignments`
        can find and no longer tie back to the person. Sweeping while the
        principal is still alive closes that gap at the root.

        Contract:
          * Resolve ``email`` to the provider principal; if the principal
            no longer exists (SCIM already deleted it), return ``[]`` —
            there is nothing left to attribute, and the orphan-scan
            surface is the fallback for any residue.
          * Return one :class:`PrincipalAssignment` per
            (account/scope × permission-set) the principal holds. Only
            USER-typed principals; group membership is out of scope.
          * MUST return a COMPLETE list or RAISE. If any error would
            truncate the enumeration — a paginated ``list`` page fails, the
            principal→assignment index errors — RAISE; the sweep deletes
            every assignment in the returned list and reports success, so a
            silently-partial list would leave some assignments live while
            recording the leaver as fully swept (a false success). Only
            per-item *enrichment* that does not drop the item itself (e.g. a
            permission-set display-name lookup) may be tolerated by logging
            and continuing.

        Delete each returned assignment through
        :meth:`delete_orphan_assignment` (the payload generalises: a
        live-principal assignment and an orphan assignment share the
        ``(scope, permission-set, principal_id)`` delete shape).

        Default raises ``NotImplementedError`` so providers without an
        Identity-Center-style principal→assignment index are visibly
        opted out; the leaver-sweep treats that as "nothing to sweep".
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement "
            "list_principal_assignments",
        )

    async def list_billing_daily(
        self, start_date: date, end_date: date,
        *, scope: CloudScopeRef | None = None,
        include_owner_breakdown: bool = False,
        bq_billing: dict[str, Any] | None = None,
    ) -> list[CloudDailyCost]:
        """Daily cost rows for a range — one per (scope, date) unless
        ``scope`` is set in which case only that scope is returned.

        End date is INCLUSIVE. Return empty list when the provider
        doesn't support billing; do NOT raise — the sweep tolerates it.

        ``include_owner_breakdown`` — when True and ``scope`` is set,
        the driver performs an extra cost-allocation-tag query (AWS:
        iamPrincipal/Owner) and populates ``CloudDailyCost.owner_breakdown``
        for each returned row.  Ignored when ``scope`` is None (tenant-wide
        call for unregistered-account reconciliation) to avoid excess API
        calls.

        ``bq_billing`` — when the integration sources billing from a
        BigQuery-hosted CUR (currently AWS only), the cost-service layer
        pre-resolves the SA credentials + dataset and passes them here.
        Shape: ``{"credentials_json": dict, "dataset": "<proj>.<ds>",
        "table_pattern": "<acc>_aws_billing_data"}``. Drivers whose
        billing surface is the cloud-native API ignore this kwarg."""
        return []

    # ── Line-item cost facts (production-grade FinOps surface) ──────
    #
    # Returns one row per (date × resource × usage_type) — the granular
    # ground truth the cost-analytics dashboard, MCP tools, and
    # chargeback reports all read from. Drivers orchestrate whatever
    # vendor-specific calls give them line-item data:
    #   * Azure: Cost Management Query API with ResourceId grouping.
    #   * AWS: Cost Explorer GetCostAndUsageWithResources (≤14d window).
    #   * Aliyun: BSS DescribeInstanceBill.
    #   * Tencent: DescribeBillDetail.
    #   * GCP: BigQuery export (gcp_billing_export_v1_*).
    #
    # Each row is platform-agnostic (see ``CostFact`` below). Drivers
    # MUST be tolerant of partial line items — set NULL on dimensions
    # the provider doesn't expose for that line, never fabricate. The
    # ingestion service hashes the natural key and upserts.
    #
    # Drivers that don't support line-item data raise NotImplementedError;
    # the cron handler logs and skips for that integration. End date
    # INCLUSIVE.

    async def list_cost_facts(
        self, start_date: date, end_date: date,
        *, scope: CloudScopeRef | None = None,
        cny_per_usd: float | None = None,
        principal_tag_keys: list[str] | None = None,
        bq_billing: dict[str, Any] | None = None,
    ) -> list[CostFact]:
        """Per-resource line items for a date range.

        Drivers that ship line-item billing implement this; the rest
        raise ``NotImplementedError`` (the ingestion handler treats it
        as "skip this integration with reason='not_supported'", not
        as an error). End date INCLUSIVE.

        ``scope`` filters to one sub-account / subscription / project.
        For tenant-wide ingestion the cron passes None; the driver is
        free to call the provider's per-scope API in a loop or use a
        natively cross-scope query — caller doesn't care.

        ``cny_per_usd`` is the FX rate (CNY-per-USD, e.g. 7.20)
        resolved by the cost-ingestion service via
        :func:`app.core.fx_rate.resolve_cny_per_usd` once per run.
        Drivers whose provider bills natively in USD (AWS / Azure /
        GCP) may ignore it. CNY-native drivers (Aliyun / Tencent)
        require it and raise if the caller passes ``None``. Keeping
        FX resolution in the service layer means drivers stay
        stateless and DB-free, even when their billing API call is
        wrapped in ``asyncio.to_thread``.

        ``principal_tag_keys`` opts the driver into IAM-principal-tag
        cost slicing — for accounts where the primary cost driver has
        no taggable resource (Bedrock on-demand, shared Lambda, etc.)
        and the admin has activated AWS IAM principal tags as cost-
        allocation tags. When non-empty, the driver SHOULD return
        rows at (date × service × principal-tag-value) granularity
        with ``tags[<key>]`` populated, instead of the default
        per-resource granularity. Resolved by the ingestion service
        from ``platform_settings.cost_facts.aws_principal_tag.<slug>``
        once per run; drivers that don't support principal-tag mode
        ignore it.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement list_cost_facts",
        )

    # ── Custom-dimension billing query (AWS tag, Azure subscription, …) ──

    @classmethod
    def supported_billing_dimensions(cls) -> frozenset[str]:
        """Canonical dimension names this driver accepts in the
        ``group_by`` argument of :meth:`list_billing_by_dimension`.

        The set is the contract surfaced to the API/UI — the layer
        responsible for dimension validation. ``tag:KEY`` is special:
        an admin-supplied tag key paired with the canonical ``tag:``
        prefix. Drivers that support tag dimensions return ``"tag"``
        in this set; the API treats any ``"tag:KEY"`` request as
        valid as long as ``"tag"`` is present.

        Default: empty — driver hasn't shipped dimension support yet.
        """
        return frozenset()

    async def list_billing_by_dimension(
        self,
        start_date: date,
        end_date: date,
        *,
        group_by: list[str],
        scope: CloudScopeRef | None = None,
        filters: dict[str, list[str]] | None = None,
        bq_billing: dict[str, Any] | None = None,
    ) -> list[BillingByDimensionRow]:
        """Return per-day cost broken out by ``group_by`` dimensions.

        ``group_by`` accepts canonical names from
        :meth:`supported_billing_dimensions` plus ``"tag:KEY"`` for
        tag-based grouping (the cloud provider's tag key, NOT a Helix
        column). Two-dimension queries (e.g. ``["service", "tag:Project"]``
        for AWS) return rows whose ``dimensions`` dict carries both keys.

        ``filters`` is an optional ``{dimension: [allowed values]}``
        map; rows not matching are excluded by the provider before the
        sum is computed. Use this to drill into one subscription, one
        service, one tag value, etc.

        ``bq_billing`` — see :meth:`list_billing_daily` for the contract.
        Carries the BQ source's SA credentials + dataset for drivers
        whose billing surface is a BigQuery-hosted CUR (AWS).

        End date is INCLUSIVE. Drivers that don't support a particular
        dimension raise ``ValueError`` (the API translates to 400).
        Drivers that ship no dimension support raise
        ``NotImplementedError`` — the API surfaces this as 501 so the
        UI can disable the feature for that integration.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement "
            "list_billing_by_dimension",
        )

    async def billing_daily_precheck(self) -> str | None:
        """Return a user-facing reason why billing daily can't run.

        Called by the lifecycle service before iterating scopes. When
        non-None, the billing_daily_sync job short-circuits with
        ``status=skipped`` and this reason is surfaced in the scheduled-
        jobs history UI. Use this for obvious config gaps — an unset
        billing table, an unsupported provider feature, a missing pip
        dependency — that would otherwise produce a silent ``upserted:
        0`` result.

        Default returns None: drivers that don't override are attempted
        and their per-scope results are what they are."""
        return None

    async def get_last_login(
        self, ctx: CloudAccessContext,
    ) -> datetime | None:
        """Last time this principal logged into the console / used the
        scope. For IAM mode, drivers should check both ``PasswordLastUsed``
        and the newest access-key ``LastUsed``. For SSO mode, if the
        provider doesn't expose last-login, return None (the inactivity
        sweep will treat it as "unknown" and skip)."""
        return None

    async def revoke_account(
        self, ctx: CloudAccountRevokeContext,
    ) -> ProvisioningResult:
        """Disable / delete a per-user account at this cloud provider.

        Optional — only providers that support IAM-level deletion
        implement this. The rest raise ``NotImplementedError`` and the
        offboarding scan marks the blocker as needing manual handling.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not support user revocation",
        )

    async def close_scope_on_provider(
        self, scope: CloudScopeRef,
    ) -> CloseScopeResult:
        """Trigger provider-side closure of the scope.

        For multi-account clouds (AWS member account, Azure subscription,
        GCP project, Aliyun Resource Directory member): calls the
        provider's account-close API. The account enters a provider-side
        suspension period (90/30 days depending on cloud) before
        permanent deletion.

        For single-tenant IAM clouds where the scope is the tenant
        connection itself (Tencent CAM, UCloud IAM, AWS standalone IAM):
        evacuates every IAM user on the scope by calling
        :meth:`revoke_account` for each. The cloud account itself is
        NOT closed — that's a billing operation the customer makes via
        their cloud provider's console — but every Helix-managed IAM
        principal is removed so no one can log in via Helix's path.

        Drivers MUST raise :class:`NotImplementedError` if they cannot
        deliver real cloud-side closure for the given ``scope.scope_type``.
        The lifecycle service surfaces this as a 501 to the admin so the
        decommission button never silently no-ops.

        Implementations MUST be idempotent: when the provider says the
        scope is already closed / removed / deleted, return
        ``ok=True, already_closed=True`` rather than raising.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement "
            "close_scope_on_provider",
        )

    async def preflight_close_scope(
        self, scope: CloudScopeRef,
    ) -> ClosePreflight:
        """Read-only: what would make :meth:`close_scope_on_provider` fail?

        Pure read — this MUST never mutate anything at the provider. It
        runs when the admin opens the 下线 dialog, so the dialog can
        state the full cost of the decommission ("this subscription
        still holds 1 resource group containing storage account
        billingexport123 — they will be deleted") instead of the admin
        discovering it as a 502 after confirming.

        Two things it is deliberately NOT:

        * **Not a gate.** A driver returning blockers does not veto the
          decommission; the service still calls
          ``close_scope_on_provider`` and still believes the provider
          over this read. Preflight is a snapshot and can be stale by
          the time the admin confirms.
        * **Not a promise of completeness.** A cloud can refuse a
          closure for reasons no enumeration API exposes. Absence of
          blockers means "nothing we can see", which is why the close
          path keeps every one of its error branches.

        Default: ``supported=False`` — "this driver did not look".
        """
        del scope
        return ClosePreflight(supported=False)

    async def clear_close_blockers(
        self, scope: CloudScopeRef, *, keys: list[str],
    ) -> ClearBlockersResult:
        """Delete the confirmed blockers so the close can proceed.

        Called only from inside a decommission the admin already
        confirmed, and only with ``keys`` the admin was shown by
        :meth:`preflight_close_scope`. **Drivers MUST NOT delete
        anything outside ``keys``** — that invariant is what makes
        automatic cleanup safe: something created after the dialog was
        rendered can never be swept up silently by it, because its key
        was not in the confirmed set. The service re-reads the live
        blocker set on every retry and stops for re-confirmation when a
        key appears that nobody has seen.

        Implementations MUST be idempotent (a key that is already gone
        is ``completed``, not ``failed``) — the resume job calls this
        repeatedly while deletions are in flight.

        Default: ``ok=False`` with ``error_code='cleanup_unsupported'``.
        """
        del scope, keys
        return ClearBlockersResult(
            ok=False,
            error_code="cleanup_unsupported",
            error_message=(
                f"Cloud driver {self.provider!r} cannot clear "
                "close-blockers automatically"
            ),
        )

    async def get_scope_closure_state(
        self, scope: CloudScopeRef,
    ) -> ScopeClosureState:
        """Re-read the provider's CURRENT view of an already-closed scope.

        Pure read — this must never mutate anything at the provider. It
        exists because :meth:`close_scope_on_provider` returns the moment
        the closure is *submitted*, while the provider then works on it
        asynchronously for weeks: AWS holds a member account 90 days in
        ``PENDING_CLOSURE``, GCP 30 days in ``DELETE_REQUESTED``, Aliyun
        runs a check-then-silence-period sequence. Helix writes
        ``status='closed'`` at submission time, so without this read
        there is no point at which Helix ever learns the closure
        finished — or failed, or was reversed by a human in the
        provider's console (``projects.undelete`` and Azure subscription
        re-activation are both self-service for the entire window).

        Drivers whose closure is synchronous and has no provider-side
        after-state — the IAM-evacuation clouds, where the "closure" is
        a set of DeleteUser calls that either happened or didn't — raise
        :class:`NotImplementedError`. The reconciler treats that as
        "nothing to reconcile" and skips the scope permanently rather
        than logging an error every run.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement "
            "get_scope_closure_state",
        )

    async def get_resource_creator(
        self,
        resource_id: str,
        *,
        resource_type: str | None = None,
        scope: CloudScopeRef | None = None,
    ) -> ResourceCreatorInfo | None:
        """Resolve "who created this resource" from the provider's
        audit log.

        Returns :class:`ResourceCreatorInfo` on success, ``None`` if
        the audit log returned no matching event (typically the
        resource pre-dates the audit retention window). Drivers MUST
        raise :class:`ResourceCreatorWindowExceeded` when the
        provider explicitly signals the lookup window was the cause
        (vs an empty result that could mean either "no event" or
        "outside window"), so the service layer can distinguish a
        transient retry from a permanent miss.

        Drivers that haven't shipped audit-log support raise
        :class:`NotImplementedError`; the service layer surfaces
        this as ``status='unsupported'`` on the cache row so the
        cron sweep stops retrying.

        Audit-window caveats per provider:

        * AWS CloudTrail: **90 days** for the management-events
          trail by default. Configure a long-term Trail to S3 for
          older history. Rate limit: 2 req/s per region.
        * GCP Cloud Audit Logs: **400 days** default
          (``_Default`` log bucket), well beyond most needs.
        * Aliyun ActionTrail: **90 days** default unless a trail
          ships events to OSS for longer retention.
        * Azure Activity Log: **90 days** default; Diagnostic
          Settings can export to a Storage Account for long-term
          retention.
        """
        raise NotImplementedError(
            f"Cloud driver {self.provider!r} does not implement "
            "get_resource_creator",
        )

    async def close(self) -> None:
        """Release underlying SDK resources (sessions, clients)."""


# ─── Software provisioning (seat + permission grants) ───────────────


@dataclass(slots=True)
class ProvisioningContext:
    user_email: str
    user_display_name: str | None = None
    plan_name: str | None = None
    permission_key: str | None = None
    # Optional successor used when the vendor requires data handoff
    # (Zoom meetings/recordings, Google Drive/Calendar ownership, …).
    # Populated from ``ProvisioningTask.context_data`` by the
    # dispatcher; drivers that don't need transfer ignore it.
    successor_email: str | None = None
    successor_display_name: str | None = None
    extra: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class ProvisioningResult:
    ok: bool
    external_ref: str | None = None
    error_message: str | None = None
    raw_response: dict[str, Any] | None = None
    # Set True when the operation could not run because the vendor plan /
    # tier does not expose the required API (e.g. Slack Pro has no SCIM,
    # GitHub org without Enterprise-Cloud SAML SCIM). This is NOT a
    # transient error and NOT something a retry fixes — the work must be
    # done by a human. ``ProvisioningService.execute_task`` reads this flag
    # and downgrades the task to ``manual_notify`` (notify + await manual
    # completion) instead of marking it FAILED, so a plan-gated revoke no
    # longer permanently wedges an offboarding process.
    capability_unsupported: bool = False


class SoftwareProvisioningDriver(IntegrationDriver):
    """Drivers that grant/revoke seats or permissions on software."""

    kind = "software_provisioning"

    # What the connector supports — determines which operations the
    # provisioning service will dispatch to it.
    supports_seat: bool = False
    supports_permission: bool = False

    # Whether revoke_seat requires a successor (``ctx.successor_email``)
    # to hand off vendor-side data before the seat can be closed. Zoom
    # transfers meetings/webinars/cloud-recordings to the successor;
    # Google Workspace transfers Drive and Calendar ownership via the
    # Admin Data Transfer API. Drivers that just flip a SCIM flag
    # (Slack, GitHub) stay at the default ``False``.
    requires_successor_on_revoke: bool = False

    # Whether the provider exposes a full-access revoke that offboarding
    # can call ONCE to strip a leaver's ENTIRE footprint — every group /
    # entitlement, not just one bound product group (which can't cover
    # access granted through untracked groups). Jira sets this True and
    # implements :meth:`revoke_all_access` (removes the account from
    # every group it belongs to; the account itself is preserved).
    # Providers whose access dies with the SSO account (Google/Feishu
    # cascade) leave it at the default ``False``.
    supports_full_access_revoke: bool = False

    # Whether the provider exposes a *per-product* access revoke — i.e. an
    # action that strips one specific product family's seat from a still-
    # employed user without touching their other product accesses on the
    # same vendor and without touching org/identity groups. Only Jira /
    # Atlassian today: a Jira user's product license can be conferred by
    # any of many "conferring groups" (``/rest/api/3/applicationrole``
    # ``groups`` list), and removing them from just the connector's bound
    # ``group_name`` is insufficient when they hold the product via other
    # conferring groups (jira-administrators, team-specific groups, …).
    # Drivers that flip this True must implement :meth:`revoke_product_access`
    # and resolve product_key + sites from ``ctx.extra``. Pairs with the
    # ``target_type='product_access'`` provisioning task path; the
    # provisioning coordinator routes self-cancel / admin-revoke / idle-
    # reclaim through this when True, instead of the single-group
    # ``revoke_seat``.
    supports_product_access_revoke: bool = False

    @classmethod
    def revoke_requires_successor(cls, binding_config: dict[str, Any] | None) -> bool:
        """Given a connector binding's ``binding_config`` JSON, decide
        whether a revoke must carry a successor.

        Default delegates to the class-level
        ``requires_successor_on_revoke`` flag. Drivers whose need is
        per-operation override this (Google: only ``user_suspend``
        needs Drive/Calendar transfer; Zoom: only ``action_on_offboard
        = delete`` needs meeting/recording transfer).
        """
        return cls.requires_successor_on_revoke

    async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        raise NotImplementedError(
            f"Driver {self.provider!r} does not support seat grant",
        )

    async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        raise NotImplementedError(
            f"Driver {self.provider!r} does not support seat revoke",
        )

    async def grant_permission(
        self, ctx: ProvisioningContext,
    ) -> ProvisioningResult:
        raise NotImplementedError(
            f"Driver {self.provider!r} does not support permission grant",
        )

    async def revoke_permission(
        self, ctx: ProvisioningContext,
    ) -> ProvisioningResult:
        raise NotImplementedError(
            f"Driver {self.provider!r} does not support permission revoke",
        )

    async def revoke_all_access(
        self, ctx: ProvisioningContext,
    ) -> ProvisioningResult:
        """Strip the leaver's ENTIRE access on this provider (idempotent).

        Only meaningful for providers where one bound seat/group removal
        can't cover all of a person's access — see
        ``supports_full_access_revoke``. Reads ``ctx.user_email`` to
        resolve the account, then removes every group / entitlement it
        holds. Must be idempotent: an already-stripped or absent account
        is a success, so the offboarding blocker closes cleanly on
        re-runs.
        """
        raise NotImplementedError(
            f"Driver {self.provider!r} does not support full-access revoke",
        )

    async def revoke_product_access(
        self, ctx: ProvisioningContext,
    ) -> ProvisioningResult:
        """Strip ONE product family's access from a still-employed user.

        For active employees who self-cancel (or admin-cancels on their
        behalf) a seat — they want the product access gone but their
        org/identity groups (``employee_in_china``, ``All users from G
        Suite``) and unrelated product accesses (Confluence when they're
        only cancelling Jira) preserved. The default ``revoke_seat`` —
        remove from one bound group — is insufficient on Atlassian
        because product licenses can be conferred by multiple groups
        (``/rest/api/3/applicationrole`` ``groups`` list); the default
        ``revoke_all_access`` is too aggressive (also strips identity
        groups).

        Implementations should resolve the product family's conferring
        groups (vendor API), intersect with the user's current groups,
        and DELETE that intersection. Must be idempotent: groups the
        user is already absent from count as success. ``ctx.extra``
        carries ``product_key`` and ``sites`` (list of site hosts) set
        by the dispatcher from the source connector binding.

        Returns ``ok=False`` if ANY removal failed for a real reason —
        the offboarding blocker / seat-cancel task stays open and the
        UI surfaces the discrepancy instead of silently closing.
        """
        raise NotImplementedError(
            f"Driver {self.provider!r} does not support product-access revoke",
        )


# ─── Email (transactional send) ────────────────────────────────────


@dataclass(slots=True)
class EmailMessage:
    to: list[str]
    subject: str
    body_html: str | None = None
    body_text: str | None = None
    cc: list[str] = field(default_factory=list)
    bcc: list[str] = field(default_factory=list)
    reply_to: str | None = None
    # Override sender for a specific message; otherwise driver's config
    # ``from_address`` is used.
    from_address: str | None = None
    from_name: str | None = None
    # Stable RFC Message-ID supplied by the durable notification outbox.
    # SMTP does not promise de-duplication by this header, but preserving it
    # across retries gives providers/operators a deterministic reconciliation
    # key and lets downstream systems that do deduplicate avoid duplicates.
    message_id: str | None = None


@dataclass(slots=True)
class EmailSendResult:
    ok: bool
    provider_message_id: str | None = None
    error_message: str | None = None


class EmailDriver(IntegrationDriver):
    """Drivers that send transactional email.

    The platform picks whichever integration of ``kind='email'`` is
    ``is_primary=True`` and ``enabled=True``. SMTP is the first provider;
    SES / SendGrid / Postmark can be added as sibling drivers later.
    """

    kind = "email"

    @abstractmethod
    async def send(self, msg: EmailMessage) -> EmailSendResult: ...


# ─── MDM (Apple device management — Jamf today, Intune/Kandji later) ─


@dataclass(slots=True)
class MdmDevice:
    """One enrolled device pulled from an MDM (a Jamf computer or mobile
    device). Platform-agnostic — drivers flatten the provider's shape.

    * ``external_id`` — provider inventory id (Jamf computer/mobile id),
      stable handle used to address the device for read/detail calls.
    * ``management_id`` — the MDM management UUID, distinct from the
      inventory id. Modern MDM-command endpoints key off this, not
      ``external_id``; drivers populate it when available.
    * ``device_type`` — ``"computer"`` (macOS) | ``"mobile"`` (iOS/iPadOS/tvOS).
    * ``managed`` — whether the device is currently MDM-managed (an
      enrolled-but-unmanaged record still consumes inventory but can't
      receive commands).
    * ``assigned_user_email`` / ``assigned_user_name`` / ``assigned_username`` —
      the device's "User and Location" assignment, best-effort. NULL when
      the device has no assigned user (shared/lab device).
    * ``last_contact_at`` — last time the device checked in. NULL when the
      provider didn't return it.
    """

    external_id: str
    device_type: str
    management_id: str | None = None
    name: str | None = None
    serial_number: str | None = None
    model: str | None = None
    os_version: str | None = None
    managed: bool = True
    assigned_user_email: str | None = None
    assigned_user_name: str | None = None
    assigned_username: str | None = None
    last_contact_at: datetime | None = None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class MdmSeatUsage:
    """Seat-consumption snapshot: an MDM seat == one managed device.

    ``total_devices`` is the contract-relevant number (managed computers +
    mobile devices). Drivers source these from the provider's inventory
    ``totalCount`` rather than paging the whole fleet, so this stays a
    cheap, single-call probe usable on a dashboard tile.
    """

    total_devices: int
    computers: int
    mobile_devices: int


@dataclass(slots=True)
class MdmActionResult:
    """Outcome of an MDM device command (lock / wipe / clear-assignment).

    * ``already_done`` — idempotency flag: the device was already in the
      target state (e.g. command already queued, user already cleared).
      The caller treats it as success and skips re-notification.
    """

    ok: bool
    external_ref: str | None = None
    error_message: str | None = None
    already_done: bool = False
    raw_response: dict[str, Any] | None = None


class MdmDriver(IntegrationDriver):
    """Apple-device MDM providers (Jamf Pro today; Kandji / Mosyle / a
    dedicated Intune-MDM driver could follow as siblings).

    The interface is deliberately split into a **read** surface (device
    inventory + seat usage — what "manage seats" needs) and an **action**
    surface (lock / wipe / clear-assignment — what offboarding needs).
    Drivers that only ship read raise ``NotImplementedError`` on the
    action methods; the caller surfaces a clear "not supported" rather
    than silently no-op'ing a destructive command.
    """

    kind = "mdm"

    @abstractmethod
    async def list_devices(self) -> list[MdmDevice]:
        """Every enrolled device (computers + mobile devices) with its
        assigned user. Used for seat reconciliation and asset matching."""

    async def get_seat_usage(self) -> MdmSeatUsage:
        """Cheap seat-count probe (managed device totals). Override with a
        ``totalCount``-only call so a dashboard tile doesn't page the fleet."""
        raise NotImplementedError(
            f"MDM driver {self.provider!r} does not implement get_seat_usage",
        )

    async def lock_device(
        self, *, device_type: str, external_id: str,
        message: str | None = None,
    ) -> MdmActionResult:
        """Remote-lock one device. Idempotent: a re-issued lock command
        returns ``already_done=True``."""
        raise NotImplementedError(
            f"MDM driver {self.provider!r} does not implement lock_device",
        )

    async def wipe_device(
        self, *, device_type: str, external_id: str,
    ) -> MdmActionResult:
        """Remote-erase (factory reset) one device. IRREVERSIBLE — callers
        gate this behind explicit confirmation / an offboarding policy."""
        raise NotImplementedError(
            f"MDM driver {self.provider!r} does not implement wipe_device",
        )

    async def clear_device_assignment(
        self, *, device_type: str, external_id: str,
    ) -> MdmActionResult:
        """Unbind the assigned user from a device's User-and-Location
        record (the leaver "解绑" step). Idempotent: a device with no
        assigned user returns ``already_done=True``."""
        raise NotImplementedError(
            f"MDM driver {self.provider!r} does not implement "
            "clear_device_assignment",
        )


# ─── GitHub organization membership ────────────────────────────────


@dataclass
class GithubOrgMember:
    """One member of a GitHub organization, as the provider reports it.

    ``login`` is the natural key on this surface — GitHub states org
    membership in handles, and a handle does not always map to a person.
    ``user_id`` is the numeric account id, which survives a rename; a
    driver populates it when the call that produced the row carried it,
    and callers must treat ``None`` as "not asked", never as "no id".

    ``state`` mirrors GitHub's own vocabulary (``active`` / ``pending``),
    NOT Helix's ledger states. The translation happens one layer up, so
    that a future provider quirk does not silently redefine what a Helix
    ``active`` means.
    """

    login: str
    state: str
    role: str
    user_id: int | None = None


@dataclass
class GithubOrgInvitation:
    """A pending organization invitation.

    ``invitation_id`` is what cancelling one needs, and it is NOT derivable
    from the login — the invite call does not return it, so a driver that
    wants to cancel must list first. ``login`` can be ``None``: GitHub
    allows inviting by email, and such rows have no handle at all.
    """

    invitation_id: str
    login: str | None
    email: str | None
    role: str
    created_at: str | None = None
    #: GitHub's own failure note on an invitation it could not deliver
    #: (``failed_reason`` on the provider object, e.g. a bounced email).
    #: Distinct from expiry: GitHub expires an invitation silently after
    #: 7 days and says nothing here, so the caller derives expiry from
    #: ``created_at`` + ``GITHUB_INVITATION_TTL_DAYS`` rather than waiting
    #: for a reason that never arrives.
    failed_reason: str | None = None


@dataclass
class GithubUserProfile:
    """一个 GitHub handle 的公开个人资料。

    存在的理由很具体:**handle 字符串本身几乎没有识别信号。** 实测 217 个
    「认不出是谁」的 handle,归一后与任何企业邮箱前缀字面相同的是 **0 个**
    —— 它们长成 ``Ehco1996`` / ``zz-jason`` / ``hawkingrei``。

    信号在个人资料里:同一批抽样 30 个,21 个填了真实姓名,12 个填了
    company,10 个公开了邮箱,其中一个直接就是企业邮箱。而 company 写着
    别家公司或基金会的那些,恰好回答了另一个问题 —— 「这个人本来就不是
    我们的员工」,那是一条能让待办队列收敛的结论,不是一次匹配失败。
    """

    login: str
    user_id: int | None = None
    name: str | None = None
    company: str | None = None
    email: str | None = None
    bio: str | None = None
    location: str | None = None
    blog: str | None = None
    #: ``User`` / ``Bot`` / ``Organization``。``Bot`` 直接就是答案:
    #: 机器账号不需要被认成某个人。
    account_type: str | None = None

    def as_dict(self) -> dict[str, Any]:
        """落库用。刻意只存这几个字段 —— 个人资料是他人数据,存下超出
        「判断这是不是我们的员工」所需的部分没有正当理由。"""
        return {
            "login": self.login,
            "user_id": self.user_id,
            "name": self.name,
            "company": self.company,
            "email": self.email,
            "bio": self.bio,
            "location": self.location,
            "blog": self.blog,
            "account_type": self.account_type,
        }

    @property
    def has_signal(self) -> bool:
        """有没有任何可用于判断的内容。全空的资料和「没拉到」在界面上要
        长得不一样,否则人会一直重试一个注定为空的查询。"""
        return any(
            (self.name, self.company, self.email, self.bio, self.location),
        )


@dataclass
class GithubOrgSeatUsage:
    """一个组织的坐席用量。

    ``GET /orgs/{org}`` 的 ``plan`` 块就能读到,只要凭证能读这个组织 ——
    ``members:write`` 的 App 也够。所以坐席不足是**可以主动监控**的,
    不必等一次邀请失败才知道。

    ``has_cap`` 是这里唯一需要小心的字段。实测(2026-08-31,生产 5 个组织):

    * ``team`` 计划:``filled_seats`` / ``seats`` 都是真实的,余量有意义;
    * ``enterprise``:数字是**整个 enterprise** 的(tikv 报 9901/50000,那是
      CNCF 的池子),不是这个组织的账单,也不是我们能动的东西;
    * ``free``:``seats`` **不是上限**。PingCAP-QE 报 ``seats=92`` 而
      ``filled_seats=203`` —— 按字面算余量是 -111,会变成一个永久假警报。
      而一个永远亮着的警报,只会训练人忽略它。

    所以 ``has_cap`` 由计划类型决定,``remaining`` 在没有真实上限时是
    ``None`` —— 调用方看到 None 就该跳过,而不是当成 0。
    """

    plan: str | None
    filled_seats: int | None
    total_seats: int | None

    #: 这个计划的坐席数是不是一个真实的、我们能动的上限。
    has_cap: bool = False

    @property
    def remaining(self) -> int | None:
        if not self.has_cap:
            return None
        if not isinstance(self.total_seats, int) or not isinstance(
            self.filled_seats, int
        ):
            return None
        return self.total_seats - self.filled_seats


@dataclass
class GithubOrgActionResult:
    """Outcome of one membership mutation.

    Four fields, each answering a question the caller cannot answer for
    itself and must not guess:

    * ``already_done`` — the org was already in the target state (the
      person was already a member, or already absent). A success, and the
      caller skips re-notifying.
    * ``invited`` — the call created a PENDING invitation rather than an
      active membership. GitHub org membership is invitation-based, so
      "the API returned 200" does NOT mean the person is in the org.
      Collapsing the two is how a ledger ends up claiming access that
      nobody has.
    * ``not_found`` — the provider said the handle does not exist. This is
      a real, actionable answer (a typo'd handle in an approved request),
      and it is deliberately distinct from ``ok=False`` with a transport
      error, which says nothing about the world.
    * ``permission_denied`` — our credential is not an owner of this org.
      Distinct for the same reason: it is a configuration problem, not a
      fact about the user.
    """

    ok: bool
    external_ref: str | None = None
    error_message: str | None = None
    already_done: bool = False
    invited: bool = False
    not_found: bool = False
    permission_denied: bool = False
    #: 组织的坐席用满了。**与 ``permission_denied`` 分开**,尽管 GitHub 用
    #: 同一个 403 表达两者:凭证权限不足要去修集成,坐席用满要去**加钱买
    #: 坐席**,而后者只有 org owner / billing manager 能做。合并成一个
    #: 「403」会让一条本该发给 org owner 的「去加坐席」变成一条发给 IT 的
    #: 「去查凭证」。
    seat_limit: bool = False


class GithubOrgDriver(IntegrationDriver):
    """GitHub organization membership — invite, remove, enumerate.

    A **separate kind** from ``software_provisioning``'s ``github`` driver,
    which does seats over SCIM. They are not variants of one thing:

    * different API surface (SCIM 2.0 vs the org REST endpoints),
    * different credential (``scim:enterprise`` vs ``admin:org`` / a
      GitHub App installation),
    * different semantics — a SCIM seat is provisioned and is billable;
      an org membership is **offered** and only exists once the person
      accepts. A driver contract that promised both would have to lie
      about one of them.

    Every method takes ``org`` explicitly rather than reading it from the
    integration config. One credential (a GitHub App installed on several
    orgs) legitimately covers many organizations, and pinning the org to
    the integration row would force one integration per org and one more
    place for the two lists to disagree.
    """

    kind = "github_org"

    @abstractmethod
    async def list_members(
        self, org: str, *, role: str | None = None,
    ) -> list[GithubOrgMember]:
        """Every ACTIVE member of the organization.

        Pending invitations are NOT included — they are a different
        question with a different answer shape; see
        :meth:`list_pending_invitations`.
        """

    @abstractmethod
    async def list_pending_invitations(
        self, org: str,
    ) -> list[GithubOrgInvitation]:
        """Invitations sent and not yet accepted, declined or expired."""

    @abstractmethod
    async def invite_member(
        self, org: str, login: str, *, role: str = "member",
    ) -> GithubOrgActionResult:
        """Invite (or re-invite) one handle to the organization.

        MUST be idempotent: inviting somebody who is already an active
        member returns ``ok=True, already_done=True``, and re-inviting
        somebody with an outstanding invitation returns
        ``ok=True, invited=True`` without erroring.
        """

    @abstractmethod
    async def remove_member(
        self, org: str, login: str,
    ) -> GithubOrgActionResult:
        """Remove one handle from the organization, cancelling any pending
        invitation.

        MUST be idempotent: removing somebody who is not in the org
        returns ``ok=True, already_done=True``. It MUST NOT report success
        when the lookup that decided "already absent" failed at the
        transport layer — that is the platform-wide driver rule, and here
        it is the difference between "we removed their access" and "we
        never reached GitHub".
        """

    async def cancel_invitation(
        self, org: str, invitation_id: str,
    ) -> GithubOrgActionResult:
        """Withdraw a pending invitation without touching an active
        membership. Optional — drivers that cannot do it raise."""
        raise NotImplementedError(
            f"GitHub org driver {self.provider!r} cannot cancel invitations",
        )

    async def get_user(self, login: str) -> GithubOrgMember | None:
        """Does this handle exist on GitHub? ``None`` = the provider said
        no. Raises on transport failure — "we could not ask" and "the
        answer is no" must not arrive as the same value."""
        raise NotImplementedError(
            f"GitHub org driver {self.provider!r} cannot look up users",
        )

    async def list_teams(self, org: str) -> list[dict[str, Any]]:
        """Teams in the organization (``slug`` / ``name`` / ``id``)."""
        raise NotImplementedError(
            f"GitHub org driver {self.provider!r} cannot list teams",
        )

    async def get_user_profile(self, login: str) -> GithubUserProfile | None:
        """这个 handle 的公开个人资料。``None`` = 提供方说没有这个人。

        与 :meth:`get_user` 分开而不是把它加宽:``get_user`` 的语义是
        「这个 handle 存不存在」,调用方(审批执行路径)只关心存在性,
        每次都拉一份画像回来是它不需要的数据。传输层失败同样必须抛,
        不能和「不存在」共用一个返回值。
        """
        raise NotImplementedError(
            f"GitHub org driver {self.provider!r} cannot read user profiles",
        )

    async def get_seat_usage(self, org: str) -> GithubOrgSeatUsage:
        """这个组织的坐席用量。见 :class:`GithubOrgSeatUsage` 关于
        ``has_cap`` 的说明 —— free / enterprise 计划的数字不是我们的上限。"""
        raise NotImplementedError(
            f"GitHub org driver {self.provider!r} cannot read seat usage",
        )


# ─── Multi-cloud scope + grant lifecycle ───────────────────────────


class CloudPrincipalNotFoundError(ValueError):
    """Raised by an SSO-mode driver when the email-keyed principal
    doesn't exist in the tenant IdP.

    The lifecycle service catches this and turns it into a 409 with a
    message pointing at the SCIM / tenant-admin onboarding step.
    """


@dataclass(slots=True)
class CloudScopeInfo:
    """Result of ``list_scopes`` — one sub-account / subscription /
    project / RAM member / standalone container.

    * ``scope_type`` matches the DB enum (``aws_account``,
      ``azure_subscription``, ``gcp_project``, ``aliyun_account``,
      ``standalone_iam``).
    * ``external_id`` is the provider-side ID (AWS 12-digit account
      number, Azure subscription GUID, GCP project id, Alicloud account
      UID).
    * ``status`` reflects the scope's state at the provider
      (``active`` / ``suspended`` / ``closed``) — synced into
      ``cmdb_cloud_scopes.status`` (unless the admin has manually
      closed the scope in Helix).
    """

    scope_type: str
    external_id: str
    display_name: str | None = None
    status: str = "active"
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True, frozen=True)
class CloudScopePartitionStatus:
    """Authority state for one independently-enumerated scope type.

    ``complete=False`` means the rows returned by ``list_scopes`` MUST NOT be
    used to infer absence for this scope type.  ``error`` is a redacted,
    operator-facing reason; it must never contain provider credentials.
    """

    complete: bool
    error: str | None = None


@dataclass(slots=True)
class CloudScopeRef:
    """Lightweight addressable handle for a scope — what a driver needs
    to do a scope-bound call without re-loading the whole row."""

    scope_type: str
    external_id: str
    display_name: str | None = None


@dataclass(slots=True)
class CloseScopeResult:
    """Outcome of :meth:`CloudDriver.close_scope_on_provider`.

    * ``ok`` — the cloud provider accepted the closure (or the scope was
      already closed). ``False`` means Helix MUST NOT mark the scope as
      closed — the admin will see ``error_message`` and either fix the
      precondition or escalate.
    * ``already_closed`` — the scope was already in a closed/removed/
      deleted state at the provider before this call. Idempotency
      indicator; the caller treats it as success but skips notifications
      that imply a fresh close.
    * ``provider_state`` — short string describing the post-call cloud-
      side state (``pending_closure``, ``cancelled``, ``delete_requested``,
      ``removed_from_org``, ``users_evacuated``, …). Surfaced in audit
      and the success toast so the admin knows what happens next (e.g.
      "AWS will permanently delete this account in 90 days").
    * ``revoked_users`` — for IAM-evacuation scopes (Tencent / UCloud /
      AWS standalone IAM), the count of IAM users actually removed. For
      multi-account closures it stays ``0``.
    * ``error_code`` — provider-classified error code so the lifecycle
      service can map common failures (insufficient-permissions /
      precondition-not-met / payment-method-required) to actionable
      admin guidance. Free-form when the provider has no canonical code.
    """

    ok: bool
    already_closed: bool = False
    provider_state: str | None = None
    revoked_users: int = 0
    error_message: str | None = None
    error_code: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)
    raw_response: dict[str, Any] | None = None


@dataclass(slots=True)
class CloseBlocker:
    """One provider-side object standing between "close it" and closed.

    Azure refuses ``Microsoft.Subscription/cancel`` while the
    subscription still holds resources (``400
    ResourcesExistingOnSubscription``); GCP refuses ``projects.delete``
    while a lien is attached. Historically both came back to the admin
    as an error message with a CLI command in it — i.e. Helix asking
    the person who clicked 下线 to go perform the decommission by hand.
    This type exists so the same fact can instead be **shown before the
    click** and **cleared after it**.

    * ``key`` — the provider-addressable identity of the thing (ARM
      resource id, ``liens/…`` name). Stable across reads: it is what
      the admin confirms and the only thing
      :meth:`CloudDriver.clear_close_blockers` is allowed to delete.
      Compared case-insensitively.
    * ``kind`` — ``resource_group`` / ``resource`` / ``lock`` / ``lien``.
      Drives copy + icon, never behaviour.
    * ``children`` / ``child_count`` — what is INSIDE this blocker.
      A resource group is a container, and "1 个资源组" is not enough
      information to consent to a deletion; "1 个资源组，内含存储账号
      billingexport123" is. ``children`` is display-capped by the
      driver, ``child_count`` is the true total.
    * ``removable`` — can Helix delete this itself as part of the
      decommission? ``False`` means a human has to act and
      ``blocked_reason`` says why — the one case where handing the
      problem back is the only honest answer.
    """

    key: str
    kind: str
    name: str
    detail: str | None = None
    children: list[str] = field(default_factory=list)
    child_count: int = 0
    removable: bool = True
    blocked_reason: str | None = None


@dataclass(slots=True)
class ClosePreflight:
    """What :meth:`CloudDriver.preflight_close_scope` found.

    ``supported=False`` means this driver / scope_type has no preflight
    at all, so an empty ``blockers`` must NOT be read as "nothing in
    the way" — only as "we did not look". ``checked=False`` is the
    narrower version for a driver that does support preflight but could
    not reach the provider on this call (token failure, 5xx). Only
    ``supported and checked`` licenses the UI to say 没有需要清理的资源.
    Same discipline as ``ScopeClosureState``'s ``unknown``: "could not
    check" and "checked, it is clear" are exactly the two facts this
    mechanism exists to keep apart.

    ``cleanup_supported`` says whether :meth:`clear_close_blockers` can
    act on what was found. A driver may enumerate blockers it cannot
    delete — still a large improvement on a post-hoc error, because the
    admin learns the cost of the decommission before spending the
    confirmation.
    """

    supported: bool = False
    checked: bool = False
    cleanup_supported: bool = False
    blockers: list[CloseBlocker] = field(default_factory=list)
    provider_state: str | None = None
    error_code: str | None = None
    error_message: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class ClearBlockersResult:
    """Outcome of :meth:`CloudDriver.clear_close_blockers`.

    Cleanup is asynchronous on every cloud that has it: an Azure
    resource-group delete answers ``202 Accepted`` and the group then
    sits in ``provisioningState=Deleting`` for seconds to tens of
    minutes depending on what is inside it. So the honest return is not
    "done" but "accepted, and here is what is still moving":

    * ``started`` — keys where a delete was accepted on this call.
    * ``completed`` — keys the provider says are already gone.
    * ``failed`` — ``{"key", "error"}`` per key the provider refused.
      Non-empty ``failed`` does not force ``ok=False``: progress on the
      rest is still progress, and the retry loop re-reads reality.
    * ``in_progress`` — deletions are still in flight, so the close
      must be retried later rather than now. The scheduled resume job
      owns that retry; the admin's interaction is already over.
    """

    ok: bool
    started: list[str] = field(default_factory=list)
    completed: list[str] = field(default_factory=list)
    failed: list[dict[str, str]] = field(default_factory=list)
    in_progress: bool = False
    error_code: str | None = None
    error_message: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)


# Dispositions a provider can report for a scope Helix already closed.
# The vocabulary is deliberately tiny — the reconciler branches on it and
# every branch has to be a decision someone can act on.
#   ``pending``  the provider is still working (AWS PENDING_CLOSURE, GCP
#                DELETE_REQUESTED, Aliyun Checking/Deleting). Keep polling.
#   ``closed``   the provider is done. Stop polling.
#   ``alive``    the provider says this scope is USABLE AGAIN. Not a
#                theoretical state: GCP ``projects.undelete`` and Azure's
#                subscription re-activation are both documented,
#                self-service operations available for the whole grace
#                window, and neither tells Helix. Helix has already
#                disabled every grant on the row, so the account is now
#                live and outside Helix's governance — alert.
#   ``failed``   the provider reports the closure itself failed (Aliyun
#                ``CheckFailed`` / ``DeleteFailed``). The account is alive
#                and nothing further will happen on its own — alert.
#   ``unknown``  could not read. Change nothing, keep polling. NEVER
#                collapse this into ``closed``: "we could not check" and
#                "the provider confirmed it is gone" are the two facts
#                this whole mechanism exists to keep apart.
CLOSURE_DISPOSITIONS = ("pending", "closed", "alive", "failed", "unknown")


@dataclass(slots=True)
class ScopeClosureState:
    """Provider's CURRENT view of a scope Helix already decommissioned.

    Read-only counterpart to :class:`CloseScopeResult`. ``close_scope_on_
    provider`` starts a closure; on every cloud worth the name that
    closure then runs asynchronously for weeks (AWS 90d, GCP 30d, Aliyun
    silence period), and Helix's row is written the moment it is
    *submitted*. Without a read-back there is no moment at which Helix
    learns the closure actually completed — or silently failed, or was
    undone by someone in the provider's console.

    ``provider_state`` uses the same vocabulary as
    ``CloseScopeResult.provider_state`` so the column it lands in keeps
    one meaning across both writers.
    """

    disposition: str = "unknown"
    provider_state: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)
    error_message: str | None = None


@dataclass(slots=True)
class CloudRoleInfo:
    """Result of ``list_scope_roles``. ``scope`` may be None when the
    role is tenant-wide (AWS Identity Center permission sets).

    * ``is_promoted`` — does this role appear in the employee-facing
      request dropdown by default? Drivers set TRUE for admin-curated
      predefined entries and for every customer-/custom-defined role
      (a customer-managed AWS policy is, by definition, something the
      cloud admin chose to define — surfacing it is the right default).
      FALSE for vendor built-ins outside the curated subset (e.g. the
      ~380 Azure RBAC roles, ~200 Aliyun system policies, ~200 Tencent
      presets). Hidden behind an "include all built-ins" toggle on the
      request modal so employees aren't drowning in choices, while the
      catalog still mirrors what's available cloud-side for the admin
      promote/demote UI.
    """

    role_key: str
    display_name: str | None = None
    description: str | None = None
    is_custom: bool = False
    source: str | None = None
    scope: CloudScopeRef | None = None
    raw: dict[str, Any] = field(default_factory=dict)
    is_promoted: bool = True


@dataclass(slots=True)
class CloudAccessContext:
    """Input to ``grant_access`` / ``revoke_access`` / ``get_last_login``.

    * ``principal_email`` is the canonical Helix identity — drivers
      resolve it to a tenant-side principal_id as needed.
    * ``role_key`` matches a ``CmdbCloudRole.role_key`` — opaque to the
      service, meaningful to the driver.
    * ``existing_external_ref`` is the value the driver returned last
      time (Identity Center assignment request id, Azure assignment
      guid, GCP binding member, RAM role policy attachment id, IAM user
      name). Passed back on revoke so the driver can target the exact
      resource to undo.
    * ``ttl_days`` is informational (some providers offer native TTL —
      e.g. Azure PIM time-bound assignments — but Helix owns expiry in
      the service layer regardless).
    """

    principal_email: str
    principal_display_name: str | None
    scope: CloudScopeRef
    role_key: str
    role_display_name: str | None = None
    existing_external_ref: str | None = None
    ttl_days: int | None = None
    extra: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class CloudGrantResult:
    """Return shape for ``grant_access`` / ``revoke_access``.

    * ``external_ref`` — what to send back on revoke (assignment id /
      binding member / user name).
    * ``detail`` — driver-specific. IAM-mode drivers return AK/SK here
      (but service layer scrubs these after one-shot delivery).
    * ``already_exists`` / ``already_absent`` — idempotency flags. The
      service treats them as success and doesn't re-notify.
    """

    ok: bool
    external_ref: str | None = None
    error_message: str | None = None
    detail: dict[str, Any] = field(default_factory=dict)
    already_exists: bool = False
    already_absent: bool = False


class ResourceCreatorWindowExceeded(LookupError):
    """Audit log explicitly signalled the lookup window was the
    cause of an empty result.

    AWS CloudTrail / Aliyun ActionTrail / Azure Activity Log default
    to a 90-day window; resources created earlier can never be
    attributed via the default API. The service layer maps this
    exception to ``status='window_exceeded'`` on the cache row so
    the cron sweep stops retrying (vs a transient empty result that
    might succeed on a re-poll once the audit log catches up).
    """


@dataclass(slots=True)
class ResourceCreatorInfo:
    """One ``get_resource_creator`` resolution.

    Every field is platform-neutral; drivers flatten provider-
    specific shapes (CloudTrail ``userIdentity`` / GCP
    ``protoPayload.authenticationInfo`` / Aliyun ``UserIdentity`` /
    Azure ``caller``) before returning.

    * ``principal_id`` — IAM ARN / SSO ARN / UPN / RAM user id.
      Stored verbatim so an operator can later manually map a
      federated-identity ARN to a Helix Person.
    * ``email`` — best-effort extraction (CloudTrail SSO ARNs
      embed the email after ``user/``; GCP / Azure return it
      natively; Aliyun returns the RAM user name which the driver
      maps if it matches the standard ``firstname.lastname@``
      pattern). NULL when not extractable.
    * ``display_name`` — friendly name. NULL when the provider
      doesn't expose one.
    * ``created_at_cloud`` — UTC datetime the resource was created
      per the audit log (NOT when Helix observed it).
    * ``source`` — one of ``cloudtrail`` / ``gcp_audit`` /
      ``aliyun_actiontrail`` / ``azure_activity``. Persisted on the
      cache row for forensics.
    * ``raw`` — the full audit event payload, capped by the service
      layer at ~16 KiB before write.
    """

    principal_id: str
    email: str | None = None
    display_name: str | None = None
    created_at_cloud: datetime | None = None
    source: str = ""
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class CloudDailyCost:
    """One day of cost. ``scope`` None = tenant-level aggregate."""

    date: date
    total_cost_usd: float
    scope: CloudScopeRef | None = None
    cost_by_service: dict[str, float] = field(default_factory=dict)
    currency: str = "USD"
    raw: dict[str, Any] = field(default_factory=dict)
    # iamPrincipal/Owner (or equivalent) breakdown: {owner_tag_value: cost_usd}.
    # Populated by drivers that support cost-allocation tag queries (currently
    # AWS CE with iamPrincipal/Owner).  Empty dict = not available for this row.
    owner_breakdown: dict[str, float] = field(default_factory=dict)
    # ── Dual measures (2026-07 cost-view split) ──────────────────────
    # Business list cost (研发口径): whitelist-only sum at actual
    # executed prices (spot / tiered kept; company-level commitments,
    # contract discounts, credits, adjustments and tax stripped).
    # None = this driver / row does not produce the measure (drivers
    # MUST NOT fabricate — None flows through to a NULL column and the
    # engineering surfaces show "口径未接入" rather than a wrong 0).
    total_cost_usd_list: float | None = None
    # Public on-demand list price for the same whitelisted usage
    # (AWS pricing_public; GCP equals the list measure). Finance
    # derives engineer-earned savings as public − list.
    total_cost_usd_public: float | None = None


# Canonical ``CostFact.cost_bucket`` values. Cross-cloud vocabulary —
# each driver maps its provider-native row types onto these buckets so
# the finance waterfall (GROUP BY cost_bucket) is provider-agnostic.
# Anything a driver cannot confidently classify goes to OTHER (never to
# USAGE — USAGE rows are the only ones carrying the engineering list
# measure and must stay whitelist-pure).
COST_BUCKET_USAGE = "usage"
COST_BUCKET_COMMITMENT_FEE = "commitment_fee"          # SP/RI/CUD 承诺费
COST_BUCKET_COMMITMENT_DISCOUNT = "commitment_discount"  # SP/RI 抵扣行
COST_BUCKET_CONTRACT_DISCOUNT = "contract_discount"    # EDP/私有价/价内合同折扣
COST_BUCKET_CREDIT = "credit"                          # 赠金/促销/订阅权益
COST_BUCKET_ADJUSTMENT = "adjustment"                  # 调账/退款/发票级调整
COST_BUCKET_TAX = "tax"
COST_BUCKET_SUPPORT = "support"
COST_BUCKET_OTHER = "other"

COST_BUCKETS: frozenset[str] = frozenset({
    COST_BUCKET_USAGE,
    COST_BUCKET_COMMITMENT_FEE,
    COST_BUCKET_COMMITMENT_DISCOUNT,
    COST_BUCKET_CONTRACT_DISCOUNT,
    COST_BUCKET_CREDIT,
    COST_BUCKET_ADJUSTMENT,
    COST_BUCKET_TAX,
    COST_BUCKET_SUPPORT,
    COST_BUCKET_OTHER,
})


@dataclass(slots=True)
class CostFact:
    """One line item — per (date × resource × usage_type) cost row.

    Driver method :meth:`CloudDriver.list_cost_facts` returns these;
    the ``cmdb.cost_facts_sync`` cron upserts them into
    ``cmdb_cost_facts`` keyed by sha256 of the natural key.

    Every dimension is platform-agnostic; provider-specific shapes are
    flattened by the driver. NULL means "the provider didn't expose
    this for this line item" — drivers MUST NOT fabricate. The
    analytics layer treats NULL differently from empty string ("") for
    that reason: NULL = unknown, empty = the provider returned an
    empty string verbatim (rare, but happens on misconfigured tags).

    * ``date`` — UTC calendar day the cost was incurred.
    * ``scope`` — sub-account / subscription / project / RAM member
      the line belongs to. None for tenant-overhead lines (e.g. Azure
      "Cost Anomaly Detection" charged at directory level).
    * ``resource_group`` — Azure RG name / GCP folder / Aliyun member
      account / AWS account-or-OU pseudonym. NULL for tenant-overhead
      lines.
    * ``resource_id`` — provider-canonical resource id (full ARN /
      Azure resource id / GCP relative name / Aliyun InstanceID).
      NULL when the cost is not tied to a specific resource (some
      Azure / AWS support / overhead lines).
    * ``resource_name`` — human-readable resource name when the
      provider returns it inline. Cached so resource-detail drawers
      don't have to round-trip through the resource sync table.
    * ``service`` — MeterCategory / ServiceName ("Compute Engine",
      "EC2 - Other", "Azure App Service"). Always non-empty; drivers
      fall back to "Other" when unknown.
    * ``usage_type`` — SKU / meter sub-category. NULL when the
      provider doesn't break SKUs out at line-item level.
    * ``tags`` — billing-side tags as ``{key: value}``. The provider's
      authoritative tag map for this line item — cost-allocation
      rules match against this dict directly.
    * ``amount_usd`` — converted to USD by the driver before return.
      Driver MUST NOT pass through native currency without conversion;
      mixed-currency analytics rolls would give wrong totals. If the
      driver can't determine the FX rate, it returns the row with
      amount_usd=0.0 + ``amount_native`` populated and a non-USD
      ``currency``; the ingestor will skip it from USD aggregations
      but persist the row for audit.
    * ``amount_native`` — provider-native currency amount.
    * ``amount_usd_gross`` / ``amount_native_gross`` — pre-discount
      (list / 原价) amount in USD / native currency. None when the
      driver doesn't populate a gross price for this line. Wired today:
      AWS (BigQuery-CUR ``pricing/public OnDemandCost``) and Azure
      (Cost Management ``MeterId`` + ``UsageQuantity`` joined to the
      public Retail Prices API by meter GUID — forward-only; historical
      aggregated rows stay None as no per-meter quantity was stored).
      GCP (billing-export ``cost`` before credits) is NOT wired yet —
      its gross stays None pending a dedicated follow-up. Drivers MUST
      NOT fabricate; None flows through and the AI dashboard aggregates
      by coverage.
    * ``currency`` — ISO 4217 of ``amount_native``.
    * ``usage_quantity`` / ``usage_unit`` — billable units (vCPU·h /
      GB·mo / requests / tokens). Optional but useful for unit-cost
      analytics.
    """

    date: date
    integration_id: str
    service: str
    amount_usd: float
    amount_native: float
    currency: str
    amount_usd_gross: float | None = None
    amount_native_gross: float | None = None
    # ── Dual measures (2026-07 cost-view split) ──────────────────────
    # Business list cost (研发口径) for this line. Non-None ONLY on
    # whitelisted usage rows (AWS line_item_type ∈ {Usage,
    # SavingsPlanCoveredUsage} at unblended, RI DiscountedUsage
    # restored via pricing_public; GCP cost_at_list on regular non-
    # Support/Invoice rows). Discount / fee / credit / adjustment / tax
    # rows stay None so the engineering measure is a bare
    # SUM(amount_usd_list) — unknown new row types default OUT.
    amount_usd_list: float | None = None
    amount_native_list: float | None = None
    # Public on-demand price for the same whitelisted usage (AWS
    # pricing_public_on_demand_cost; GCP equals list). public − list =
    # engineer-earned savings (spot / tiered / actual-rate).
    amount_usd_public: float | None = None
    # Coarse row classification for the finance waterfall:
    # usage / commitment_fee / commitment_discount / contract_discount /
    # credit / adjustment / tax / support / other.
    cost_bucket: str = "usage"
    scope: CloudScopeRef | None = None
    resource_group: str | None = None
    resource_id: str | None = None
    resource_name: str | None = None
    resource_type: str | None = None
    region: str | None = None
    usage_type: str | None = None
    tags: dict[str, str] = field(default_factory=dict)
    usage_quantity: float | None = None
    usage_unit: str | None = None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class BillingByDimensionRow:
    """One row of cost broken out across one or more dimensions.

    Emitted by :meth:`CloudDriver.list_billing_by_dimension`. Unlike
    :class:`CloudDailyCost` which carries a fixed ``cost_by_service``
    breakdown, this carries an arbitrary ``dimensions`` map so the
    same shape can answer:

      * AWS: GROUP BY ``service`` + ``tag:Project``
      * Azure: GROUP BY ``subscription`` + ``service``
      * GCP: GROUP BY ``project`` + ``sku``

    Frontends consume the map as a column key — the API contract is
    "give me back whatever dimensions I asked to group by".

    Attributes:
        date: The day this cost was incurred.
        dimensions: Map of canonical dimension name → value, e.g.
            ``{"service": "EC2 - Other", "tag:Project": "tidb"}``.
            Keys mirror the ``group_by`` argument exactly so the
            caller can index back into its request.
        cost: Cost amount in ``currency``. Float (not Decimal) so the
            frontend doesn't have to render Decimal serialisation.
        currency: ISO 4217 (USD, CNY, …). Drivers that return native
            currency must set this; default USD is provider-dependent.
        raw: Original provider payload for the matching cell. Useful
            for audit + debugging unexpected drift.
    """

    date: date
    dimensions: dict[str, str]
    cost: float
    currency: str = "USD"
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class DiscoveredGrant:
    """A role/policy assignment the driver *found* on the cloud side.

    Emitted by :meth:`CloudDriver.list_scope_grants`. The lifecycle
    service turns each row where ``principal_type='user'`` + the email
    resolves to an active ``IdentityPerson`` into a
    ``CmdbCloudAccessGrant`` with ``source='discovered'``.

    Non-user principals (service principals, groups, managed identities)
    still flow through so the service can count / surface them in drift
    incidents — but they never become grant rows, because a Helix person
    doesn't exist behind them.

    * ``principal_email`` — canonical Helix identity key. SSO clouds
      return this directly from the tenant IdP (Google Workspace email,
      AAD UPN, Identity Center email). IAM-mode clouds (Tencent, AWS
      SG) return the IAM user's ``email`` attribute when set, else the
      IAM user name (no ``@``) — the service skips unresolvable rows
      instead of guessing.
    * ``principal_type`` — ``"user"`` | ``"service_principal"`` |
      ``"group"`` | ``"unknown"``. Only ``"user"`` becomes a grant row;
      the rest are counted for reporting.
    * ``role_key`` / ``role_display_name`` — same shape as
      :class:`CloudRoleInfo`; drivers reuse their role-catalog key so
      the pre-grant lookup into ``cmdb_cloud_roles`` works unchanged.
    * ``external_ref`` — provider-side handle. Same format the driver
      returns from :meth:`grant_access` so an admin can later
      :meth:`revoke_access` a discovered grant without rediscovery.
    * ``scope`` — which scope the grant is on. Kept here (not implicit)
      so multi-scope drivers can emit a single flat stream.
    * ``granted_at`` — when the provider says the assignment was made.
      None if the provider doesn't expose it. The service falls back to
      NOW() on insert.
    * ``raw`` — verbatim provider payload for debugging + audit.
    """

    principal_email: str
    principal_type: str
    role_key: str
    scope: CloudScopeRef
    role_display_name: str | None = None
    external_ref: str | None = None
    granted_at: datetime | None = None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class OrphanAssignment:
    """A cloud-side role/policy assignment whose principal cannot be
    resolved back to the tenant directory — i.e. AWS Identity Center
    shows ``"User not found (UUID)"`` next to a permission set on an
    account.

    Emitted by :meth:`CloudDriver.list_scope_orphan_assignments`. The
    typical cause is a SCIM-synced directory user being deleted
    (offboarded) while the per-account permission set assignment was
    never withdrawn — AWS keeps the assignment row indefinitely with a
    stale ``PrincipalId``.

    * ``principal_id`` — the cloud-side opaque id (Identity Store user
      UUID for AWS, AAD object id for Azure, etc.). Stable enough to
      pivot the cleanup call against; not human-readable.
    * ``role_key`` / ``role_display_name`` — same shape as
      :class:`CloudRoleInfo`; the permission set / role being held.
    * ``external_ref`` — the same string the driver would return from
      :meth:`grant_access`, suitable to feed back to
      :meth:`delete_orphan_assignment`. Format identical to
      ``DiscoveredGrant.external_ref`` so existing tooling generalizes.
    * ``recovered_email`` — best-effort email recovered from past
      Helix records (audit, prior grant rows). ``None`` when the
      principal was never SCIM-synced into Helix; the cleanup writes
      the row with email NULL in that case.
    * ``last_seen_at`` — when the assignment was last seen alive on
      the cloud side (today's sync, by definition; kept for symmetry
      with DiscoveredGrant). Drivers may set None.
    """

    principal_id: str
    role_key: str
    scope: CloudScopeRef
    role_display_name: str | None = None
    external_ref: str | None = None
    recovered_email: str | None = None
    last_seen_at: datetime | None = None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class PrincipalAssignment:
    """One account/permission-set assignment held by a *live* principal,
    resolved from their email.

    Emitted by :meth:`CloudDriver.list_principal_assignments` and consumed
    by the offboarding leaver-sweep. Unlike :class:`OrphanAssignment`
    (whose principal is already gone from the directory), this describes
    an assignment we can still positively attribute to a known person —
    which is exactly why the sweep deletes it *before* SCIM deprovision
    turns it into an un-attributable orphan.

    * ``account_id`` — the AWS Org member account id (the scope's
      ``external_id``). The sweep matches this against a
      :class:`CmdbCloudScope` to enforce the prod-write lock; assignments
      on accounts Helix doesn't track as a scope are surfaced but NOT
      deleted (no scope row means no environment classification to
      prod-gate against).
    * ``principal_id`` — the resolved Identity Store user id.
    * ``role_key`` / ``role_display_name`` — permission-set ARN + name.
    * ``external_ref`` — ``account_id::permission_set_arn::principal_id``,
      identical to the grant/orphan shape so
      :meth:`delete_orphan_assignment` accepts it unchanged.
    """

    account_id: str
    principal_id: str
    role_key: str
    role_display_name: str | None = None
    external_ref: str | None = None
    raw: dict[str, Any] = field(default_factory=dict)


# ─── Per-user cloud account revocation ─────────────────────────────


@dataclass(slots=True)
class CloudAccountRevokeContext:
    """Input to ``revoke_account`` — platform-agnostic IAM identifier.

    ``provider`` is informational (the driver's own ``.provider`` is
    authoritative). ``account_id`` is whatever the provider uses as the
    user handle — for AWS IAM that's the ``UserName``; for other clouds
    it might be an email or numeric ID. Drivers document their expected
    format in ``display_name``.
    """

    provider: str
    account_id: str
    # Arbitrary per-call knobs passed through so the
    # driver can read per-account knobs (e.g. "also_delete_mfa_devices").
    extra: dict[str, Any] = field(default_factory=dict)


# ─── Partner directory (合作方通讯录) ───────────────────────────────


@dataclass
class PartnerDirectoryMemberRecord:
    """One employee of a PARTNER organisation, as their directory reports it.

    "Partner" means a company that is not us but whose people hold
    accounts in our systems — CBG is the concrete case: their staff carry
    ``@pingcap.cn`` addresses in their own Lark tenant while holding
    ``@pingcap.com`` Google accounts and GitHub handles in ours. Those
    accounts show up in every unmatched/orphan bucket we have, and
    somebody has been classifying them by hand.

    The natural key is ``external_id`` — the partner directory's own
    stable user id (Feishu ``open_id``). NOT the email: a partner can
    rename a mailbox, and an email that moves between two of their people
    must not silently transfer the tag.

    ``email`` / ``secondary_email`` are kept apart because the matcher
    treats them differently: the primary (Feishu ``enterprise_email``) is
    the address their IT issued and is what the local-part rule keys on;
    the secondary (the ``email`` field) is frequently a personal address
    and is only ever used for an EXACT full-string match.
    """

    external_id: str
    #: Cross-tenant stable id (Feishu ``union_id``). Informational.
    union_id: str | None = None
    #: The partner's own HR-facing id (Feishu ``user_id`` / 工号).
    partner_user_id: str | None = None
    employee_no: str | None = None

    name: str | None = None
    en_name: str | None = None
    #: Work address issued by the partner's IT (Feishu ``enterprise_email``).
    email: str | None = None
    #: Feishu's ``email`` field — often personal. Exact matches only.
    secondary_email: str | None = None
    job_title: str | None = None
    department_ids: list[str] = field(default_factory=list)
    #: Human-readable department chain, outermost first.
    department_path: list[str] = field(default_factory=list)

    #: GitHub handle as the partner's directory records it (a custom
    #: attribute on their side). ``None`` = the field was absent or empty
    #: — never "they have no GitHub account", which we cannot know.
    github_login: str | None = None

    #: Still on the partner's payroll. Drivers compute this from whatever
    #: their provider calls it; the platform never re-derives it.
    is_active: bool = True
    #: Suspended but not departed (Feishu ``is_frozen``).
    is_suspended: bool = False
    #: Raw provider status flags, kept verbatim for triage.
    status_raw: dict[str, Any] = field(default_factory=dict)
    #: Other custom attributes, keyed by their display name (HRBP,
    #: Base 地, …). Stored so a future rule can key on one without a
    #: schema change.
    custom_attrs: dict[str, str] = field(default_factory=dict)


@dataclass
class PartnerDirectorySnapshot:
    """A whole roster pull, plus what the driver could not see.

    ``members`` alone cannot distinguish "the partner has 40 staff" from
    "we could only read 40 of them". ``departments_scanned`` /
    ``departments_unreadable`` carry that difference so the sync refuses
    to soft-delete rows off a partial read — the failure mode that would
    otherwise strip CBG tags from everybody the moment a scope lapses.
    """

    members: list[PartnerDirectoryMemberRecord] = field(default_factory=list)
    departments_scanned: int = 0
    departments_unreadable: int = 0
    #: True when every department the driver knew about was read
    #: successfully. Only a complete snapshot may retire absent rows.
    complete: bool = True
    notes: list[str] = field(default_factory=list)


class PartnerDirectoryDriver(IntegrationDriver):
    """Read-only roster of a partner organisation's staff.

    Deliberately read-only: we do not administer the partner's directory
    and must never be able to. Every write-shaped verb is absent from the
    interface rather than present-and-raising, so a future caller cannot
    reach for one by mistake.
    """

    kind = "partner_directory"

    @abstractmethod
    async def list_members(self) -> PartnerDirectorySnapshot:
        """The partner's full roster, including departed staff.

        Departed people are INCLUDED (flagged ``is_active=False``) on
        purpose: a GitHub handle left behind by someone who has already
        left the partner is exactly the row we most need to be able to
        name.
        """


__all__ = [
    "ApprovalSubmission",
    "ApprovalSubmissionResult",
    "ChatApprovalDriver",
    "CloseScopeResult",
    "CloudAccessContext",
    "CloudAccountRevokeContext",
    "CloudDailyCost",
    "CloudDriver",
    "CostFact",
    "DiscoveredGrant",
    "CloudGrantResult",
    "CloudPrincipalNotFoundError",
    "OrphanAssignment",
    "CloudRoleInfo",
    "CloudScopeInfo",
    "CloudScopeRef",
    "EmailDriver",
    "EmailMessage",
    "EmailSendResult",
    "HrisDriver",
    "IntegrationDriver",
    "MdmActionResult",
    "MdmDevice",
    "MdmDriver",
    "MdmSeatUsage",
    "PartnerDirectoryDriver",
    "PartnerDirectoryMemberRecord",
    "PartnerDirectorySnapshot",
    "ProvisioningContext",
    "ProvisioningResult",
    "SoftwareProvisioningDriver",
    "WebhookVerificationResult",
]