"""Microsoft Graph software-provisioning driver (M365 licenses + Intune).

One integration row = one Entra (Azure AD) tenant, authenticated via the
OAuth2 **client-credentials** flow (app registration with admin-consented
*application* permissions — no user login). Credentials:

  * ``tenant_id``     — Entra directory (tenant) id (GUID)
  * ``client_id``     — app registration's application (client) id
  * ``client_secret`` — app registration client secret (the only secret)

Per-software bindings carry the operation parameters in ``binding_config``:

    {"operation": "assign_license",
     "sku_id": "00e1ec7b-e4a3-40d1-9441-b69b597ab222",
     "upn_suffix": "pingcapjp.onmicrosoft.com",
     "usage_location": "JP"}

Identity mapping
    Helix stores a person's ``work_email`` (e.g. ``alice@pingcap.com``),
    which is NOT the M365 user-principal-name (``alice@<tenant>
    .onmicrosoft.com``). The two share a **localpart** in this tenant, so
    the driver resolves the Graph user as ``localpart(work_email) + '@' +
    upn_suffix`` when ``upn_suffix`` is configured; otherwise it falls back
    to ``ctx.user_email`` verbatim (tenants where the two coincide).

assignLicense requires the target user to have a ``usageLocation``; the
driver PATCHes the configured ``usage_location`` (default unset → caller
must configure) before assigning when the user has none.

Graph reference
    token:    https://learn.microsoft.com/graph/auth-v2-service
    license:  https://learn.microsoft.com/graph/api/user-assignlicense
    skus:     https://learn.microsoft.com/graph/api/subscribedsku-list
    devices:  https://learn.microsoft.com/graph/api/intune-devices-manageddevice-list
"""
from __future__ import annotations

import asyncio
import logging
import time
from typing import Any
from urllib.parse import quote as _url_quote

import httpx

from app.integrations.driver_base import (
    ProvisioningContext,
    ProvisioningResult,
    SoftwareProvisioningDriver,
)
from app.integrations.registry import register_driver
from app.integrations.schemas import (
    IntegrationFieldInfo,
    IntegrationTestResult,
)

logger = logging.getLogger(__name__)

_LOGIN_BASE = "https://login.microsoftonline.com"
_GRAPH_BASE = "https://graph.microsoft.com/v1.0"
_GRAPH_SCOPE = "https://graph.microsoft.com/.default"
_HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0)

# Transient statuses worth retrying (Graph overload / throttling).
_TRANSIENT_STATUSES: frozenset[int] = frozenset({429, 500, 502, 503, 504})
_RETRY_BACKOFF_SECONDS: tuple[float, ...] = (2.0, 6.0, 18.0)
_MAX_PAGINATOR_PAGES = 200

# Refresh the cached app token this many seconds BEFORE its stated
# expiry so an in-flight request never races the boundary.
_TOKEN_EXPIRY_SKEW = 60.0

# ── Error tags (mirrors google/slack convention) ─────────────────
# Pairs with entries the failure-DM dispatcher maps to remediation
# hints (``_PROVISIONING_REMEDIATION_HINTS``).
MSGRAPH_TAG_AUTH = "MSGRAPH_AUTH_FAILED"
MSGRAPH_TAG_SCOPE = "MSGRAPH_SCOPE_MISSING"
MSGRAPH_TAG_NO_USAGE_LOCATION = "MSGRAPH_NO_USAGE_LOCATION"
MSGRAPH_TAG_NO_SEATS = "MSGRAPH_NO_SEATS_AVAILABLE"
MSGRAPH_TAG_USER_NOT_FOUND = "MSGRAPH_USER_NOT_FOUND"


class MsGraphError(RuntimeError):
    """A non-transient Graph failure the caller surfaces verbatim."""

    def __init__(self, operation: str, status: int | None, detail: str) -> None:
        super().__init__(f"{operation} failed (HTTP {status}): {detail}")
        self.operation = operation
        self.status = status
        self.detail = detail


@register_driver
class MicrosoftGraphDriver(SoftwareProvisioningDriver):
    provider = "microsoft_graph"
    display_name = "Microsoft 365 / Intune"
    description = (
        "Microsoft Graph(client-credentials)。支持分配/回收 M365 许可"
        "(assignLicense)与读取 Intune 托管设备。需要 Entra 应用注册 + "
        "管理员同意的应用权限(User.ReadWrite.All / Organization.Read.All / "
        "DeviceManagementManagedDevices.Read.All 等)。"
    )
    supports_seat = True
    supports_permission = False
    supports_multi_instance = True
    # Removing a license needs no data hand-off (mailbox/OneDrive content
    # is governed by retention policy, not seat ownership), so no successor.
    requires_successor_on_revoke = False

    config_schema = [
        IntegrationFieldInfo(
            key="tenant_id", label="Tenant ID",
            is_secret=False, required=True,
            placeholder="2a13cb65-0b91-4286-a6a7-9ea0038dd3ba",
            hint="Entra 目录(租户)ID。",
        ),
        IntegrationFieldInfo(
            key="client_id", label="Client ID",
            is_secret=False, required=True,
            placeholder="4dc79b2c-4d5d-4dae-8abe-8361fb86bb9a",
            hint="应用注册的应用程序(客户端)ID。",
        ),
        IntegrationFieldInfo(
            key="client_secret", label="Client Secret",
            is_secret=True, required=True,
            placeholder="••••••••",
            hint="应用注册的客户端密码。",
        ),
    ]
    # Per-software binding. Admin picks the SKU + identity-mapping for each
    # software that uses this tenant; the driver dispatches in grant_seat.
    binding_schema = [
        IntegrationFieldInfo(
            key="operation", label="操作类型",
            is_secret=False, required=True, default="assign_license",
            options=[
                {"value": "assign_license", "label": "分配/回收 M365 许可(assignLicense)"},
            ],
            hint="授权 = 给用户分配该 SKU 的许可;回收 = 移除该 SKU。",
        ),
        IntegrationFieldInfo(
            key="sku_id", label="License SKU ID",
            is_secret=False, required=True,
            placeholder="00e1ec7b-e4a3-40d1-9441-b69b597ab222",
            hint="subscribedSkus 里的 skuId(GUID)。",
        ),
        IntegrationFieldInfo(
            key="upn_suffix", label="M365 UPN 后缀",
            is_secret=False, required=False,
            placeholder="pingcapjp.onmicrosoft.com",
            hint=(
                "若 Helix 工作邮箱域与 M365 UPN 域不同,填 M365 侧域名;"
                "驱动按 localpart@该后缀 解析用户。留空则直接用工作邮箱。"
            ),
        ),
        IntegrationFieldInfo(
            key="usage_location", label="默认 Usage Location",
            is_secret=False, required=False,
            placeholder="JP",
            hint="assignLicense 要求用户有 usageLocation;无则按此 ISO 国家码补设。",
        ),
    ]

    def __init__(
        self,
        *,
        integration_id: str,
        config: dict[str, Any],
        enabled_capabilities: list[str] | None = None,
    ) -> None:
        super().__init__(
            integration_id=integration_id,
            config=config,
            enabled_capabilities=enabled_capabilities,
        )
        self._token: str | None = None
        self._token_expiry: float = 0.0

    # ── config accessors ─────────────────────────────────────────
    def _cfg(self, key: str) -> str:
        v = self.config.get(key)
        if not v:
            raise ValueError(f"{key} is not configured")
        return str(v)

    # ── auth ─────────────────────────────────────────────────────
    async def _get_token(self) -> str:
        """Return a cached app token, fetching a fresh one when stale."""
        now = time.monotonic()
        if self._token and now < self._token_expiry - _TOKEN_EXPIRY_SKEW:
            return self._token
        tenant = self._cfg("tenant_id")
        data = {
            "grant_type": "client_credentials",
            "client_id": self._cfg("client_id"),
            "client_secret": self._cfg("client_secret"),
            "scope": _GRAPH_SCOPE,
        }
        async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT, trust_env=False) as c:
            resp = await c.post(
                f"{_LOGIN_BASE}/{_url_quote(tenant)}/oauth2/v2.0/token",
                data=data,
            )
        if resp.status_code != 200:
            raise MsGraphError("token", resp.status_code, resp.text[:300])
        body = resp.json()
        self._token = str(body["access_token"])
        self._token_expiry = now + float(body.get("expires_in", 3600))
        return self._token

    async def _request(
        self,
        method: str,
        path: str,
        *,
        json_body: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
    ) -> httpx.Response:
        """Graph request with bearer auth + bounded transient retry.

        Non-transient responses (incl. 4xx) are returned to the caller to
        interpret (idempotency on 404 etc.). Token-expiry 401 triggers one
        forced token refresh + retry. Raises on exhausted transient retries
        or transport errors.
        """
        url = path if path.startswith("http") else f"{_GRAPH_BASE}{path}"
        attempts = len(_RETRY_BACKOFF_SECONDS) + 1
        refreshed = False
        for attempt in range(attempts):
            token = await self._get_token()
            req_headers = {
                "Authorization": f"Bearer {token}",
                "Content-Type": "application/json",
                # ``ConsistencyLevel: eventual`` is required by the
                # advanced $filter forms (proxyAddresses/any(...)) the
                # UPN-confirmation lookup uses.
                **(headers or {}),
            }
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT, trust_env=False,
            ) as c:
                resp = await c.request(
                    method, url, headers=req_headers,
                    json=json_body, params=params,
                )
            # Expired/invalid token mid-life: refresh once and retry.
            if resp.status_code == 401 and not refreshed:
                self._token = None
                refreshed = True
                continue
            if resp.status_code in _TRANSIENT_STATUSES and attempt < len(
                _RETRY_BACKOFF_SECONDS,
            ):
                backoff = _RETRY_BACKOFF_SECONDS[attempt]
                ra = resp.headers.get("Retry-After")
                if ra:
                    try:
                        backoff = max(backoff, float(ra))
                    except ValueError:
                        pass
                logger.warning(
                    "microsoft_graph %s %s transient HTTP %s "
                    "(attempt %d/%d); backing off %.1fs",
                    method, path, resp.status_code, attempt + 1, attempts,
                    backoff,
                )
                await asyncio.sleep(backoff)
                continue
            return resp
        return resp  # last response after exhausting retries

    # ── identity mapping ─────────────────────────────────────────
    def _resolve_upn(self, ctx: ProvisioningContext) -> str:
        """Map a Helix work_email to the M365 user-principal-name."""
        email = (ctx.user_email or "").strip().lower()
        suffix = str((ctx.extra or {}).get("upn_suffix") or "").strip().lstrip("@")
        if suffix and "@" in email:
            return f"{email.split('@', 1)[0]}@{suffix}"
        return email

    async def _find_upn_by_mail(
        self, ctx: ProvisioningContext,
    ) -> str | None:
        """Resolve the tenant's real UPN for this person, by mail.

        Returns the ``userPrincipalName`` when the tenant knows the
        address, ``""`` when the tenant is reachable and genuinely has
        no such user, and ``None`` when the question could not be
        answered (missing scope / transient error) — the caller must
        treat ``None`` as "unknown", never as "absent" (#2131).

        Searches BOTH ``mail`` and ``proxyAddresses``: a migrated tenant
        routinely carries the Helix work_email as an SMTP alias rather
        than as the primary ``mail``.
        """
        email = (ctx.user_email or "").strip().lower()
        if not email:
            return None
        safe = email.replace("'", "''")
        resp = await self._request(
            "GET", "/users",
            params={
                "$filter": (
                    f"mail eq '{safe}' or "
                    f"proxyAddresses/any(p:p eq 'SMTP:{safe}')"
                ),
                "$select": "id,userPrincipalName,mail",
                "$top": "2",
            },
            headers={"ConsistencyLevel": "eventual"},
        )
        if resp.status_code != 200:
            logger.warning(
                "msgraph: tenant UPN lookup for %s failed with %s — "
                "treating the result as UNKNOWN, not as absent",
                email, resp.status_code,
            )
            return None
        rows = (resp.json() or {}).get("value") or []
        if not rows:
            return ""
        return str(rows[0].get("userPrincipalName") or "") or ""

    # ── health probe ─────────────────────────────────────────────
    async def test_connection(self) -> IntegrationTestResult:
        try:
            resp = await self._request("GET", "/subscribedSkus")
        except Exception as e:  # noqa: BLE001
            return IntegrationTestResult(
                ok=False, message=f"{type(e).__name__}: {e}",
                detail={"type": type(e).__name__},
            )
        if resp.status_code != 200:
            return IntegrationTestResult(
                ok=False,
                message=f"Graph error {resp.status_code}: {resp.text[:200]}",
                detail={"status": resp.status_code},
            )
        skus = resp.json().get("value", [])
        return IntegrationTestResult(
            ok=True,
            message=f"OK · {len(skus)} subscribed SKU(s)",
            detail={"sku_count": len(skus)},
        )

    # ── seat (license) grant / revoke ────────────────────────────
    async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        op = str((ctx.extra or {}).get("operation") or "assign_license")
        if op != "assign_license":
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"Microsoft Graph binding unknown 'operation': {op!r} "
                    "(expected assign_license)"
                ),
            )
        sku_id = str((ctx.extra or {}).get("sku_id") or "")
        if not sku_id:
            return ProvisioningResult(
                ok=False, error_message="binding missing 'sku_id'",
            )
        upn = self._resolve_upn(ctx)

        # 1. Resolve user + current state (idempotency + usageLocation).
        user_resp = await self._request(
            "GET",
            f"/users/{_url_quote(upn)}",
            params={"$select": "id,usageLocation,assignedLicenses"},
        )
        if user_resp.status_code == 404:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"[{MSGRAPH_TAG_USER_NOT_FOUND}] M365 user {upn!r} not "
                    "found. Check the binding's upn_suffix or provision the "
                    "account first."
                ),
                raw_response={"error_tag": MSGRAPH_TAG_USER_NOT_FOUND, "upn": upn},
            )
        if user_resp.status_code == 403:
            return self._scope_error("users.read", user_resp)
        if user_resp.status_code != 200:
            return ProvisioningResult(
                ok=False,
                error_message=f"user lookup {user_resp.status_code}: "
                f"{user_resp.text[:200]}",
            )
        user = user_resp.json()
        already = any(
            (lic or {}).get("skuId") == sku_id
            for lic in user.get("assignedLicenses", [])
        )
        if already:
            return ProvisioningResult(
                ok=True,
                external_ref=f"{sku_id}:{user.get('id')}",
                raw_response={"already_licensed": True, "upn": upn},
            )

        # 2. usageLocation is mandatory for assignLicense.
        if not user.get("usageLocation"):
            loc = str((ctx.extra or {}).get("usage_location") or "").strip()
            if not loc:
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        f"[{MSGRAPH_TAG_NO_USAGE_LOCATION}] {upn} has no "
                        "usageLocation and the binding has no default "
                        "usage_location to set. Configure usage_location "
                        "(ISO country code, e.g. JP)."
                    ),
                    raw_response={"error_tag": MSGRAPH_TAG_NO_USAGE_LOCATION},
                )
            patch = await self._request(
                "PATCH", f"/users/{_url_quote(upn)}",
                json_body={"usageLocation": loc},
            )
            if patch.status_code not in (200, 204):
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        f"failed to set usageLocation={loc} on {upn}: "
                        f"{patch.status_code} {patch.text[:160]}"
                    ),
                )

        # 3. Assign the license. Graph sometimes 400s immediately after a
        #    usageLocation PATCH (replication lag) — retry briefly.
        body = {"addLicenses": [{"skuId": sku_id}], "removeLicenses": []}
        last: httpx.Response | None = None
        for i in range(4):
            resp = await self._request(
                "POST", f"/users/{_url_quote(upn)}/assignLicense",
                json_body=body,
            )
            last = resp
            if resp.status_code in (200, 201, 202):
                return ProvisioningResult(
                    ok=True,
                    external_ref=f"{sku_id}:{user.get('id')}",
                    raw_response={"assigned": True, "upn": upn},
                )
            if resp.status_code == 403:
                return self._scope_error("user.assignLicense", resp)
            # countViolation = no seats left in the pool; a "subscription"
            # error means the SKU has no valid/active subscription to draw
            # from. Both are permanent seat-availability failures — report
            # NO_SEATS and stop retrying (a 400 here is not replication lag).
            _rtext = resp.text
            _low = _rtext.lower()
            # Only treat a "subscription" 400 as a permanent seat-availability
            # failure when the body actually signals unavailability — a BARE
            # "subscription" substring matches unrelated error prose / doc-URL
            # references and would falsely report NO_SEATS (stopping the
            # replication-lag retry and sending admins to buy licenses that
            # were never the problem). ``countViolation`` stays a specific
            # exact token.
            _subscription_unavailable = "subscription" in _low and any(
                k in _low for k in (
                    "expired", "suspended", "disabled", "cancel",
                    "no valid", "not found", "no active", "not active",
                    "inactive", "is not active", "cannot be assigned",
                )
            )
            if "countViolation" in _rtext or _subscription_unavailable:
                _reason = (
                    "no available seats in "
                    f"SKU {sku_id} — buy more licenses or free a seat."
                    if "countViolation" in _rtext
                    else f"no valid/active subscription for SKU {sku_id}."
                )
                return ProvisioningResult(
                    ok=False,
                    error_message=f"[{MSGRAPH_TAG_NO_SEATS}] {_reason}",
                    raw_response={"error_tag": MSGRAPH_TAG_NO_SEATS},
                )
            if resp.status_code == 400 and i < 3:
                await asyncio.sleep(4.0)
                continue
            break
        return ProvisioningResult(
            ok=False,
            error_message=(
                f"assignLicense failed {last.status_code if last else '?'}: "
                f"{last.text[:200] if last else 'no response'}"
            ),
        )

    async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        op = str((ctx.extra or {}).get("operation") or "assign_license")
        if op != "assign_license":
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"Microsoft Graph binding unknown 'operation': {op!r}"
                ),
            )
        sku_id = str((ctx.extra or {}).get("sku_id") or "")
        if not sku_id:
            return ProvisioningResult(
                ok=False, error_message="binding missing 'sku_id'",
            )
        upn = self._resolve_upn(ctx)

        # Idempotency: user gone → access already revoked.
        user_resp = await self._request(
            "GET", f"/users/{_url_quote(upn)}",
            params={"$select": "id,assignedLicenses"},
        )
        if user_resp.status_code == 404:
            # "Revoking a user that no longer exists" IS legitimately a
            # success — but a 404 on the DERIVED UPN cannot tell that
            # apart from "we asked about the wrong person" (#2131).
            # ``_resolve_upn`` is pure string surgery: it glues the
            # work_email's local-part onto the binding's ``upn_suffix``,
            # so a mis-typed suffix or an M365 naming convention that
            # differs from the Helix local-part (the single-surname /
            # double-given-name split, for one) produces a UPN nobody
            # has. Returning ok=True there marks the provisioning task
            # succeeded and clears the offboarding blocker while the
            # leaver still holds the licence.
            #
            # So: before trusting the 404, ask the tenant whether this
            # PERSON exists under any other UPN. Found → retry the
            # revoke against the real one. Genuinely absent → the
            # idempotent success stands. (``grant_seat`` already treats
            # the same 404 as a hard failure with
            # MSGRAPH_USER_NOT_FOUND — this path was the asymmetric one.)
            resolved_upn = await self._find_upn_by_mail(ctx)
            if resolved_upn and resolved_upn.lower() != upn.lower():
                logger.warning(
                    "msgraph revoke: derived UPN %r not found; tenant "
                    "lookup resolved the user to %r — retrying against "
                    "the real UPN", upn, resolved_upn,
                )
                upn = resolved_upn
                user_resp = await self._request(
                    "GET", f"/users/{_url_quote(upn)}",
                    params={"$select": "id,assignedLicenses"},
                )
            if user_resp.status_code == 404:
                if resolved_upn is None:
                    # We could not even run the confirmation search
                    # (scope missing / transient). Do NOT claim success
                    # on an unverified 404 — a false "revoked" is the
                    # one outcome offboarding must never record.
                    return ProvisioningResult(
                        ok=False,
                        error_message=(
                            f"[{MSGRAPH_TAG_USER_NOT_FOUND}] M365 user "
                            f"{upn!r} not found and the tenant lookup "
                            "that would confirm the user is genuinely "
                            "absent could not be completed. Refusing to "
                            "report the licence as revoked — check the "
                            "binding's upn_suffix and the app's "
                            "User.Read.All scope."
                        ),
                        raw_response={
                            "error_tag": MSGRAPH_TAG_USER_NOT_FOUND,
                            "upn": upn, "lookup_inconclusive": True,
                        },
                    )
                return ProvisioningResult(
                    ok=True, external_ref=f"{sku_id}:{upn}",
                    raw_response={
                        "user_gone": True, "upn": upn,
                        "confirmed_absent_by_mail_lookup": True,
                    },
                )
        if user_resp.status_code == 403:
            return self._scope_error("users.read", user_resp)
        if user_resp.status_code != 200:
            return ProvisioningResult(
                ok=False,
                error_message=f"user lookup {user_resp.status_code}: "
                f"{user_resp.text[:200]}",
            )
        user = user_resp.json()
        has = any(
            (lic or {}).get("skuId") == sku_id
            for lic in user.get("assignedLicenses", [])
        )
        if not has:
            return ProvisioningResult(
                ok=True, external_ref=f"{sku_id}:{user.get('id')}",
                raw_response={"already_not_licensed": True, "upn": upn},
            )
        resp = await self._request(
            "POST", f"/users/{_url_quote(upn)}/assignLicense",
            json_body={"addLicenses": [], "removeLicenses": [sku_id]},
        )
        if resp.status_code in (200, 201, 202):
            return ProvisioningResult(
                ok=True, external_ref=f"{sku_id}:{user.get('id')}",
                raw_response={"revoked": True, "upn": upn},
            )
        if resp.status_code == 403:
            return self._scope_error("user.assignLicense", resp)
        # A license INHERITED from a group cannot be removed per-user via
        # assignLicense — Graph returns 400 with an "inherited"/"group"
        # message. This never succeeds on retry (400 isn't transient) and
        # would otherwise dead-letter as a permanent FAILED, pinning the
        # leaver's offboarding blocker. Removing the seat requires taking the
        # user out of the licensing group — a manual action — so downgrade to
        # ``capability_unsupported`` (→ manual_notify) instead of hard-failing.
        _low = resp.text.lower()
        if resp.status_code == 400 and (
            "inherit" in _low or ("group" in _low and "license" in _low)
        ):
            return ProvisioningResult(
                ok=False,
                capability_unsupported=True,
                error_message=(
                    f"license for SKU {sku_id} is inherited from a group and "
                    "cannot be removed per-user; remove the user from the "
                    "licensing group manually."
                ),
            )
        return ProvisioningResult(
            ok=False,
            error_message=f"removeLicense failed {resp.status_code}: "
            f"{resp.text[:200]}",
        )

    def _scope_error(
        self, operation: str, resp: httpx.Response,
    ) -> ProvisioningResult:
        return ProvisioningResult(
            ok=False,
            error_message=(
                f"[{MSGRAPH_TAG_SCOPE}] {operation} refused 403 — the app "
                "registration is missing an admin-consented application "
                f"permission. Detail: {resp.text[:160]}"
            ),
            raw_response={"error_tag": MSGRAPH_TAG_SCOPE, "status": 403},
        )

    # ── read helpers (Intune view / cost reconciliation) ─────────
    async def list_subscribed_skus(self) -> list[dict[str, Any]]:
        """Return license pools (skuId, skuPartNumber, enabled, consumed)."""
        resp = await self._request("GET", "/subscribedSkus")
        if resp.status_code != 200:
            raise MsGraphError("subscribedSkus", resp.status_code, resp.text[:200])
        out: list[dict[str, Any]] = []
        for s in resp.json().get("value", []):
            units = s.get("prepaidUnits", {}) or {}
            out.append({
                "sku_id": s.get("skuId"),
                "sku_part_number": s.get("skuPartNumber"),
                "enabled": units.get("enabled"),
                "consumed": s.get("consumedUnits"),
            })
        return out

    async def list_managed_devices(self) -> list[dict[str, Any]]:
        """Return Intune managed devices (paginated, capped)."""
        out: list[dict[str, Any]] = []
        path: str | None = (
            "/deviceManagement/managedDevices"
            "?$select=id,deviceName,userPrincipalName,operatingSystem,"
            "osVersion,complianceState,managedDeviceOwnerType,lastSyncDateTime"
        )
        pages = 0
        while path and pages < _MAX_PAGINATOR_PAGES:
            resp = await self._request("GET", path)
            if resp.status_code != 200:
                raise MsGraphError(
                    "managedDevices", resp.status_code, resp.text[:200],
                )
            body = resp.json()
            out.extend(body.get("value", []))
            path = body.get("@odata.nextLink")
            pages += 1
        return out