"""Azure cloud driver.

One integration per Azure AD tenant. Credentials are an application
(service principal) client-credential pair that needs:

  * ``Directory.Read.All`` on Microsoft Graph — resolve AAD user object
    id by email, detect orphan accounts before ``setIamPolicy``.
  * ``User Access Administrator`` at the tenant root management group —
    create/remove role assignments on any subscription the SP has access
    to. The revoke-account sweep fans out over every subscription.
  * (optional) ``AuditLog.Read.All`` on Graph + Entra ID P1 — enables
    ``get_last_login`` via ``auditLogs/signIns``. Without this the
    inactivity sweep treats Azure grants as "unknown last-login" and
    skips them (safe default).

The driver is **REST-only via httpx** — we deliberately don't pull in
the ``azure-mgmt-*`` SDKs to keep the backend image lean. Every call
is a straightforward OAuth2-bearer + JSON request, so respx mocks
cover the surface cleanly.

CMDB resource sync (servers / DNS / storage / load balancers) routes
through :class:`app.domains.cmdb.providers.azure.AzureProvider` via
``_CloudDriverBase``. Enabling ``integration.capabilities=['servers',
'dns_records', 'object_storage', 'load_balancers']`` wires up the
corresponding scheduled_jobs rows automatically.

Scope / grant semantics:
  * ``scope_type=azure_subscription``
  * ``external_id`` = subscription GUID
  * ``role_key`` = full ARM path to the role definition
    (``/subscriptions/.../providers/Microsoft.Authorization/roleDefinitions/<guid>``
    for subscription-scoped references, or the tenant path for built-ins).
  * Grants = role assignments on ``/subscriptions/{id}`` scope, stored
    as ``<scope_path>::<assignment_guid>`` in ``external_ref``.
  * ``revoke_account`` sweeps every visible subscription, removes every
    role assignment where ``principalId == <resolved AAD objectId>``.
"""
from __future__ import annotations

import asyncio
import json
import logging
import re
import time
import uuid
from datetime import date, datetime
from typing import Any

import httpx

from app.core.exceptions import format_exception_for_log
from app.domains.cmdb.providers.azure import AzureProvider
from app.domains.cmdb.providers.exceptions import (
    ProviderAuthError,
    redact_provider_error,
)
from app.integrations.driver_base import (  # noqa: F401
    COST_BUCKET_ADJUSTMENT,
    COST_BUCKET_COMMITMENT_FEE,
    COST_BUCKET_CREDIT,
    COST_BUCKET_OTHER,
    COST_BUCKET_SUPPORT,
    COST_BUCKET_TAX,
    COST_BUCKET_USAGE,
    BillingByDimensionRow,
    ClearBlockersResult,
    CloseBlocker,
    ClosePreflight,
    CloseScopeResult,
    CloudAccessContext,
    CloudAccountRevokeContext,
    CloudDailyCost,
    CloudGrantResult,
    CloudPrincipalNotFoundError,
    CloudRoleInfo,
    CloudScopeInfo,
    CloudScopeRef,
    CostFact,
    DiscoveredGrant,
    ProvisioningResult,
    ResourceCreatorInfo,
    ResourceCreatorWindowExceeded,
    ScopeClosureState,
)
from app.integrations.drivers._cloud_defaults import load_default_roles
from app.integrations.drivers._cloud_shared import _CloudDriverBase
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_BETA_BASE = "https://graph.microsoft.com/beta"
_ARM_BASE = "https://management.azure.com"

# Azure Retail Prices API — public, unauthenticated catalogue of
# pay-as-you-go (list / 原价) unit prices. Used by ``list_cost_facts``
# to derive a pre-discount gross amount the Cost Management ``query``
# API cannot return (its ActualCost metrics — Cost / PreTaxCost /
# CostUSD — are all POST-discount / EA-MCA negotiated). The catalogue
# keys on ``meterId`` (a GUID), the SAME meter GUID the Cost Management
# ``MeterId`` dimension returns, so the join is exact. Currency is
# requested in USD so ``retailPrice`` lands directly in the AI
# dashboard's gross-USD axis. ``2023-01-01-preview`` is the api-version
# that honours the ``currencyCode`` query param.
_RETAIL_PRICES_URL = "https://prices.azure.com/api/retail/prices"
_RETAIL_PRICES_API_VERSION = "2023-01-01-preview"
# meterId GUIDs per ``$filter`` OR-clause. ~20 GUIDs ≈ a 1 KB filter,
# comfortably under any URL-length limit; the API caps 1000 records /
# page and we paginate via ``NextPageLink``.
_RETAIL_PRICE_BATCH = 20


# Process-level Microsoft Graph directoryObject cache, keyed by
# ``(integration_id, oid)``. The resource-creator cron walks 200
# rows/tick and a single Service Principal regularly owns hundreds of
# resources (top prod SP ``721a00b8-…`` = 627 disks). Without a cache
# every row burns one ``GET /directoryObjects`` + one
# ``GET /servicePrincipals/{id}/owners`` Graph call → 400 calls / tick
# for ~5-30 distinct SPs.
#
# Values:
#   * ``dict`` — successful resolution, see :meth:`_graph_resolve_oid`.
#   * ``None`` — terminal "Graph can't help" (403 / 404 / not a
#     directoryObject). Cached so we don't re-burn budget on the same
#     dead oid every cron tick.
# TTL: 6 h. SP→owner mappings change slowly (App Registration owners
# are typically static for months); 6h is comfortably within "a single
# cron tick" while still picking up newly-added owners between ticks.
_GRAPH_OID_CACHE: dict[tuple[str, str], tuple[Any, float]] = {}
_GRAPH_OID_CACHE_TTL_S = 6 * 3600.0
_GRAPH_OID_CACHE_MAX = 1024


def _graph_oid_cache_get(
    key: tuple[str, str],
) -> tuple[bool, Any]:
    """Return ``(hit, value)`` — `hit=True` means cached (even if value is None)."""
    entry = _GRAPH_OID_CACHE.get(key)
    if entry is None:
        return False, None
    value, expires_at = entry
    if time.monotonic() >= expires_at:
        _GRAPH_OID_CACHE.pop(key, None)
        return False, None
    return True, value


def _graph_oid_cache_put(
    key: tuple[str, str], value: Any,
) -> None:
    if len(_GRAPH_OID_CACHE) >= _GRAPH_OID_CACHE_MAX:
        # Drop the oldest by expiry. Simple O(n) sweep is fine at this
        # cap; the cache is small and writes are infrequent.
        try:
            oldest_key = min(
                _GRAPH_OID_CACHE.items(), key=lambda kv: kv[1][1],
            )[0]
            _GRAPH_OID_CACHE.pop(oldest_key, None)
        except ValueError:
            pass
    _GRAPH_OID_CACHE[key] = (
        value, time.monotonic() + _GRAPH_OID_CACHE_TTL_S,
    )


_OID_GUID_RE = re.compile(
    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
    r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$",
)


def _safe_body(resp: httpx.Response, *, max_len: int = 400) -> str:
    """Redact and length-cap an Azure HTTP response body for error messages.

    Pre-launch P1 fix (Round 45 follow-up, CR-212): Azure AAD token
    endpoint + ARM/Graph error responses occasionally echo request
    metadata (client_id, signature fragments on retries) and correlation
    tokens. Same family as the Aliyun fix already applied — route the
    raw resp.text through the shared credential scrubber before
    interpolating into exception messages or audit detail.
    """
    return redact_provider_error(resp.text, max_len=max_len)


def _parse_arm_error(resp: httpx.Response) -> tuple[str, str]:
    """Extract ``(code, message)`` from an Azure ARM error response.

    Azure ARM emits errors in two shapes — the canonical nested
    ``{"error": {"code": ..., "message": ...}}`` AND a top-level
    ``{"code": ..., "message": ...}`` form (the user-reported
    ``ResourcesExistingOnSubscription`` 400 on
    ``Microsoft.Subscription/cancel`` arrived in the top-level form,
    likely because the subscription-cancel endpoint predates the
    standardised error envelope). Falling back through both shapes lets
    the driver dispatch on the symbolic code regardless of which API
    family returned the error. Returns ``("", "")`` when the body is
    not JSON or the fields are absent so callers can degrade to the
    redacted-text path without a try/except.
    """
    if not resp.text:
        return "", ""
    try:
        payload = resp.json()
    except ValueError:
        # ``json.JSONDecodeError`` is a subclass of ``ValueError`` — one
        # catch covers both.
        return "", ""
    if not isinstance(payload, dict):
        return "", ""
    nested = payload.get("error")
    if isinstance(nested, dict):
        code = str(nested.get("code") or "").strip()
        message = str(nested.get("message") or "").strip()
        if code or message:
            return code, message
    code = str(payload.get("code") or "").strip()
    message = str(payload.get("message") or "").strip()
    return code, message


def _azure_resource_kind_label(arm_type: str) -> str:
    """Short label for an ARM resource type, e.g. ``storageAccounts``.

    Deliberately the vendor's own last segment rather than a translated
    noun: these labels land in the decommission-confirmation list, which
    renders in zh / en / ja, and Azure's resource types are proper nouns
    that read correctly in all three (the portal shows the same words).
    Inventing a per-locale translation table here would put user-facing
    Chinese in the backend AND leave every unmapped type looking broken.

    Never returns empty for a real type — the label is how a resource is
    identified in a list the admin is consenting to delete, so a missing
    mapping must degrade to something identifying, not to nothing.
    """
    key = (arm_type or "").strip()
    if not key:
        return "resource"
    return key.rsplit("/", 1)[-1] or key


def _parse_iso(value: Any) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None


def _normalize_role_key(role_key: str) -> str:
    """Strip the ``/subscriptions/<guid>`` prefix off Azure ARM role
    definition ids so every stored ``role_key`` uses the tenant-scoped
    form.

    Azure returns role definitions with the full path when listed at
    subscription scope:
        ``/subscriptions/<guid>/providers/Microsoft.Authorization/roleDefinitions/<role-guid>``
    …which is ~138 chars and overflows ``cmdb_cloud_roles.role_key
    VARCHAR(128)`` AND is redundant: the role GUID is globally unique,
    and ``PUT /roleAssignments`` accepts either form as
    ``roleDefinitionId``. The tenant-scoped form:
        ``/providers/Microsoft.Authorization/roleDefinitions/<role-guid>``
    is ~93 chars, matches the driver's ``owner_role_keys``, and is
    what grants-sync stores. Use this helper EVERYWHERE the driver
    emits a role_key so catalog + grants share the same value.
    """
    if role_key.startswith("/subscriptions/"):
        after = role_key.split("/providers/", 1)
        if len(after) == 2:
            return "/providers/" + after[1]
    return role_key


# The only scope_type whose ``external_id`` is a subscription GUID —
# i.e. the only one this driver can put in an ARM URL path.
_SUBSCRIPTION_SCOPE_TYPE = "azure_subscription"


def _require_subscription_scope(scope: CloudScopeRef, what: str) -> None:
    """Refuse a sub-tenant scope row on a subscription-addressed call.

    Every ARM path this driver builds is ``/subscriptions/{external_id}
    /...``, so a scope row whose ``external_id`` is NOT a subscription
    GUID — today only ``azure_resource_group``, one minted per approved
    AI-API-key request by the Azure OpenAI auto-grant flow, carrying an
    RG NAME — makes ARM answer ``400 InvalidSubscriptionId``.

    ``NotImplementedError`` (not a provider error) is deliberate: it
    routes through the base contract's "this driver can't address that
    scope" path, which every caller already handles gracefully — the
    integration-wide sync loops ``continue``, per-scope callers return
    ``skipped``. Recording it as a provider failure instead is what
    turned one AOAI RG row into ``cmdb.grants_sync`` failing on EVERY
    run from 2026-07-29 onward (RG
    ``helix-aoai-20260729-5b25-pingcap-csp``); no retry could clear it,
    because scopes_sync never sweeps auto-grant-minted rows
    (``last_seen_sync_job_id IS NULL``).

    The primary fix is caller-side: the sync loops filter to
    ``TENANT_LEVEL_SCOPE_TYPES``. This is the defence-in-depth net for
    a future caller that forgets.
    """
    if scope.scope_type != _SUBSCRIPTION_SCOPE_TYPE:
        raise NotImplementedError(
            f"Azure {what} is subscription-scoped; cannot query "
            f"scope_type={scope.scope_type!r} "
            f"(external_id={scope.external_id!r})",
        )


# OData $filter values can embed a single quote by doubling it; allow
# nothing else so a crafted email can't inject new clauses.
_EMAIL_SAFE = re.compile(r"[^a-zA-Z0-9._%+\-@]")


# Admin-editable fallback surfaced in the request form when no scope is
# selected yet. Built-in role names are the same across tenants; their
# "role_key" here uses the canonical ``Microsoft.Authorization/roleDefinitions``
# path keyed off the well-known GUID. When the user picks a scope, the
# driver's live ``list_scope_roles`` overrides the UI dropdown.
#
# The curated set deliberately mirrors Azure portal's "Privileged
# administrator roles" tab — these are the 5 roles a normal user would
# ever realistically request access to. Everything else (~380 built-ins)
# remains synced into ``cmdb_cloud_roles`` (``is_promoted=False``) and
# surfaces in the "附加角色" multi-select for power users who know
# exactly which obscure role they need.
_AZURE_FALLBACK_ROLES: list[dict[str, Any]] = [
    {
        "role_key": (
            "/providers/Microsoft.Authorization/roleDefinitions/"
            "8e3af657-a8ff-443c-a75c-2fe8c4bcb635"
        ),
        "display_name": "Owner",
        "description": "Grants full access to manage all resources, "
        "including the ability to assign roles in Azure RBAC.",
    },
    {
        "role_key": (
            "/providers/Microsoft.Authorization/roleDefinitions/"
            "b24988ac-6180-42a0-ab88-20f7382dd24c"
        ),
        "display_name": "Contributor",
        "description": "Grants full access to manage all resources, "
        "but does not allow you to assign roles in Azure RBAC.",
    },
    {
        "role_key": (
            "/providers/Microsoft.Authorization/roleDefinitions/"
            "76cc9ee4-d5d3-4a45-a930-26add3d73475"
        ),
        "display_name": "Access Review Operator Service Role",
        "description": "Lets you grant Access Review System app "
        "permissions to discover and revoke access as needed by the "
        "access review process.",
    },
    {
        "role_key": (
            "/providers/Microsoft.Authorization/roleDefinitions/"
            "f58310d9-a9f6-439a-9e8d-f62e7b41a168"
        ),
        "display_name": "Role Based Access Control Administrator",
        "description": "Manage access to Azure resources by assigning "
        "roles using Azure RBAC. This role does not allow you to manage "
        "access using other ways, such as Azure Policy.",
    },
    {
        "role_key": (
            "/providers/Microsoft.Authorization/roleDefinitions/"
            "18d7d88d-d35e-4fb5-a5c3-7773c20a72d9"
        ),
        "display_name": "User Access Administrator",
        "description": "Lets you manage user access to Azure resources.",
    },
]


# ═══════════════════════════════════════════════════════════════════
# Azure Cost Exports 白名单（2026-07 成本口径改造 PR6）
#
# 云管理团队定稿规则（原文，改语义前须重新核验）：
#   list = payg_cost_in_billing_currency，白名单 charge_type='Usage'
#   （OnDemand/Spot/RI/SP 覆盖行都保留，RI/SP 覆盖行以 PAYG 价还原）
#   + 非承诺类 Purchase（Power BI、Marketplace 等，pricing_model ∉
#   {Reservation, SavingsPlan} 且 reservation_id/benefit_id 为空）；
#   排除 RI/SP 购买行、UnusedReservation/UnusedSavingsPlan、
#   Refund/Credit/RoundingAdjustment/税费、Microsoft Support（单独
#   统计）；禁用列 cost_in_billing_currency/effective_price/unit_price
#   （含合同价）；PAYG 缺失行不得用实际成本兜底，单独告警核查。
#
# 数据面现状：Cost Exports 尚未开通。Exports 数据落在 storage
# account 的 CSV/parquet，接入方式（SAS / RBAC / Data Lake API /
# 定期拉取 vs 事件触发）等云管理开通后才定 —— 本 PR 只落
# 「规则实现 + 配置面 + 数据面适配层」，真实拉取器留
# :meth:`AzureCloudDriver._fetch_exports_rows` 的显式
# NotImplementedError 分支。规则/变换/聚合全部可离线测（注入行）。
#
# 行契约：适配层消费**snake_case 归一化后的行 dict**（见
# ``_normalize_exports_row`` —— 未来拉取器解析 CSV/parquet 表头后
# 先过它再进变换，测试与拉取器共享同一套归一）。

# charge_type='Purchase' 中属于承诺购买的 pricing_model —— RI/SP 购买
# 行进 commitment_fee 桶，绝不入研发口径。
_EXPORTS_COMMITMENT_PRICING_MODELS = frozenset({
    "reservation", "savingsplan",
})

# Exports 表头（PascalCase / camelCase 两代 schema）→ 适配层 snake_case
# 契约键。未来真实拉取器解析出原始表头后必须过这张映射；测试直接用
# snake_case 行。只映射本口径用到的列 —— 其余列透传进 raw。
_EXPORTS_COLUMN_ALIASES: dict[str, str] = {
    "chargetype": "charge_type",
    "pricingmodel": "pricing_model",
    "reservationid": "reservation_id",
    "benefitid": "benefit_id",
    "paygcostinbillingcurrency": "payg_cost_in_billing_currency",
    "costinbillingcurrency": "cost_in_billing_currency",
    "costinusd": "cost_in_usd",
    "billingcurrency": "billing_currency",
    "billingcurrencycode": "billing_currency",
    "date": "date",
    "subscriptionid": "subscription_id",
    "resourceid": "resource_id",
    "resourcegroup": "resource_group",
    "resourcegroupname": "resource_group",
    "metercategory": "meter_category",
    "metersubcategory": "meter_sub_category",
    "resourcelocation": "resource_location",
    "consumedservice": "consumed_service",
    "productname": "product_name",
    "quantity": "quantity",
    "unitofmeasure": "unit_of_measure",
}


def _normalize_exports_row(raw_row: dict[str, Any]) -> dict[str, Any]:
    """Exports 原始表头行 → snake_case 契约行。

    大小写不敏感匹配（``ChargeType`` / ``chargeType`` / ``charge_type``
    都收）；未识别列原样保留（key 原封不动），不丢数据。
    """
    out: dict[str, Any] = {}
    for key, value in raw_row.items():
        canon = _EXPORTS_COLUMN_ALIASES.get(
            key.replace("_", "").strip().lower(),
        )
        out[canon or key] = value
    return out


def _exports_str(row: dict[str, Any], key: str) -> str:
    value = row.get(key)
    return str(value).strip() if value is not None else ""


def _exports_is_support_row(row: dict[str, Any]) -> bool:
    """Microsoft Support 行 —— 云管理规则:单独统计,不入研发口径。"""
    for key in ("consumed_service", "meter_category", "product_name"):
        if _exports_str(row, key).lower().startswith("microsoft support"):
            return True
    return False


def _exports_is_noncommitment_purchase(row: dict[str, Any]) -> bool:
    """非承诺类 Purchase（Power BI / Marketplace 等）判定。

    规则原文：pricing_model ∉ {Reservation, SavingsPlan} **且**
    reservation_id / benefit_id 为空。两个条件都要 —— pricing_model
    字段在老 schema 上可能缺失，reservation/benefit id 是兜底信号。
    """
    pricing_model = _exports_str(row, "pricing_model").lower()
    if pricing_model in _EXPORTS_COMMITMENT_PRICING_MODELS:
        return False
    if _exports_str(row, "reservation_id") or _exports_str(row, "benefit_id"):
        return False
    return True


def _classify_exports_bucket(row: dict[str, Any]) -> str:
    """Exports 行 → 财务瀑布九桶（闭集）。

    白名单思路：未知 charge_type 落 OTHER，绝不误入 usage。
    """
    if _exports_is_support_row(row):
        return COST_BUCKET_SUPPORT
    charge_type = _exports_str(row, "charge_type").lower()
    if charge_type == "usage":
        # OnDemand / Spot / RI / SP 覆盖行都在这里（pricing_model 区分
        # 但桶一致）—— 全部保留在研发口径，RI/SP 覆盖行以 PAYG 价还原。
        return COST_BUCKET_USAGE
    if charge_type == "purchase":
        if _exports_is_noncommitment_purchase(row):
            # Power BI / Marketplace 等按用付费型购买 —— 研发口径保留。
            return COST_BUCKET_USAGE
        return COST_BUCKET_COMMITMENT_FEE  # RI/SP 购买行
    if charge_type in ("unusedreservation", "unusedsavingsplan"):
        return COST_BUCKET_COMMITMENT_FEE
    if charge_type in ("refund", "roundingadjustment"):
        return COST_BUCKET_ADJUSTMENT
    if charge_type == "credit":
        return COST_BUCKET_CREDIT
    if charge_type == "tax":
        return COST_BUCKET_TAX
    return COST_BUCKET_OTHER


def _exports_in_list_whitelist(row: dict[str, Any]) -> bool:
    """行是否进研发口径（list 白名单）。

    == 「分类为 usage 桶」：charge_type='Usage' 全量 + 非承诺类
    Purchase；Support / 承诺购买 / Unused* / Refund / Credit /
    RoundingAdjustment / Tax / 未知类型全部出局。
    """
    return _classify_exports_bucket(row) == COST_BUCKET_USAGE


def _exports_payg_amount(row: dict[str, Any]) -> float | None:
    """白名单行的 PAYG 牌价（billing currency）。

    云管理规则：PAYG 缺失行**不得**用实际成本
    （cost_in_billing_currency —— 禁用列，含合同价）兜底 —— 返回
    None，调用方聚合计数后 WARN 告警核查。坏值（非数字）同样 None。
    """
    value = row.get("payg_cost_in_billing_currency")
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


@register_driver
class AzureCloudDriver(_CloudDriverBase):
    provider = "azure"
    display_name = "Microsoft Azure"
    description = (
        "Azure 租户 (AAD) 下多订阅管理。依赖一个拥有 "
        "User Access Administrator + Billing Reader + "
        "Directory.Read.All 的 service principal。可选 AuditLog.Read.All + "
        "Entra ID P1 开启登录时间追踪。"
    )
    supported_capabilities = frozenset({
        "servers", "dns_records", "object_storage", "load_balancers",
        "user_revoke",
    })
    capabilities = [
        "servers", "dns_records", "object_storage", "load_balancers",
        "user_revoke",
    ]
    account_mode = "sso"

    # Azure "Owner" built-in role — stable GUID across every tenant.
    # Used by the grants-sync sweep to auto-adopt ``CmdbCloudScope``
    # owners when exactly one person holds this role on a previously
    # unowned scope.
    owner_role_keys = frozenset({
        "/providers/Microsoft.Authorization/roleDefinitions/"
        "8e3af657-a8ff-443c-a75c-2fe8c4bcb635",
    })

    # Role governance (see app/integrations/cloud_role_governance.py).
    # Owner (above) is owner-equivalent. User Access Administrator + Role
    # Based Access Control Administrator grant roleAssignments/write
    # (escalation) → restricted. Contributor = full resource mgmt without
    # role assignment → elevated. Role keys are tenant-scoped paths.
    _RD = "/providers/Microsoft.Authorization/roleDefinitions/"
    restricted_role_keys = frozenset({
        _RD + "18d7d88d-d35e-4fb5-a5c3-7773c20a72d9",  # User Access Administrator
        _RD + "f58310d9-a9f6-439a-9e8d-f62e7b41a168",  # Role Based Access Control Administrator
    })
    elevated_role_keys = frozenset({
        _RD + "b24988ac-6180-42a0-ab88-20f7382dd24c",  # Contributor
    })

    # Bridge to the CMDB CloudProvider (used by _CloudDriverBase for
    # list_servers / list_dns_records; list_object_storage /
    # list_load_balancers are overridden below because _CloudDriverBase
    # only bridges the three GA methods).
    provider_cls = AzureProvider

    config_schema = [
        IntegrationFieldInfo(
            key="tenant_id", label="Tenant ID (AAD)",
            is_secret=False, required=True,
            placeholder="xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
        ),
        IntegrationFieldInfo(
            key="client_id", label="Application (client) ID",
            is_secret=False, required=True,
        ),
        IntegrationFieldInfo(
            key="client_secret", label="Client Secret",
            is_secret=True, required=True,
        ),
        IntegrationFieldInfo(
            key="resource_inventory_enabled", label="启用全资源目录",
            is_secret=False, required=False, placeholder="false", default="false",
        ),
        # 成本查询客户端节流(默认关)。开着时驱动会把一个订阅学到的
        # 限流冷却值带到下一个订阅之前主动等待,而不是直接撞 429 再等。
        # 默认关的理由不是"没实现",是**实测**:2026-09-05 发布后的验证
        # 跑撞上了 1800s 任务超时(36 个订阅只跑到 27 个),而同一 job 的
        # 基线是 587-775s;那一跑里节流预算一次都没耗尽(⇒ 主动等待
        # < 600s),真正的大头是 34 次常规 429 退避。详见
        # ``AzureCloudDriver.__init__`` 里的取证。要评估它,请只在**一个**
        # Azure 集成上打开,拿一整周的 run 时长 + 429 计数和基线对比。
        IntegrationFieldInfo(
            key="cost_pacing_enabled", label="成本查询客户端节流(实验)",
            is_secret=False, required=False, placeholder="false", default="false",
        ),
        # ── Billing source（成本口径改造 PR6）──────────────────────
        # 默认 "cost_management":现状路径,Cost Management Query API
        # ActualCost(EA/MCA 折后价),研发口径 list 恒 NULL(读侧
        # COALESCE 回落实付,口径说明页已标注)。
        # "exports":Cost Exports 白名单路径(payg_cost_in_billing_
        # currency = 研发口径)。规则/变换已实现;真实拉取器等云管理
        # 开通导出后接(_fetch_exports_rows 现为 NotImplementedError,
        # 开了这个开关而拉取器未接时 cron 会如实记 per-scope error,
        # 不产出静默空数据)——开通前请保持默认值。
        IntegrationFieldInfo(
            key="billing_source", label="账单数据源",
            is_secret=False, required=False,
            placeholder="cost_management",
            default="cost_management",
        ),
        # Exports 落点(storage account)。仅 billing_source="exports"
        # 时相关;具体读取方式(SAS vs RBAC、目录布局、CSV/parquet)
        # 待云管理开通导出后确定,这三个字段是为其预留的配置面。
        IntegrationFieldInfo(
            key="exports_storage_account", label="Exports Storage Account",
            is_secret=False, required=False,
            placeholder="stcostexports",
            visible_when={"billing_source": "exports"},
        ),
        IntegrationFieldInfo(
            key="exports_container", label="Exports Container",
            is_secret=False, required=False,
            placeholder="cost-exports",
            visible_when={"billing_source": "exports"},
        ),
        IntegrationFieldInfo(
            key="exports_path_prefix", label="Exports Path Prefix",
            is_secret=False, required=False,
            placeholder="daily/actual-cost",
            visible_when={"billing_source": "exports"},
        ),
    ]

    # ── CMDB list passthrough for methods _CloudDriverBase skips ─

    async def list_object_storage(self) -> list[dict[str, Any]]:
        provider = self._provider()
        try:
            return await provider.list_object_storage()
        finally:
            await provider.close()

    async def list_load_balancers(self) -> list[dict[str, Any]]:
        import dataclasses as _dc
        provider = self._provider()
        try:
            return [_dc.asdict(lb) for lb in await provider.list_load_balancers()]
        finally:
            await provider.close()

    # ── HTTP + token ─────────────────────────────────────────────

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        # Per-resource token cache: resource_url -> (token, expiry_monotonic)
        self._token_cache: dict[str, tuple[str, float]] = {}
        # Shared across every Cost Management call this driver instance
        # makes. ``cost_service`` reuses ONE driver for all scopes in a
        # run, so this is what carries "the server just told us to wait
        # 45s" from subscription N to subscription N+1. See the
        # ``_AzureThrottlePacer`` header for the prod trace this comes
        # from.
        #
        # 🔴 PROACTIVE PACING IS OPT-IN AND DEFAULTS **OFF** — measured,
        # not assumed. The post-deploy verification run on 2026-09-05
        # (manual tick, 04:20 UTC) **hit the job's 1800s timeout after
        # only 27 of 36 subscriptions**, against a 587-775s baseline.
        # Forensics from that run:
        #   * 88 CM queries, 34 × 429, 0 give-ups
        #   * ``pacing budget spent`` never logged ⇒ proactive pacing
        #     added < 600s; the dominant cost was 34 × ~36s ≈ 1224s of
        #     ordinary reactive 429 backoff
        #   * server-stated cooldowns flat at 34-38s throughout, so the
        #     limiter was not progressively tightening
        # i.e. under that (heavily loaded) limiter, pacing did NOT stop
        # the 429s — it only spent headroom on a job that has now
        # demonstrably reached its timeout. The limiter state was partly
        # self-inflicted (an earlier 11-minute attempt in the same hour),
        # so this is not proof that pacing is harmful — but it is proof
        # that the "pacing is wall-clock-neutral" model can be wrong,
        # and this cron cannot afford to find that out nightly.
        #
        # What ships ON by default is the part the reported incident
        # actually needed and that cannot cost wall clock: a header-less
        # 429 reuses the last SERVER-STATED cooldown instead of
        # restarting the exponential at 1s (``observe_throttle`` still
        # learns with pacing off, so ``suggested_backoff_s`` works),
        # plus the cumulative retry budget and the throttle diagnostics.
        #
        # To evaluate pacing properly, set ``cost_pacing_enabled: true``
        # on ONE Azure integration and compare a full week of that job's
        # run durations + ``grep -c 'Azure HTTP 429'`` against the
        # baseline — on an otherwise-quiet limiter, not minutes after a
        # manual run.
        self._cm_pacer = _AzureThrottlePacer(
            pacing_enabled=str(
                self.config.get("cost_pacing_enabled") or "",
            ).strip().lower() in ("1", "true", "yes", "on"),
            min_backoff_s=_CM_MIN_THROTTLE_BACKOFF_SEC,
        )

    def _http(self) -> httpx.AsyncClient:
        return httpx.AsyncClient(timeout=30, trust_env=False)

    async def _token(self, resource: str) -> str:
        """Fetch an OAuth2 bearer for the given resource base URL.

        ``resource`` is one of the ARM or Graph base URLs — ``/.default``
        is appended to tell AAD "every scope granted to this SP".
        Tokens are cached per-resource for the lifetime of the driver
        instance and refreshed 60 seconds before expiry.
        """
        # Check cache first
        cached = self._token_cache.get(resource)
        if cached is not None:
            token, expiry = cached
            if time.monotonic() < expiry - 60:
                return token

        tenant = str(self.config.get("tenant_id") or "").strip()
        client_id = str(self.config.get("client_id") or "").strip()
        client_secret = str(self.config.get("client_secret") or "").strip()
        if not tenant or not client_id or not client_secret:
            raise RuntimeError(
                "Azure driver missing tenant_id/client_id/client_secret",
            )
        url = f"{_LOGIN_BASE}/{tenant}/oauth2/v2.0/token"
        data = {
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
            "scope": f"{resource}/.default",
        }
        # Transport + 429/5xx retry: this single POST is the prerequisite
        # for every other call in the driver — a transport-level flake
        # against login.microsoftonline.com cascades to every cost /
        # resource / billing sync in the same run. AAD doesn't expose
        # vendor-specific Retry-After headers beyond the canonical
        # ``Retry-After`` (which ``_post_with_retry`` already honours),
        # so no parse_retry_after callback needed.
        async with self._http() as client:
            resp = await _post_with_retry(
                client, url,
                data=data,
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                endpoint_label=f"_token resource={resource}",
            )
        if resp.status_code != 200:
            raise RuntimeError(
                f"Azure token endpoint {resp.status_code}: {_safe_body(resp, max_len=300)}",
            )
        payload = resp.json()
        token = str(payload.get("access_token") or "")
        expires_in = int(payload.get("expires_in") or 3600)
        self._token_cache[resource] = (token, time.monotonic() + expires_in)
        return token

    # ── Health probe (ARM + Graph) ───────────────────────────────

    async def test_connection(self) -> IntegrationTestResult:
        """Two-step probe: ARM subscription list + Graph directory read.

        Fails fast if either permission is missing — cheaper for the
        admin to diagnose now than on the first offboarding sweep.
        """
        try:
            arm_token = await self._token(_ARM_BASE)
            async with self._http() as client:
                resp = await client.get(
                    f"{_ARM_BASE}/subscriptions",
                    params={"api-version": "2022-12-01"},
                    headers={"Authorization": f"Bearer {arm_token}"},
                )
            if resp.status_code != 200:
                return IntegrationTestResult(
                    ok=False,
                    message=f"list subscriptions {resp.status_code}: "
                    f"{_safe_body(resp, max_len=200)}",
                )
            subs = resp.json().get("value") or []
            msg = f"OK · {len(subs)} subscription(s) visible"
        except Exception as e:  # noqa: BLE001
            return IntegrationTestResult(ok=False, message=str(e))
        # Graph probe — warn but don't fail if it misses; IT might be
        # onboarding an ARM-only SP and adding Graph later.
        try:
            graph_token = await self._token("https://graph.microsoft.com")
            async with self._http() as client:
                g = await client.get(
                    f"{_GRAPH_BASE}/users",
                    params={"$top": "1", "$select": "id"},
                    headers={"Authorization": f"Bearer {graph_token}"},
                )
            if g.status_code == 200:
                msg += " · Graph OK"
            else:
                msg += (
                    f" · Graph 访问失败 ({g.status_code}): "
                    "授予 Directory.Read.All 后再试"
                )
                return IntegrationTestResult(ok=False, message=msg)
        except Exception as exc:  # noqa: BLE001
            msg += f" · Graph 访问失败: {exc}"
            return IntegrationTestResult(ok=False, message=msg)
        return IntegrationTestResult(ok=True, message=msg)

    # ── Scope + role catalog ─────────────────────────────────────

    async def list_scopes(self) -> list[CloudScopeInfo]:
        MAX_PAGES = 200
        token = await self._token(_ARM_BASE)
        out: list[CloudScopeInfo] = []
        next_url: str | None = f"{_ARM_BASE}/subscriptions?api-version=2022-12-01"
        pages = 0
        async with self._http() as client:
            while next_url:
                if pages >= MAX_PAGES:
                    logger.warning(
                        "Azure list_scopes: hit MAX_PAGES=%d cap; "
                        "stopping pagination (collected %d subscriptions so far)",
                        MAX_PAGES, len(out),
                    )
                    break
                resp = await client.get(
                    next_url, headers={"Authorization": f"Bearer {token}"},
                )
                if resp.status_code != 200:
                    raise RuntimeError(
                        f"list_subscriptions {resp.status_code}: "
                        f"{_safe_body(resp, max_len=300)}",
                    )
                payload = resp.json()
                for sub in payload.get("value") or []:
                    state = str(sub.get("state", "Enabled"))
                    out.append(CloudScopeInfo(
                        scope_type="azure_subscription",
                        external_id=str(sub["subscriptionId"]),
                        display_name=sub.get("displayName"),
                        status="active" if state.lower() == "enabled"
                        else state.lower(),
                        raw={
                            "tenant_id": sub.get("tenantId"),
                            "state": state,
                            "authorization_source": sub.get(
                                "authorizationSource",
                            ),
                        },
                    ))
                next_url = payload.get("nextLink")
                pages += 1
        return out

    async def list_scope_roles(
        self, scope: CloudScopeRef,
    ) -> list[CloudRoleInfo]:
        """Curated builtins (promoted) + every customer-defined role
        (promoted) + every other vendor builtin (NOT promoted).

        Pre-068, this method returned all ~380 vendor built-in RBAC
        roles indiscriminately — the request modal's role dropdown
        showed dozens of "API Management Service Workspace API
        Developer"-style entries that an employee would never know how
        to choose between. Now:

        * Roles in ``cloud.azure.default_roles`` (admin-curated) →
          ``is_promoted=True``, surfaced by default in the dropdown.
        * Custom roles defined in the customer's tenant → also
          ``is_promoted=True`` — by definition someone in the org
          intentionally created them, so showing them is right.
        * Every other vendor built-in → ``is_promoted=False``. They're
          still synced into ``cmdb_cloud_roles`` so the admin-side
          "promote/demote" catalog UI can surface them, and so the
          "include all built-ins" toggle on the request modal can
          fall back to the full set if the curated subset is too
          narrow for a particular request.
        """
        # ``roleDefinitions`` is listed at subscription scope — an RG
        # row's name in the URL path is a 400. See
        # ``_require_subscription_scope``.
        _require_subscription_scope(scope, "role catalog")
        defaults = await load_default_roles(
            "cloud.azure.default_roles", _AZURE_FALLBACK_ROLES,
        )
        promoted_keys: set[str] = set()
        out: list[CloudRoleInfo] = []
        for entry in defaults:
            key = str(entry.get("role_key") or "").strip()
            if not key:
                continue
            promoted_keys.add(key)
            out.append(CloudRoleInfo(
                role_key=key,
                display_name=str(entry.get("display_name") or key),
                description=entry.get("description"),
                is_custom=False,
                source="azure_predefined",
                scope=scope,
                is_promoted=True,
            ))
        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{scope.external_id}/providers/"
            f"Microsoft.Authorization/roleDefinitions"
        )
        seen = {r.role_key for r in out}
        async with self._http() as client:
            next_url: str | None = (
                f"{url}?api-version=2022-04-01"
            )
            while next_url:
                resp = await client.get(
                    next_url, headers={"Authorization": f"Bearer {token}"},
                )
                if resp.status_code != 200:
                    raise RuntimeError(
                        f"list_role_definitions {resp.status_code}: "
                        f"{_safe_body(resp, max_len=300)}",
                    )
                payload = resp.json()
                for rd in payload.get("value") or []:
                    # Normalize subscription-prefixed role ids to the
                    # tenant-scoped form so the role catalog stores the
                    # same key shape grants-sync persists — otherwise
                    # the ``owner_role_keys`` match and the request-
                    # form role dropdown fall out of sync.
                    key = _normalize_role_key(str(rd["id"]))
                    if key in seen:
                        continue
                    seen.add(key)
                    props = rd.get("properties") or {}
                    role_name = props.get("roleName") or rd.get("name")
                    kind = props.get("type") or ""
                    is_custom = kind == "CustomRole"
                    out.append(CloudRoleInfo(
                        role_key=key,  # tenant-scoped — accepted by ARM
                        display_name=role_name,
                        description=props.get("description"),
                        is_custom=is_custom,
                        source=(
                            "azure_builtin"
                            if kind == "BuiltInRole"
                            else "azure_custom"
                        ),
                        scope=scope,
                        raw={"name": rd.get("name"), "type": kind},
                        is_promoted=is_custom or key in promoted_keys,
                    ))
                next_url = payload.get("nextLink")
        return out

    # ── Principal resolution (AAD Graph) ─────────────────────────

    def _sanitize_email_for_odata(self, email: str) -> str:
        """OData filters reject anything but a narrow ASCII set for emails.

        We assert upfront — a crafted email like ``a' or 1 eq 1--`` would
        otherwise extend the filter. Since upstream validation already
        restricts emails to RFC-ish chars, this is belt-and-braces.
        """
        if _EMAIL_SAFE.search(email):
            raise CloudPrincipalNotFoundError(
                f"Azure principal lookup: email {email!r} contains characters "
                "not allowed in OData filters",
            )
        return email

    async def _resolve_user_object_id(self, email: str) -> str:
        safe = self._sanitize_email_for_odata(email)
        token = await self._token("https://graph.microsoft.com")
        filt = (
            f"mail eq '{safe}' or userPrincipalName eq '{safe}' "
            f"or otherMails/any(x:x eq '{safe}')"
        )
        async with self._http() as client:
            resp = await client.get(
                f"{_GRAPH_BASE}/users",
                params={
                    "$filter": filt,
                    "$select": "id,displayName,mail,userPrincipalName,userType",
                },
                headers={"Authorization": f"Bearer {token}"},
            )
        if resp.status_code == 403:
            raise RuntimeError(
                "Azure Graph /users 403 — grant Directory.Read.All to the "
                "service principal",
            )
        if resp.status_code != 200:
            raise RuntimeError(
                f"graph users {resp.status_code}: {_safe_body(resp, max_len=300)}",
            )
        users = resp.json().get("value") or []
        if not users:
            raise CloudPrincipalNotFoundError(
                f"Azure AD has no user with mail/UPN {email!r}. "
                "Sync the user from your IdP first.",
            )
        # Disambiguate when the OData filter returns multiple matches —
        # an attacker-relevant case in tenants that invite B2B guests
        # (member + guest invitee + a service principal masquerading as
        # a user can all share the email/otherMails). Picking
        # ``users[0]`` arbitrarily would land the role assignment on
        # the wrong AAD object — privilege grant to wrong principal.
        #
        # Resolution order:
        #   1. Member (non-guest) with exact UPN match. The UPN is the
        #      authoritative tenant-scoped identifier.
        #   2. Member with mail match.
        #   3. Single match (guest or member) — accept it.
        #   4. Ambiguous — raise, never silently pick.
        norm = email.lower()
        members_upn = [
            u for u in users
            if (u.get("userType") or "Member") == "Member"
            and (u.get("userPrincipalName") or "").lower() == norm
        ]
        if len(members_upn) == 1:
            return str(members_upn[0]["id"])
        members_mail = [
            u for u in users
            if (u.get("userType") or "Member") == "Member"
            and (u.get("mail") or "").lower() == norm
        ]
        if len(members_mail) == 1:
            return str(members_mail[0]["id"])
        if len(users) == 1:
            return str(users[0]["id"])
        raise CloudPrincipalNotFoundError(
            f"Azure AD lookup for {email!r} is ambiguous: {len(users)} "
            "objects match (member + guest + other). Use the exact "
            "userPrincipalName so the grant lands deterministically.",
        )

    # ── Grant / revoke ───────────────────────────────────────────

    async def grant_access(
        self, ctx: CloudAccessContext,
    ) -> CloudGrantResult:
        # RBAC assignments are bound at ``/subscriptions/{guid}`` below.
        # A sub-tenant scope row would ARM-400 AFTER the approver said
        # yes, so refuse with a legible message instead (the request
        # flow blocks these up front — see
        # ``_refuse_ungrantable_scope_type`` in lifecycle_service).
        if ctx.scope.scope_type != _SUBSCRIPTION_SCOPE_TYPE:
            return CloudGrantResult(
                ok=False,
                error_message=(
                    f"Azure role assignments are subscription-scoped; "
                    f"cannot grant on scope_type="
                    f"{ctx.scope.scope_type!r} "
                    f"(external_id={ctx.scope.external_id!r})"
                ),
            )
        try:
            principal_id = await self._resolve_user_object_id(ctx.principal_email)
        except CloudPrincipalNotFoundError as e:
            return CloudGrantResult(ok=False, error_message=str(e))
        token = await self._token(_ARM_BASE)
        assignment_id = str(uuid.uuid4())
        scope_path = f"/subscriptions/{ctx.scope.external_id}"
        url = (
            f"{_ARM_BASE}{scope_path}/providers/Microsoft.Authorization/"
            f"roleAssignments/{assignment_id}?api-version=2022-04-01"
        )
        body = {
            "properties": {
                "roleDefinitionId": ctx.role_key,
                "principalId": principal_id,
                "principalType": "User",
            }
        }
        async with self._http() as client:
            resp = await client.put(
                url, json=body,
                headers={"Authorization": f"Bearer {token}"},
            )
        if resp.status_code in (200, 201):
            return CloudGrantResult(
                ok=True,
                external_ref=f"{scope_path}::{assignment_id}",
                detail={
                    "principal_id": principal_id,
                    "assignment_id": assignment_id,
                    "scope": scope_path,
                },
            )
        # 409 with RoleAssignmentExists → find the pre-existing one so
        # revoke later targets the right assignment guid. Gate on BOTH the
        # 409 status AND the ARM error code — a 400/403 whose body merely
        # mentions "RoleAssignmentExists" must not be misread as idempotent
        # success (#1649).
        if resp.status_code == 409 and "RoleAssignmentExists" in resp.text:
            matching = await self._find_existing_assignment(
                token, scope_path, principal_id, ctx.role_key,
            )
            if matching is None:
                # Azure says the assignment already exists but we can't
                # locate its real GUID — emitting our locally-generated
                # ``assignment_id`` would persist a PHANTOM external_ref that
                # a later revoke can never resolve, silently orphaning the
                # binding. Fail loudly so the caller retries / reconciles
                # instead of recording a false success.
                return CloudGrantResult(
                    ok=False,
                    error_message=(
                        "Azure reported RoleAssignmentExists but the existing "
                        "assignment GUID could not be resolved; refusing to "
                        "record a phantom external_ref. Retry after the ARM "
                        "read replica catches up."
                    ),
                    detail={"principal_id": principal_id},
                )
            return CloudGrantResult(
                ok=True, already_exists=True,
                external_ref=f"{scope_path}::{matching}",
                detail={"principal_id": principal_id},
            )
        # "PrincipalNotFound" from ARM when the AAD object was deleted
        # between Graph lookup and PUT — surface as our canonical error.
        if (
            resp.status_code == 400
            and "PrincipalNotFound" in resp.text
        ):
            raise CloudPrincipalNotFoundError(
                f"Azure ARM rejected principal {principal_id!r} for "
                f"{ctx.principal_email}: {_safe_body(resp, max_len=200)}",
            )
        return CloudGrantResult(
            ok=False,
            error_message=f"role assignment {resp.status_code}: "
            f"{_safe_body(resp, max_len=300)}",
        )

    async def _find_existing_assignment(
        self, token: str, scope_path: str,
        principal_id: str, role_key: str,
    ) -> str | None:
        # Paginate: Azure returns role assignments across multiple pages
        # when a subscription has >~20 assignments. Stopping at page 1
        # would cause revoke_access to generate a fresh UUID external_ref
        # on subsequent 409s, leaving the real assignment untracked and
        # surviving leaver offboarding.
        _MAX_PAGES = 50
        next_url: str | None = (
            f"{_ARM_BASE}{scope_path}/providers/Microsoft.Authorization/"
            "roleAssignments?api-version=2022-04-01"
            f"&$filter=principalId eq '{principal_id}'"
        )
        pages = 0
        async with self._http() as client:
            while next_url and pages < _MAX_PAGES:
                list_resp = await client.get(
                    next_url,
                    headers={"Authorization": f"Bearer {token}"},
                )
                pages += 1
                if list_resp.status_code != 200:
                    return None
                payload = list_resp.json()
                for ra in payload.get("value") or []:
                    rprops = ra.get("properties") or {}
                    if rprops.get("roleDefinitionId") == role_key:
                        return str(ra.get("name") or "") or None
                next_url = payload.get("nextLink")
        return None

    async def revoke_access(
        self, ctx: CloudAccessContext,
    ) -> CloudGrantResult:
        if not ctx.existing_external_ref or "::" not in ctx.existing_external_ref:
            return CloudGrantResult(
                ok=False,
                error_message=(
                    "Azure revoke requires stored assignment id (external_ref)"
                ),
            )
        scope_path, assignment_id = ctx.existing_external_ref.split("::", 1)
        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}{scope_path}/providers/Microsoft.Authorization/"
            f"roleAssignments/{assignment_id}?api-version=2022-04-01"
        )
        async with self._http() as client:
            resp = await client.delete(
                url, headers={"Authorization": f"Bearer {token}"},
            )
        if resp.status_code in (200, 204):
            return CloudGrantResult(
                ok=True, external_ref=ctx.existing_external_ref,
            )
        if resp.status_code == 404:
            return CloudGrantResult(ok=True, already_absent=True)
        return CloudGrantResult(
            ok=False,
            error_message=(
                f"delete role assignment {resp.status_code}: "
                f"{_safe_body(resp, max_len=300)}"
            ),
        )

    # ── Account revocation (full subscription sweep) ─────────────

    async def revoke_account(
        self, ctx: CloudAccountRevokeContext,
    ) -> ProvisioningResult:
        """Remove every role assignment tied to the user across every
        subscription the SP can see.

        ``account_id`` must be the user's email — we resolve it to an
        AAD objectId, then for each subscription list assignments by
        ``principalId`` and delete them. Each DELETE is idempotent
        (404 = already gone = success).

        Assignments at the management-group or root level are NOT swept
        — the SP typically can't see those, and most customer-facing
        grants are subscription-scoped. Any leftover is surfaced in
        ``raw_response.subscriptions_failed`` for admin follow-up.
        """
        email = (ctx.account_id or "").strip()
        if not email or "@" not in email:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"Azure revoke_account requires an email as account_id; "
                    f"got {ctx.account_id!r}"
                ),
            )
        try:
            principal_id = await self._resolve_user_object_id(email)
        except CloudPrincipalNotFoundError:
            # No AAD user = no assignments to revoke — it's a success
            # from the lifecycle point of view. We still return the
            # sweep metadata so audit can reason about it.
            return ProvisioningResult(
                ok=True, external_ref=email,
                raw_response={
                    "principal_id": None,
                    "subscriptions_scanned": 0,
                    "reason": "principal not in AAD; nothing to revoke",
                },
            )
        scopes = await self.list_scopes()
        token = await self._token(_ARM_BASE)
        scanned = len(scopes)
        modified: list[str] = []
        failed: list[tuple[str, str]] = []
        async with self._http() as client:
            for scope in scopes:
                scope_path = f"/subscriptions/{scope.external_id}"
                try:
                    assignments = await self._list_principal_assignments(
                        client, token, scope_path, principal_id,
                    )
                except Exception as exc:  # noqa: BLE001
                    logger.warning(
                        "Azure revoke_account list failed on sub %s: %s",
                        scope.external_id, exc,
                    )
                    failed.append((scope.external_id, str(exc)))
                    continue
                if not assignments:
                    continue
                # Leavers must be ALL revoked. Previously the loop
                # marked the subscription ``modified`` whenever ANY of
                # N assignments returned 200/204/404 — partial revoke
                # left residual access on the ones that 5xx'd. Track
                # success-vs-failure per assignment and only count the
                # subscription as "modified" if every assignment was
                # cleanly handled; otherwise record it as failed with
                # the residual count so the lifecycle service surfaces
                # the gap.
                deleted_count = 0
                sub_failed = False
                for ra_name in assignments:
                    del_url = (
                        f"{_ARM_BASE}{scope_path}/providers/"
                        f"Microsoft.Authorization/roleAssignments/{ra_name}"
                        "?api-version=2022-04-01"
                    )
                    resp = await client.delete(
                        del_url,
                        headers={"Authorization": f"Bearer {token}"},
                    )
                    if resp.status_code in (200, 204, 404):
                        deleted_count += 1
                    else:
                        sub_failed = True
                        failed.append((scope.external_id, (
                            f"{resp.status_code}: {_safe_body(resp, max_len=200)}"
                        )))
                if sub_failed:
                    # Don't double-count: the per-assignment failures
                    # are already in ``failed`` from the loop above. Do
                    # NOT append a second summary row — len(failed)
                    # would otherwise overstate the subscription count
                    # and mislead ops triage. The per-assignment rows
                    # carry the diagnostic; the absence from
                    # ``modified`` carries the "subscription not
                    # cleanly revoked" signal.
                    pass
                else:
                    modified.append(scope.external_id)
        detail = {
            "principal_id": principal_id,
            "subscriptions_scanned": scanned,
            "subscriptions_modified": modified,
            "subscriptions_failed": failed,
        }
        if failed:
            return ProvisioningResult(
                ok=False,
                external_ref=email,
                error_message=(
                    f"Azure revoke_account touched {len(modified)}/{scanned} "
                    f"subscription(s); {len(failed)} failed"
                ),
                raw_response=detail,
            )
        return ProvisioningResult(
            ok=True, external_ref=email, raw_response=detail,
        )

    # ── Scope-level closure (admin "下线") ─────────────────────

    async def close_scope_on_provider(
        self,
        scope: CloudScopeRef,
        *,
        iam_account_ids: list[str] | None = None,
    ) -> CloseScopeResult:
        """Real cloud-side scope closure.

        For ``azure_subscription`` → POST the
        ``Microsoft.Subscription/cancel`` action on the subscription.
        Azure moves the subscription to ``Disabled`` immediately and
        permanently deletes it after a 90-day grace period
        (subscription holders can re-activate during the window through
        the portal — Helix surfaces this as ``cancelled`` in audit but
        the Helix-side bookkeeping still flips, consistent with the
        single source of truth contract).

        Anything else raises ``NotImplementedError``. Azure has no
        single-tenant IAM scopes — every Azure scope in Helix is a
        subscription. ``iam_account_ids`` is accepted for protocol
        symmetry with sibling drivers but ignored here.
        """
        del iam_account_ids  # unused; subscription-only model
        if scope.scope_type != "azure_subscription":
            raise NotImplementedError(
                f"Azure driver cannot close scope_type={scope.scope_type!r}",
            )
        sub_id = scope.external_id.strip()
        if not sub_id:
            return CloseScopeResult(
                ok=False, error_code="missing_subscription_id",
                error_message="Azure subscription id is empty",
            )
        try:
            token = await self._token(_ARM_BASE)
        except RuntimeError as exc:
            return CloseScopeResult(
                ok=False, error_code="auth_failed",
                error_message=str(exc),
            )

        # Pre-flight: read the subscription state. Azure returns 404
        # when the SP can't see the subscription (covers both "doesn't
        # exist" and "moved to a different tenant"); we treat that as
        # already_closed so the Helix-side bookkeeping doesn't get
        # stuck on a subscription Azure has already retired.
        async with self._http() as client:
            describe = await client.get(
                f"{_ARM_BASE}/subscriptions/{sub_id}",
                params={"api-version": "2022-12-01"},
                headers={"Authorization": f"Bearer {token}"},
            )
        if describe.status_code == 404:
            return CloseScopeResult(
                ok=True, already_closed=True,
                provider_state="not_found",
                detail={"reason": "Subscription not visible to this SP"},
            )
        if describe.status_code == 403:
            return CloseScopeResult(
                ok=False, error_code="access_denied",
                error_message=(
                    "Azure denied subscription read. The integration's "
                    "service principal needs Reader (or higher) on this "
                    "subscription, plus the cancel-subscription action via "
                    "Owner / Microsoft.Subscription/subscriptions/cancel."
                ),
            )
        if describe.status_code != 200:
            return CloseScopeResult(
                ok=False, error_code="describe_failed",
                error_message=(
                    f"Azure subscription describe {describe.status_code}: "
                    f"{_safe_body(describe, max_len=300)}"
                ),
            )
        state = str(describe.json().get("state") or "").strip()
        # Disabled / Deleted are post-cancel terminal states — treat as
        # already-closed. PastDue (billing problem, still active) falls
        # through to the cancel call below, and so does ``Warned``:
        # a subscription can be Warned either because it is on its way
        # OUT (a cancel already accepted — see
        # ``get_scope_closure_state``) or because Azure warned a live
        # subscription. Re-POSTing the cancel is harmless in the first
        # case and correct in the second, so this branch stays narrow
        # rather than guessing which one it is looking at.
        if state in {"Disabled", "Deleted"}:
            return CloseScopeResult(
                ok=True, already_closed=True,
                provider_state=state.lower(),
                detail={"prior_state": state},
            )

        async with self._http() as client:
            cancel = await client.post(
                f"{_ARM_BASE}/subscriptions/{sub_id}/providers/"
                "Microsoft.Subscription/cancel",
                params={"api-version": "2021-10-01"},
                headers={"Authorization": f"Bearer {token}"},
            )
        if cancel.status_code in (200, 202, 204):
            payload: dict[str, Any] = {}
            try:
                payload = cancel.json() if cancel.text else {}
            except ValueError:
                payload = {}
            return CloseScopeResult(
                ok=True,
                provider_state="cancelled",
                detail={
                    "grace_period_days": 90,
                    # Do NOT assert the post-call state here. This text
                    # used to read "Subscription moved to Disabled",
                    # which is not what Azure does: an accepted cancel
                    # lands the subscription in ``Warned`` first and it
                    # can sit there for hours (prod 2026-09-02:
                    # qa-efficiency was still Warned 48 minutes after a
                    # ``cancel/action`` the ARM activity log recorded as
                    # Succeeded). The archive tab reads this note, so a
                    # hardcoded state that disagrees with the provider
                    # is a lie shown to an admin. Describe the PATH; the
                    # actual state is what ``cmdb.decommission_reconcile``
                    # writes into ``decommission_provider_state``.
                    "note": (
                        "Azure accepted the cancel. The subscription "
                        "moves Warned → Disabled and is permanently "
                        "deleted ~90 days after this call."
                    ),
                    "response": payload,
                },
            )
        if cancel.status_code == 403:
            return CloseScopeResult(
                ok=False, error_code="access_denied",
                error_message=(
                    "Azure denied Microsoft.Subscription/cancel. The SP "
                    "needs the Subscription Cancel action — granted via "
                    "the Owner role on the subscription, or via a custom "
                    "role that includes "
                    "'Microsoft.Subscription/cancel/action'."
                ),
            )
        if cancel.status_code == 401:
            return CloseScopeResult(
                ok=False, error_code="auth_failed",
                error_message=(
                    "Azure rejected the bearer token for "
                    "Microsoft.Subscription/cancel. The cached ARM token "
                    "is invalid or expired — verify the service "
                    "principal's client_secret has not rotated/expired."
                ),
            )
        if cancel.status_code == 404:
            # Race against Azure deleting it concurrently.
            return CloseScopeResult(
                ok=True, already_closed=True,
                provider_state="not_found",
                detail={"reason": "Subscription disappeared during cancel"},
            )
        # Azure surfaces several recoverable preconditions as HTTP 400.
        # Dispatch on the symbolic code from the ARM error envelope so
        # the admin sees an actionable hint (and the audit detail
        # carries a stable ``error_code``) instead of the raw JSON
        # body. Anything we can't classify falls back to ``cancel_failed``
        # with the redacted body preserved for forensics.
        raw_arm_code, arm_message = _parse_arm_error(cancel)
        # Defense-in-depth: even though ARM ``error.code`` is documented
        # as a symbolic enum (CamelCase identifier) and shouldn't carry
        # credentials, a malformed top-level body could put arbitrary
        # JSON into the ``code`` field. Cap length and route through the
        # redactor so the value is safe to interpolate into the admin
        # message AND audit detail. (Round-52 security review.)
        arm_code = redact_provider_error(raw_arm_code, max_len=120)
        if cancel.status_code == 400 and arm_code:
            redacted_message = redact_provider_error(arm_message, max_len=400)
            if arm_code == "ResourcesExistingOnSubscription":
                return CloseScopeResult(
                    ok=False, error_code="resources_existing",
                    error_message=(
                        "Azure refuses to cancel a subscription that "
                        "still contains resources. Delete every "
                        "resource group in the subscription first — on "
                        "the portal: Subscription → Resource groups → "
                        "Delete; via CLI: "
                        "`az group list --subscription <id> -o tsv "
                        "--query '[].name' | xargs -I {} az group "
                        "delete --subscription <id> --name {} --yes "
                        "--no-wait` — then retry decommission. "
                        f"[Azure: {arm_code}]"
                    ),
                    detail={"arm_code": arm_code, "arm_message": redacted_message},
                )
            if arm_code == "SubscriptionNotInRegisteredState":
                # The pre-flight describe saw an active state, but
                # between that read and this POST Azure flipped the
                # subscription to a non-cancellable state (Deleted /
                # Disabled in mid-transition). Treat as idempotent
                # success so the Helix-side bookkeeping converges and
                # the admin doesn't see a confusing error on a
                # subscription that is, in fact, on its way out.
                return CloseScopeResult(
                    ok=True, already_closed=True,
                    provider_state="not_registered",
                    detail={
                        "reason": (
                            "Subscription is no longer in a registered "
                            "state — already cancelled or being deleted."
                        ),
                        "arm_code": arm_code,
                        "arm_message": redacted_message,
                    },
                )
            if arm_code in {
                "BillingAccountInactive", "PendingCharges",
                "OutstandingCharges", "BillingProfileNotActive",
            }:
                return CloseScopeResult(
                    ok=False, error_code="pending_charges",
                    error_message=(
                        "Azure refuses to cancel this subscription "
                        "while it has outstanding billing obligations. "
                        "Settle the balance on the Azure Cost Management "
                        "console (or work with the billing-account "
                        "owner) and retry. "
                        f"[Azure: {arm_code}]"
                    ),
                    detail={"arm_code": arm_code, "arm_message": redacted_message},
                )
            if arm_code in {
                "SubscriptionMustHaveOwner", "SubscriptionHasNoOwner",
            }:
                return CloseScopeResult(
                    ok=False, error_code="missing_owner",
                    error_message=(
                        "Azure requires an active Owner role assignment "
                        "to cancel this subscription. Assign Owner to a "
                        "human admin (NOT the service principal — Azure "
                        "rejects SP-only ownership for cancel) and "
                        f"retry. [Azure: {arm_code}]"
                    ),
                    detail={"arm_code": arm_code, "arm_message": redacted_message},
                )
            return CloseScopeResult(
                ok=False, error_code="cancel_precondition_failed",
                error_message=(
                    f"Azure rejected Microsoft.Subscription/cancel "
                    f"with code {arm_code!r}: {redacted_message}"
                ),
                detail={"arm_code": arm_code, "arm_message": redacted_message},
            )
        if cancel.status_code == 429:
            return CloseScopeResult(
                ok=False, error_code="throttled",
                error_message=(
                    "Azure ARM throttled Microsoft.Subscription/cancel. "
                    "Wait a few minutes (the Retry-After header carries "
                    "the exact backoff) and retry."
                ),
                detail={"retry_after": cancel.headers.get("Retry-After")},
            )
        if cancel.status_code >= 500:
            return CloseScopeResult(
                ok=False, error_code="transient_failure",
                error_message=(
                    f"Azure ARM returned {cancel.status_code} on "
                    "Microsoft.Subscription/cancel — server-side "
                    "failure, safe to retry."
                ),
            )
        return CloseScopeResult(
            ok=False, error_code="cancel_failed",
            error_message=(
                f"Azure subscription cancel {cancel.status_code}: "
                f"{_safe_body(cancel, max_len=300)}"
            ),
            # Defense-in-depth: route the parsed message through the
            # same redactor the user-facing message uses so audit
            # detail can't leak credential-shaped substrings either.
            detail={
                "arm_code": arm_code,
                "arm_message": redact_provider_error(
                    arm_message, max_len=400,
                ),
            } if arm_code else {},
        )

    # ── Close pre-flight + blocker cleanup ─────────────────────
    #
    # Why this pair exists: ``ResourcesExistingOnSubscription`` is not a
    # bug report, it is a work item. The pre-Round-58 behaviour turned
    # it into a red modal carrying an ``az group delete`` one-liner —
    # i.e. Helix telling the admin who just asked it to decommission a
    # subscription to go decommission the subscription. Preflight moves
    # that fact BEFORE the click (so the dialog can name what will be
    # destroyed, which is information needed to consent at all), and
    # ``clear_close_blockers`` does the work AFTER it.
    #
    # ARM api-versions are pinned deliberately:
    #   * ``2021-04-01`` — resourcegroups list/delete + subscription
    #     resources list. The oldest version that carries
    #     ``provisioningState`` on the RG (which is how "Deleting" is
    #     detected without polling the async operation URL).
    #   * ``2016-09-01`` — Microsoft.Authorization/locks. Still the GA
    #     version for management locks; there is no newer one.

    _RG_API_VERSION = "2021-04-01"
    _LOCKS_API_VERSION = "2016-09-01"
    # Display cap on the resources listed inside one resource group.
    # The full count always travels in ``child_count`` — the cap is
    # about a readable dialog, not about hiding scale.
    _PREFLIGHT_CHILDREN_CAP = 8

    async def _arm_list_all(
        self, url: str, token: str, *, params: dict[str, str],
        collection: str = "value",
    ) -> tuple[list[dict[str, Any]], str | None]:
        """GET an ARM collection, following ``nextLink`` to the end.

        Returns ``(rows, error)``. ``error`` is a redacted one-liner
        when any page failed — callers surface it as "could not check"
        rather than as "nothing found", because an empty list from a
        failed read is the one lie this whole mechanism must not tell.

        ``nextLink`` is an absolute URL that already carries the
        api-version and a continuation token, so the follow-up requests
        must NOT re-apply ``params`` (ARM 400s on a duplicated
        ``$skipToken``).
        """
        rows: list[dict[str, Any]] = []
        next_url: str | None = url
        send_params: dict[str, str] | None = dict(params)
        # Hard page ceiling: a subscription with tens of thousands of
        # resources should degrade to "we saw the first N pages" rather
        # than pin the worker on one ARM collection.
        for _ in range(50):
            if not next_url:
                break
            async with self._http() as client:
                resp = await client.get(
                    next_url,
                    params=send_params,
                    headers={"Authorization": f"Bearer {token}"},
                )
            if resp.status_code != 200:
                return rows, (
                    f"ARM {resp.status_code} on {url.rsplit('/', 1)[-1]}: "
                    f"{_safe_body(resp, max_len=200)}"
                )
            try:
                payload = resp.json()
            except ValueError:
                return rows, "ARM returned a non-JSON collection body"
            if not isinstance(payload, dict):
                return rows, "ARM returned a non-object collection body"
            page = payload.get(collection)
            if isinstance(page, list):
                rows.extend(r for r in page if isinstance(r, dict))
            next_url = str(payload.get("nextLink") or "") or None
            send_params = None
        return rows, None

    async def preflight_close_scope(
        self, scope: CloudScopeRef,
    ) -> ClosePreflight:
        """Enumerate what stands between this subscription and cancelled.

        Three ARM reads, all read-only:

        1. resource groups (name, location, ``provisioningState``)
        2. every resource in the subscription — ONE call, then grouped
           by the resource-group segment of each ARM id. The per-RG
           ``resourceGroups/{n}/resources`` endpoint would be N calls
           for the same answer and would throttle on a busy tenant.
        3. management locks — the blocker's blocker. A
           ``CanNotDelete`` lock makes the resource-group delete fail
           with 409, so cleanup that ignores locks succeeds on empty
           test subscriptions and fails on exactly the real ones that
           needed it.

        A resource group already in ``provisioningState=Deleting`` is
        reported with ``kind='resource_group'`` and
        ``detail`` naming the state: it IS still a blocker (Azure keeps
        refusing the cancel until it is gone) but it needs no new
        delete call, only patience.
        """
        if scope.scope_type != "azure_subscription":
            return ClosePreflight(supported=False)
        sub_id = scope.external_id.strip()
        if not sub_id:
            return ClosePreflight(
                supported=True, checked=False,
                error_code="missing_subscription_id",
                error_message="Azure subscription id is empty",
            )
        try:
            token = await self._token(_ARM_BASE)
        except RuntimeError as exc:
            return ClosePreflight(
                supported=True, checked=False,
                cleanup_supported=True,
                error_code="auth_failed", error_message=str(exc),
            )

        groups, rg_err = await self._arm_list_all(
            f"{_ARM_BASE}/subscriptions/{sub_id}/resourcegroups",
            token, params={"api-version": self._RG_API_VERSION},
        )
        if rg_err:
            return ClosePreflight(
                supported=True, checked=False, cleanup_supported=True,
                error_code="preflight_read_failed", error_message=rg_err,
            )
        resources, res_err = await self._arm_list_all(
            f"{_ARM_BASE}/subscriptions/{sub_id}/resources",
            token, params={"api-version": self._RG_API_VERSION},
        )
        locks, lock_err = await self._arm_list_all(
            f"{_ARM_BASE}/subscriptions/{sub_id}/providers/"
            "Microsoft.Authorization/locks",
            token, params={"api-version": self._LOCKS_API_VERSION},
        )

        # Group resources under their RG. ``resourceGroup`` is present
        # on the subscription-level resources response, but parse the
        # ARM id as the fallback — the field is absent on a few
        # provider shapes and the id always carries the segment.
        by_group: dict[str, list[dict[str, Any]]] = {}
        for row in resources:
            rg_name = str(row.get("resourceGroup") or "").strip()
            if not rg_name:
                parsed_rg, _, _ = _parse_azure_resource_id(
                    str(row.get("id") or ""),
                )
                rg_name = (parsed_rg or "").strip()
            by_group.setdefault(rg_name.lower(), []).append(row)

        # Locks live on the RG or on a resource inside it. Either way
        # they block the RG delete, so they are their own blockers and
        # get deleted first during cleanup.
        lock_blockers: list[CloseBlocker] = []
        for lock in locks:
            lock_id = str(lock.get("id") or "").strip()
            if not lock_id:
                continue
            level = str(
                (lock.get("properties") or {}).get("level") or "",
            ).strip()
            lock_blockers.append(CloseBlocker(
                key=lock_id,
                kind="lock",
                name=str(lock.get("name") or lock_id.rsplit("/", 1)[-1]),
                detail=level or None,
                # Deleting a management lock needs
                # ``Microsoft.Authorization/locks/delete``, which
                # Contributor does NOT carry (Owner / User Access
                # Administrator do). We do not know which the SP holds
                # until we try, so report it as removable and let the
                # 403 during cleanup be the authority — guessing here
                # would either block a decommission that would have
                # worked or promise one that cannot.
                removable=True,
            ))

        rg_blockers: list[CloseBlocker] = []
        for group in groups:
            name = str(group.get("name") or "").strip()
            if not name:
                continue
            state = str(
                (group.get("properties") or {}).get("provisioningState")
                or "",
            ).strip()
            children = by_group.get(name.lower(), [])
            labels = [
                f"{_azure_resource_kind_label(str(c.get('type') or ''))} "
                f"{c.get('name') or '—'}".strip()
                for c in children[: self._PREFLIGHT_CHILDREN_CAP]
            ]
            detail_bits = [str(group.get("location") or "").strip()]
            if state and state.lower() != "succeeded":
                detail_bits.append(state)
            rg_blockers.append(CloseBlocker(
                key=str(group.get("id") or "").strip()
                or f"/subscriptions/{sub_id}/resourceGroups/{name}",
                kind="resource_group",
                name=name,
                detail=" · ".join(b for b in detail_bits if b) or None,
                children=labels,
                child_count=len(children),
                removable=True,
            ))

        # A partial read is still worth showing, but it must be labelled
        # as partial: ``checked=False`` when the resource or lock
        # enumeration failed, so the UI says "还有未能确认的内容" rather
        # than presenting an under-count as the whole truth.
        partial = [e for e in (res_err, lock_err) if e]
        return ClosePreflight(
            supported=True,
            checked=not partial,
            cleanup_supported=True,
            blockers=[*lock_blockers, *rg_blockers],
            error_code="preflight_partial" if partial else None,
            error_message="; ".join(partial) or None,
            detail={
                "resource_group_count": len(rg_blockers),
                "resource_count": len(resources),
                "lock_count": len(lock_blockers),
            },
        )

    async def clear_close_blockers(
        self, scope: CloudScopeRef, *, keys: list[str],
    ) -> ClearBlockersResult:
        """Delete the confirmed locks + resource groups.

        Order matters: **locks first**. A ``CanNotDelete`` lock on a
        resource group (or on anything inside it) makes the group
        delete fail 409, so a cleanup that deletes groups first burns
        one retry cycle per lock and, on a subscription where the lock
        cannot be removed, never converges.

        Only keys in ``keys`` are touched — the set the admin was shown
        and confirmed. A resource group created after that dialog was
        rendered is NOT in the set and is therefore never deleted here;
        the caller notices it on the next preflight and asks for
        re-confirmation instead.

        Azure semantics that shape the return value:

        * RG delete → ``202 Accepted`` and the group enters
          ``Deleting``. There is no synchronous form of this call, so
          ``in_progress=True`` whenever anything was accepted.
        * RG delete on a group that is already gone → ``204``. That is
          ``completed``, not a failure — this method is called
          repeatedly by the resume job and must be idempotent.
        * ``409`` → almost always a lock we could not remove (or a
          child resource with its own delete-protection). Reported as
          ``failed`` with the ARM code so the escalation names the
          cause instead of "cleanup failed".
        """
        if scope.scope_type != "azure_subscription":
            return ClearBlockersResult(
                ok=False, error_code="cleanup_unsupported",
                error_message=(
                    f"Azure driver cannot clean scope_type="
                    f"{scope.scope_type!r}"
                ),
            )
        sub_id = scope.external_id.strip()
        if not sub_id:
            return ClearBlockersResult(
                ok=False, error_code="missing_subscription_id",
                error_message="Azure subscription id is empty",
            )
        try:
            token = await self._token(_ARM_BASE)
        except RuntimeError as exc:
            return ClearBlockersResult(
                ok=False, error_code="auth_failed",
                error_message=str(exc),
            )

        # Cross-subscription safety: every key must address THIS
        # subscription. A key that doesn't is a caller bug (or a stale
        # confirmation carried across a scope), and deleting on the
        # strength of it would take out resources in an unrelated
        # subscription. Refuse the whole call rather than the key.
        prefix = f"/subscriptions/{sub_id}/".lower()
        foreign = [
            k for k in keys
            if not str(k or "").lower().startswith(prefix)
        ]
        if foreign:
            return ClearBlockersResult(
                ok=False, error_code="key_scope_mismatch",
                error_message=(
                    f"{len(foreign)} blocker key(s) do not belong to "
                    f"subscription {sub_id} — refusing to delete"
                ),
                detail={"foreign_keys": foreign[:5]},
            )

        locks = [k for k in keys if "/providers/microsoft.authorization/locks/"
                 in k.lower()]
        rgs = [k for k in keys if k not in locks]

        started: list[str] = []
        completed: list[str] = []
        failed: list[dict[str, str]] = []

        for lock_key in locks:
            async with self._http() as client:
                resp = await client.delete(
                    f"{_ARM_BASE}{lock_key}",
                    params={"api-version": self._LOCKS_API_VERSION},
                    headers={"Authorization": f"Bearer {token}"},
                )
            if resp.status_code in (200, 202, 204, 404):
                completed.append(lock_key)
                continue
            arm_code, arm_message = _parse_arm_error(resp)
            failed.append({
                "key": lock_key,
                "error": (
                    f"{resp.status_code} "
                    f"{redact_provider_error(arm_code, max_len=80)}: "
                    f"{redact_provider_error(arm_message, max_len=200)}"
                    if arm_code else
                    f"{resp.status_code}: {_safe_body(resp, max_len=200)}"
                ),
            })

        for rg_key in rgs:
            rg_name = rg_key.rstrip("/").rsplit("/", 1)[-1]
            if not rg_name:
                failed.append({"key": rg_key, "error": "unparseable key"})
                continue
            async with self._http() as client:
                resp = await client.delete(
                    f"{_ARM_BASE}/subscriptions/{sub_id}"
                    f"/resourcegroups/{rg_name}",
                    params={"api-version": self._RG_API_VERSION},
                    headers={"Authorization": f"Bearer {token}"},
                )
            if resp.status_code == 202:
                started.append(rg_key)
                continue
            if resp.status_code in (200, 204, 404):
                # 204 = "no such group" per ARM's delete contract, 404
                # = same via the generic path. Already gone.
                completed.append(rg_key)
                continue
            arm_code, arm_message = _parse_arm_error(resp)
            failed.append({
                "key": rg_key,
                "error": (
                    f"{resp.status_code} "
                    f"{redact_provider_error(arm_code, max_len=80)}: "
                    f"{redact_provider_error(arm_message, max_len=200)}"
                    if arm_code else
                    f"{resp.status_code}: {_safe_body(resp, max_len=200)}"
                ),
            })

        return ClearBlockersResult(
            # Progress on anything is ok=True: the resume loop re-reads
            # reality and will report the leftovers next cycle. ok=False
            # is reserved for "we could not act at all", which is what
            # deserves an escalation.
            ok=bool(started or completed) or not failed,
            started=started, completed=completed, failed=failed,
            in_progress=bool(started),
            error_code="cleanup_partial" if failed else None,
            error_message=(
                f"{len(failed)} blocker(s) could not be deleted"
                if failed else None
            ),
            detail={"locks": len(locks), "resource_groups": len(rgs)},
        )

    # ── Post-close reconciliation (read-only) ──────────────────

    async def get_scope_closure_state(
        self, scope: CloudScopeRef,
    ) -> ScopeClosureState:
        """Re-read ARM's view of a subscription Helix already cancelled.

        Cancelling walks a subscription ``Warned`` → ``Disabled``; Azure
        deletes it ~90 days after the call. **Re-activation is
        self-service in the portal for that entire window** and never
        reaches Helix, so ``Enabled`` on a scope Helix marked closed is a
        live, ungoverned subscription — which is the whole reason this
        read exists. Everything on the way out is ``pending``: an alert
        that fires on a closure doing exactly what it should is a false
        alarm, and this module's alerts are terminal.
        """
        if scope.scope_type != "azure_subscription":
            raise NotImplementedError(
                "Azure driver has no post-close state for "
                f"scope_type={scope.scope_type!r}",
            )
        sub_id = (scope.external_id or "").strip()
        if not sub_id:
            return ScopeClosureState(
                disposition="unknown",
                error_message="Azure subscription id is empty",
            )
        try:
            token = await self._token(_ARM_BASE)
        except RuntimeError as exc:
            return ScopeClosureState(
                disposition="unknown", error_message=str(exc)[:300],
            )
        try:
            async with self._http() as client:
                resp = await client.get(
                    f"{_ARM_BASE}/subscriptions/{sub_id}",
                    params={"api-version": "2022-12-01"},
                    headers={"Authorization": f"Bearer {token}"},
                )
        except Exception as exc:  # noqa: BLE001
            return ScopeClosureState(
                disposition="unknown",
                error_message=f"Azure subscription read: {exc}"[:300],
            )
        if resp.status_code == 404:
            return ScopeClosureState(
                disposition="closed", provider_state="not_found",
                detail={"reason": "Subscription no longer visible to this SP"},
            )
        if resp.status_code != 200:
            return ScopeClosureState(
                disposition="unknown",
                error_message=(
                    f"Azure subscription read {resp.status_code}: "
                    f"{_safe_body(resp, max_len=250)}"
                ),
            )
        try:
            state = str((resp.json() or {}).get("state") or "").strip()
        except Exception:  # noqa: BLE001
            state = ""
        detail = {"azure_state": state}
        # State → disposition, and the reason each mapping is what it is:
        #
        # ``Deleted``   the 90-day window elapsed. Terminal.
        # ``Disabled``  cancel has landed; Azure deletes it ~90 days on.
        # ``Warned``    **also a post-cancel state**, not only the
        #               pre-cancellation grace state this code used to
        #               assume. Measured in prod 2026-09-02: subscription
        #               918de8c4 (qa-efficiency) read ``Enabled`` in
        #               Helix's own pre-close snapshot, Helix POSTed
        #               ``Microsoft.Subscription/cancel``, ARM's activity
        #               log recorded ``cancel/action`` **Succeeded**, and
        #               the subscription then sat in ``Warned`` — 0
        #               resources, still Warned 48 minutes later.
        #               Calling that ``alive`` produced a false "someone
        #               re-activated this" alert AND, because ``active``
        #               is in ``TERMINAL_PROVIDER_STATES``, dropped the
        #               subscription out of the reconcile queue for good,
        #               so the real Disabled → Deleted transition would
        #               never have been confirmed. It is ``pending``: a
        #               closure in flight, re-read tomorrow.
        # ``Enabled`` / ``PastDue``
        #               genuinely usable subscriptions. Re-activation in
        #               the portal returns a subscription to ``Enabled``
        #               (never to ``Warned``), so the alert this module
        #               exists to raise is preserved in full by keeping
        #               exactly these two — and only these two — alive.
        if state in {"Deleted"}:
            return ScopeClosureState(
                disposition="closed", provider_state="deleted", detail=detail,
            )
        if state in {"Disabled"}:
            return ScopeClosureState(
                disposition="pending", provider_state="disabled",
                detail=detail,
            )
        if state in {"Warned"}:
            return ScopeClosureState(
                disposition="pending", provider_state="warned", detail=detail,
            )
        if state in {"Enabled", "PastDue"}:
            return ScopeClosureState(
                disposition="alive", provider_state="active", detail=detail,
                error_message=(
                    f"Azure reports this subscription as {state!r}, but "
                    "Helix has it marked closed and has already disabled "
                    "its grants. Re-activation is self-service in the "
                    "portal during the 90-day window."
                ),
            )
        return ScopeClosureState(
            disposition="unknown", detail=detail,
            error_message=(
                f"Azure returned an unrecognised subscription state {state!r}"
            ),
        )
    async def _list_principal_assignments(
        self, client: httpx.AsyncClient, token: str,
        scope_path: str, principal_id: str,
    ) -> list[str]:
        list_url = (
            f"{_ARM_BASE}{scope_path}/providers/Microsoft.Authorization/"
            "roleAssignments?api-version=2022-04-01"
            f"&$filter=principalId eq '{principal_id}'"
        )
        out: list[str] = []
        next_url: str | None = list_url
        _MAX_PAGES = 500
        pages_walked = 0
        while next_url and pages_walked < _MAX_PAGES:
            resp = await client.get(
                next_url, headers={"Authorization": f"Bearer {token}"},
            )
            pages_walked += 1
            if resp.status_code == 403:
                # Per-subscription permission gap — caller records and
                # continues to the next subscription.
                raise RuntimeError(
                    f"403 on {scope_path} roleAssignments list — SP needs "
                    "User Access Administrator on this subscription",
                )
            if resp.status_code != 200:
                raise RuntimeError(
                    f"list roleAssignments {resp.status_code}: "
                    f"{_safe_body(resp, max_len=200)}",
                )
            payload = resp.json()
            for ra in payload.get("value") or []:
                name = str(ra.get("name") or "")
                if name:
                    out.append(name)
            next_url = payload.get("nextLink")
        if pages_walked >= _MAX_PAGES and next_url:
            logger.warning(
                "_list_principal_assignments: hit %d page cap on %s principal=%s "
                "— %d assignments collected; remaining pages skipped",
                _MAX_PAGES, scope_path, principal_id, len(out),
            )
        return out

    # ── Reverse grants discovery (cloud → Helix) ─────────────────

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

        Walks ARM ``roleAssignments`` at ``atScope()`` — includes
        assignments inherited from management-group / root via the
        ``$filter=atScope()`` semantics Azure exposes (management-
        group-level roles that trickle down are visible here as
        ``scope=/providers/.../managementGroups/<id>`` rows; we keep
        them so "scope owner" detection picks up Enterprise Admins
        who hold Owner at root).

        Principal resolution: AAD object-id → email via a single
        batched ``/directoryObjects/getByIds`` Graph call (1 call per
        ≤1000 ids instead of N round-trips). We filter to
        ``principalType='User'`` before the Graph call, so service
        principals and groups never hit Graph.
        """
        from app.integrations.driver_base import DiscoveredGrant

        # Subscription GUID goes straight into the ARM path below, so a
        # sub-tenant row (``azure_resource_group``) must never reach it
        # — that 400 is what took cmdb.grants_sync red on 2026-07-29.
        # See ``_require_subscription_scope``.
        _require_subscription_scope(scope, "role assignments")
        arm_token = await self._token(_ARM_BASE)
        scope_path = f"/subscriptions/{scope.external_id}"
        list_url: str | None = (
            f"{_ARM_BASE}{scope_path}/providers/Microsoft.Authorization/"
            "roleAssignments?api-version=2022-04-01&$filter=atScope()"
        )
        assignments: list[dict[str, Any]] = []
        async with self._http() as client:
            while list_url:
                resp = await client.get(
                    list_url,
                    headers={"Authorization": f"Bearer {arm_token}"},
                )
                if resp.status_code in (403, 404):
                    logger.info(
                        "Azure list_scope_grants: %s on %s — skip",
                        resp.status_code, scope_path,
                    )
                    return []
                if resp.status_code != 200:
                    raise RuntimeError(
                        f"roleAssignments list {resp.status_code}: "
                        f"{_safe_body(resp, max_len=300)}",
                    )
                payload = resp.json()
                assignments.extend(payload.get("value") or [])
                list_url = payload.get("nextLink")

        # Prune to user principals — groups / SPs / managed identities
        # don't map to a Helix person.
        user_rows: list[dict[str, Any]] = []
        for ra in assignments:
            props = ra.get("properties") or {}
            if str(props.get("principalType") or "").lower() != "user":
                continue
            pid = str(props.get("principalId") or "")
            if pid:
                user_rows.append(ra)

        # Resolve AAD object ids → UPN/mail in one batch call. The
        # Graph endpoint ``/directoryObjects/getByIds`` accepts up to
        # 1000 ids per call; we page just in case.
        principal_ids = list({
            (r.get("properties") or {}).get("principalId") for r in user_rows
        })
        principal_ids = [p for p in principal_ids if p]
        principal_email: dict[str, str] = {}
        if principal_ids:
            graph_token = await self._token("https://graph.microsoft.com")
            async with self._http() as client:
                for i in range(0, len(principal_ids), 1000):
                    batch = principal_ids[i:i + 1000]
                    r = await client.post(
                        f"{_GRAPH_BASE}/directoryObjects/getByIds",
                        json={
                            "ids": batch,
                            "types": ["user"],
                        },
                        headers={
                            "Authorization": f"Bearer {graph_token}",
                            "Content-Type": "application/json",
                        },
                    )
                    if r.status_code == 403:
                        # CRITICAL: Graph 403 means we cannot resolve any
                        # principalId → email. Returning ``[]`` (or a
                        # partial list) silently looks like "no users
                        # assigned", which the grants-sync sweep would
                        # interpret as "all grants on this scope have
                        # been revoked outside Helix" → mass-revoke. Raise
                        # ``ProviderAuthError`` so the sync handler
                        # marks the run ``error`` instead of writing a
                        # corrupting empty result.
                        raise ProviderAuthError(
                            "Azure Graph /directoryObjects/getByIds "
                            "returned 403 — service principal needs "
                            "Directory.Read.All to resolve role-"
                            "assignment principals. Refusing to write "
                            "a partial grants snapshot (would mass-"
                            "revoke against drift).",
                        )
                    if r.status_code != 200:
                        # Same data-corrupting risk as 403. A transient
                        # 500/429 on ANY batch leaves later grants
                        # unresolved → ``not email`` → silently dropped
                        # → grants-sync sweep treats as drift and
                        # mass-revokes. Raise loudly so the sweep
                        # aborts with a partial-failure status rather
                        # than committing a corrupting snapshot.
                        raise ProviderAuthError(
                            f"Azure Graph /directoryObjects/getByIds "
                            f"returned {r.status_code} (transient) — "
                            f"refusing to write a partial grants "
                            f"snapshot. Body: {_safe_body(r, max_len=200)}",
                        )
                    for u in r.json().get("value") or []:
                        uid = str(u.get("id") or "")
                        mail = u.get("mail") or u.get("userPrincipalName")
                        if uid and mail:
                            principal_email[uid] = str(mail).lower()

        out: list[DiscoveredGrant] = []
        # Round-42 audit: track principals dropped due to missing email
        # so a persistent Graph projection bug (e.g. AAD guest users
        # with mail=None) is operator-visible. A single drop is normal
        # (deleted user between assignment + Graph call), but a cluster
        # of drops needs investigation — without this counter the
        # ``continue`` below was invisible.
        dropped_no_email = 0
        for ra in user_rows:
            props = ra.get("properties") or {}
            pid = str(props.get("principalId") or "")
            email = principal_email.get(pid)
            if not email:
                # User object gone from AAD (deleted between
                # roleAssignment creation and our Graph call) or Graph
                # forbidden. Skip — we can't map to a Helix person.
                dropped_no_email += 1
                continue
            role_key = _normalize_role_key(
                str(props.get("roleDefinitionId") or ""),
            )
            if not role_key:
                continue
            out.append(DiscoveredGrant(
                principal_email=email,
                principal_type="user",
                role_key=role_key,
                role_display_name=None,  # lookup via roles catalog
                external_ref=f"{scope_path}::{ra.get('name') or ''}",
                scope=scope,
                granted_at=_parse_iso(props.get("createdOn")),
                raw={
                    "principalId": pid,
                    "roleDefinitionId": props.get("roleDefinitionId"),
                    "scope": props.get("scope"),
                },
            ))
        if dropped_no_email > 0:
            logger.info(
                "Azure list_scope_grants scope=%s: dropped %d role "
                "assignment(s) with no resolvable email (AAD guest / "
                "deleted user). Total resolved: %d.",
                scope_path, dropped_no_email, len(out),
            )
        return out

    # ── Last login (Graph signIns, optional) ─────────────────────

    async def get_last_login(
        self, ctx: CloudAccessContext,
    ) -> datetime | None:
        """Best-effort last-sign-in via Graph ``auditLogs/signIns``.

        Requires ``AuditLog.Read.All`` + an Entra ID P1 license on the
        tenant. Returns None on 403/404 so the inactivity sweep treats
        the grant as unknown and skips — this is the safer default.
        """
        try:
            safe = self._sanitize_email_for_odata(ctx.principal_email)
        except CloudPrincipalNotFoundError:
            return None
        try:
            token = await self._token("https://graph.microsoft.com")
        except Exception as exc:  # noqa: BLE001
            logger.info("Azure get_last_login: token fetch failed: %s", exc)
            return None
        # Filter to successful sign-ins by UPN, newest first. The signIns
        # endpoint only returns 30 days by default — for older data the
        # grant is long past any inactivity threshold anyway.
        filt = (
            f"userPrincipalName eq '{safe}' "
            "and status/errorCode eq 0"
        )
        async with self._http() as client:
            resp = await client.get(
                f"{_GRAPH_BETA_BASE}/auditLogs/signIns",
                params={
                    "$filter": filt,
                    "$top": "1",
                    "$orderby": "createdDateTime desc",
                },
                headers={"Authorization": f"Bearer {token}"},
            )
        if resp.status_code in (401, 403, 404):
            # AuditLog.Read.All not granted, or tenant lacks P1 license.
            return None
        if resp.status_code != 200:
            logger.info(
                "Azure signIns lookup non-200: %s %s",
                resp.status_code, _safe_body(resp, max_len=200),
            )
            return None
        rows = resp.json().get("value") or []
        if not rows:
            return None
        raw = rows[0].get("createdDateTime")
        if not raw:
            return None
        try:
            return datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
        except ValueError:
            return None

    # ``list_cognitive_services_resources`` removed in PR #253 — its
    # only caller was the retired ``ai_spending.azure_openai_resources_sync``
    # cron (which populated the dropped ``cmdb_ai_resources`` table).
    # AI cost classification is now service-name-based directly on
    # ``cmdb_cost_facts``.

    # ── Microsoft Foundry auto-provisioning (apikey autogrant) ───
    #
    # Four primitives the apikey ``azure_openai_autogrant`` module
    # composes when an ApiKeyAssignment for the Azure OpenAI / Foundry
    # provider transitions to "approved":
    #
    #   1. ``ensure_resource_group`` — idempotent PUT.
    #   2. ``provision_foundry_account`` — create the parent
    #      ``Microsoft.CognitiveServices/accounts`` of
    #      ``kind=AIServices`` with ``allowProjectManagement=True`` so
    #      Foundry projects can be nested underneath. Poll until
    #      ``provisioningState=Succeeded`` (a freshly-created account
    #      still in ``Creating`` rejects child-project PUTs and
    #      ``listKeys`` with 409). Replaces the legacy
    #      ``kind=OpenAI`` flow because Microsoft is consolidating
    #      Azure OpenAI Service under the Foundry surface — the
    #      Foundry portal's "Create a project" wizard creates exactly
    #      this shape (RG → AIServices account → project).
    #   3. ``provision_foundry_project`` — create a Foundry project
    #      (``Microsoft.CognitiveServices/accounts/{name}/projects/{project}``)
    #      with displayName + description. Poll until
    #      ``provisioningState=Succeeded`` so the project's inference
    #      endpoint is live before the assignee tries to use it.
    #   4. ``list_foundry_account_keys`` — POST the parent account's
    #      listKeys action; returns ``{"key1": ..., "key2": ...}``.
    #      Foundry projects share keys with the parent AIServices
    #      account (no per-project keys — the project URL embeds the
    #      project name and the key authenticates against the
    #      account-level endpoint).
    #
    # SP role requirements (admin-side prep):
    #   * ``Cognitive Services Contributor`` at the subscription
    #     scope — for creating + reading keys on AIServices accounts
    #     AND child projects (the project sub-resource inherits
    #     account-level RBAC).
    #   * ``Contributor`` at the subscription scope — for the
    #     ``Microsoft.Resources/resourceGroups/write`` action.
    # If the SP lacks either, the caller surfaces the 403 verbatim
    # with response body so the admin can diagnose without diving
    # into ARM logs.

    async def ensure_resource_group(
        self,
        *,
        subscription_id: str,
        name: str,
        location: str,
        tags: dict[str, str] | None = None,
    ) -> dict[str, Any]:
        """Idempotently create or update an Azure Resource Group.

        ARM ``PUT /resourcegroups/{name}`` is upsert-style: 200 on
        existing (returns updated body), 201 on fresh-create. We
        accept both as success. ``tags`` are merged-replace per ARM
        semantics (PUT overwrites; admins editing tags out-of-band
        will see Helix re-apply on the next autogrant for the same
        RG, but per design each autogrant creates a fresh RG so this
        is mostly a no-op past the first PUT).
        """
        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{subscription_id}"
            f"/resourcegroups/{name}?api-version=2021-04-01"
        )
        body: dict[str, Any] = {"location": location}
        if tags:
            body["tags"] = tags
        async with self._http() as client:
            resp = await client.put(
                url, json=body,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
            )
        if resp.status_code not in (200, 201):
            raise RuntimeError(
                f"ensure_resource_group {resp.status_code}: "
                f"{_safe_body(resp, max_len=400)}",
            )
        return resp.json()

    # ARM API version for the Foundry surface. ``2025-06-01`` is the
    # GA stable that introduced ``allowProjectManagement`` on the
    # AIServices account and the ``accounts/projects`` sub-resource —
    # both required to mirror what the Azure portal's "Create a
    # project" wizard does. Bumping past this version requires
    # re-validating that every shape we send is still accepted; the
    # change-log is at
    # ``learn.microsoft.com/.../change-log/accounts/projects``.
    _FOUNDRY_API_VERSION = "2025-06-01"

    async def provision_foundry_account(
        self,
        *,
        subscription_id: str,
        resource_group: str,
        account_name: str,
        location: str,
        custom_subdomain: str,
        tags: dict[str, str] | None = None,
        sku_name: str = "S0",
        poll_seconds: int = 300,
    ) -> dict[str, Any]:
        """Create a ``Microsoft.CognitiveServices/accounts`` of
        ``kind=AIServices`` (Microsoft Foundry resource) and poll until
        ``provisioningState=Succeeded``.

        Sets ``allowProjectManagement=True`` so child Foundry projects
        can be nested underneath via :meth:`provision_foundry_project`
        — without it the project PUT returns 400 ``ProjectsNotEnabled``.
        Identity is ``SystemAssigned`` so the account can later be
        granted access to user-side resources (storage, search) without
        an explicit identity dance — this matches the portal default.

        ``custom_subdomain`` is the globally-unique subdomain the
        Foundry endpoint lives at (``https://{subdomain}.cognitive
        services.azure.com`` and ``https://{subdomain}.services.ai
        .azure.com``). Globally unique across Azure — name collisions
        return 400 and the caller must retry with a different subdomain
        (the autogrant uses an assignment-id-derived suffix to make
        names deterministic-yet-unique).

        Polls every 5s up to ``poll_seconds`` (default 5 min). Cognitive
        Services creation usually completes in 30-60s but can stretch
        to a few minutes in busy regions; failing fast risks a still-
        Creating state where the next ``provision_foundry_project`` /
        ``listKeys`` call returns 409. Returns the final ``GET account``
        body so the caller has the endpoint URL
        (``properties.endpoint``) and the ARM resource id (``id``).
        """
        token = await self._token(_ARM_BASE)
        # Track when we last fetched the AAD token so the polling
        # loop only re-fetches when the token is genuinely stale.
        # AAD app-token default TTL is 60 min; we refresh after 30 min
        # (well before expiry) to keep the failure surface narrow,
        # but NOT every poll tick — fetching ~60 tokens per provision
        # would burn AAD rate-limit headroom for no benefit.
        token_acquired_at = 0  # seconds since loop start
        base_url = (
            f"{_ARM_BASE}/subscriptions/{subscription_id}"
            f"/resourceGroups/{resource_group}/providers/"
            f"Microsoft.CognitiveServices/accounts/{account_name}"
            f"?api-version={self._FOUNDRY_API_VERSION}"
        )
        body: dict[str, Any] = {
            "location": location,
            "kind": "AIServices",
            "identity": {"type": "SystemAssigned"},
            "sku": {"name": sku_name},
            "properties": {
                # ``allowProjectManagement`` unlocks the
                # ``accounts/projects`` sub-resource — the whole point
                # of provisioning a Foundry account vs the legacy
                # kind=OpenAI account.
                "allowProjectManagement": True,
                "customSubDomainName": custom_subdomain,
                "disableLocalAuth": False,
                "publicNetworkAccess": "Enabled",
            },
        }
        if tags:
            body["tags"] = tags
        async with self._http() as client:
            create_resp = await client.put(
                base_url, json=body,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
            )
            if create_resp.status_code not in (200, 201):
                raise RuntimeError(
                    f"create_foundry_account {create_resp.status_code}: "
                    f"{_safe_body(create_resp, max_len=400)}",
                )

            # Poll until Succeeded / Failed. ARM also exposes
            # async-operation tracking via ``Azure-AsyncOperation``
            # header but the simpler GET-loop is sufficient here and
            # matches the existing test_connection style elsewhere
            # in this driver.
            deadline = poll_seconds
            elapsed = 0
            current = create_resp.json()
            while True:
                state = (
                    (current.get("properties") or {}).get("provisioningState")
                    or ""
                )
                if state == "Succeeded":
                    return current
                if state in ("Failed", "Canceled"):
                    raise RuntimeError(
                        f"provision_foundry_account ended in {state}: "
                        f"{str(current)[:400]}",
                    )
                if elapsed >= deadline:
                    raise RuntimeError(
                        f"provision_foundry_account timeout after "
                        f"{deadline}s; last state={state!r}",
                    )
                await asyncio.sleep(5)
                elapsed += 5
                # Refresh SP token only when it's genuinely stale (>30
                # min old). Fetching a fresh token on every 5s tick
                # would burn ~60 AAD requests per provision and
                # eat into the tenant's AAD rate-limit headroom for
                # no benefit — token TTL is 60 min by default.
                if elapsed - token_acquired_at >= 1800:
                    token = await self._token(_ARM_BASE)
                    token_acquired_at = elapsed
                get_resp = await client.get(
                    base_url,
                    headers={"Authorization": f"Bearer {token}"},
                )
                if get_resp.status_code != 200:
                    raise RuntimeError(
                        f"poll_foundry_account {get_resp.status_code}: "
                        f"{_safe_body(get_resp, max_len=400)}",
                    )
                current = get_resp.json()

    async def delete_azure_openai_account(
        self,
        *,
        subscription_id: str,
        resource_group: str,
        account_name: str,
    ) -> dict[str, Any]:
        """ARM DELETE on a ``Microsoft.CognitiveServices/accounts`` row.

        Used by the apikey revoke path: when an Azure OpenAI auto-granted
        assignment is revoked we must invalidate the underlying account
        keys (the credential is the ``api-key`` header against the AOAI
        endpoint; clearing Helix's local copy doesn't help anyone who
        already exfiltrated it). DELETE on the account both invalidates
        the keys AND stops the per-resource billing meter.

        Idempotent: 404 → already gone → success; other non-2xx → raise.
        Returns a small audit dict so callers can attribute the cleanup.
        """
        from urllib.parse import quote as _q
        token = await self._token(_ARM_BASE)
        # Round-20 post-fix: defense-in-depth — quote the path segments
        # so a malformed ``account_name`` / ``resource_group`` (e.g.
        # injected ``/`` or ``?``) can't escape the ARM path and target
        # a different resource. In practice Azure naming rules forbid
        # these characters AND the values come from the trusted
        # autograny detail blob, but the cost of escaping is zero and
        # the contract holds even if a future caller passes
        # untrusted input.
        url = (
            f"{_ARM_BASE}/subscriptions/{_q(subscription_id, safe='')}"
            f"/resourceGroups/{_q(resource_group, safe='')}/providers/"
            f"Microsoft.CognitiveServices/accounts/{_q(account_name, safe='')}"
            f"?api-version=2023-05-01"
        )
        async with self._http() as client:
            resp = await client.delete(
                url, headers={"Authorization": f"Bearer {token}"},
            )
        if resp.status_code in (200, 202, 204):
            return {
                "deleted": True,
                "account_name": account_name,
                "resource_group": resource_group,
                "subscription_id": subscription_id,
            }
        if resp.status_code == 404:
            return {
                "deleted": False,
                "already_absent": True,
                "account_name": account_name,
                "resource_group": resource_group,
                "subscription_id": subscription_id,
            }
        raise RuntimeError(
            f"delete_azure_openai_account {resp.status_code}: "
            f"{_safe_body(resp, max_len=400)}",
        )

    async def provision_foundry_project(
        self,
        *,
        subscription_id: str,
        resource_group: str,
        account_name: str,
        project_name: str,
        location: str,
        display_name: str,
        description: str = "",
        tags: dict[str, str] | None = None,
        poll_seconds: int = 180,
    ) -> dict[str, Any]:
        """Create a Foundry project under an AIServices account.

        The project is the unit of "what an end user actually uses"
        in the Foundry portal — it gets its own inference endpoint
        (``https://{accountSubdomain}.services.ai.azure.com/api/
        projects/{projectName}``) and its own RBAC surface. The keys
        live on the parent account, not the project — see
        :meth:`list_foundry_account_keys`.

        ``display_name`` is what shows up in the Foundry portal's
        project picker; ``project_name`` is the URL-safe ARM name.
        Both are required (the portal forces displayName, even when
        equal to the resource name).

        ``poll_seconds`` defaults to 180 — projects provision faster
        than parent accounts (typically 10-30s) because no DNS
        record is created.
        """
        token = await self._token(_ARM_BASE)
        token_acquired_at = 0
        base_url = (
            f"{_ARM_BASE}/subscriptions/{subscription_id}"
            f"/resourceGroups/{resource_group}/providers/"
            f"Microsoft.CognitiveServices/accounts/{account_name}"
            f"/projects/{project_name}"
            f"?api-version={self._FOUNDRY_API_VERSION}"
        )
        body: dict[str, Any] = {
            "location": location,
            "identity": {"type": "SystemAssigned"},
            "properties": {
                "displayName": display_name,
                "description": description,
            },
        }
        if tags:
            body["tags"] = tags
        async with self._http() as client:
            create_resp = await client.put(
                base_url, json=body,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
            )
            if create_resp.status_code not in (200, 201):
                raise RuntimeError(
                    f"create_foundry_project {create_resp.status_code}: "
                    f"{_safe_body(create_resp, max_len=400)}",
                )
            deadline = poll_seconds
            elapsed = 0
            current = create_resp.json()
            while True:
                state = (
                    (current.get("properties") or {}).get("provisioningState")
                    or ""
                )
                if state == "Succeeded":
                    return current
                if state in ("Failed", "Canceled"):
                    raise RuntimeError(
                        f"provision_foundry_project ended in {state}: "
                        f"{str(current)[:400]}",
                    )
                if elapsed >= deadline:
                    raise RuntimeError(
                        f"provision_foundry_project timeout after "
                        f"{deadline}s; last state={state!r}",
                    )
                await asyncio.sleep(5)
                elapsed += 5
                if elapsed - token_acquired_at >= 1800:
                    token = await self._token(_ARM_BASE)
                    token_acquired_at = elapsed
                get_resp = await client.get(
                    base_url,
                    headers={"Authorization": f"Bearer {token}"},
                )
                if get_resp.status_code != 200:
                    raise RuntimeError(
                        f"poll_foundry_project {get_resp.status_code}: "
                        f"{_safe_body(get_resp, max_len=400)}",
                    )
                current = get_resp.json()

    async def list_foundry_account_keys(
        self,
        *,
        subscription_id: str,
        resource_group: str,
        account_name: str,
    ) -> dict[str, str]:
        """Return ``{"key1": ..., "key2": ...}`` for a Foundry
        AIServices account.

        Foundry projects share keys with the parent account — the
        project name is part of the inference URL, not the auth
        material. Both keys rotate independently; Helix stores
        ``key1`` and the assignee rotates via the Azure portal if
        it leaks. Auth header is ``api-key`` (Cognitive Services
        convention) for direct REST calls and
        ``Authorization: Bearer`` only after exchanging the key
        for a token via the AOAI auth endpoint.
        """
        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{subscription_id}"
            f"/resourceGroups/{resource_group}/providers/"
            f"Microsoft.CognitiveServices/accounts/{account_name}"
            f"/listKeys?api-version={self._FOUNDRY_API_VERSION}"
        )
        async with self._http() as client:
            resp = await client.post(
                url,
                headers={"Authorization": f"Bearer {token}"},
            )
        if resp.status_code != 200:
            raise RuntimeError(
                f"list_foundry_account_keys {resp.status_code}: "
                f"{_safe_body(resp, max_len=400)}",
            )
        data = resp.json()
        return {
            "key1": str(data.get("key1") or ""),
            "key2": str(data.get("key2") or ""),
        }

    async def regenerate_foundry_account_key(
        self,
        *,
        subscription_id: str,
        resource_group: str,
        account_name: str,
        key_name: str,
    ) -> dict[str, str]:
        """Regenerate one of a Foundry AIServices account's two keys.

        ``key_name`` is ``"Key1"`` or ``"Key2"`` (ARM's casing). ARM
        ``POST .../accounts/{name}/regenerateKey`` with body
        ``{"keyName": ...}`` mints a NEW value for the named slot and
        returns BOTH current keys (``{"key1", "key2"}``); the OTHER slot
        is untouched.

        This is the primitive behind Azure OpenAI dual-active rotation:
        the account natively carries two keys that rotate independently,
        so we hand out slot A, later regenerate slot B (both valid =
        overlap), swap the employee to slot B, then regenerate slot A to
        kill the old credential. See ``apikey/rotation.py``.

        Note the case-normalisation: Azure accepts ``Key1``/``Key2`` in
        the request body but returns the keys under lowercase
        ``key1``/``key2`` in the response — mirrors
        :meth:`list_foundry_account_keys`.
        """
        from urllib.parse import quote as _q

        normalized = (key_name or "").strip().lower()
        if normalized not in ("key1", "key2"):
            raise ValueError(
                f"key_name must be 'Key1' or 'Key2', got {key_name!r}",
            )
        arm_key_name = "Key1" if normalized == "key1" else "Key2"
        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{_q(subscription_id, safe='')}"
            f"/resourceGroups/{_q(resource_group, safe='')}/providers/"
            f"Microsoft.CognitiveServices/accounts/{_q(account_name, safe='')}"
            f"/regenerateKey?api-version={self._FOUNDRY_API_VERSION}"
        )
        async with self._http() as client:
            resp = await client.post(
                url,
                json={"keyName": arm_key_name},
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
            )
        if resp.status_code != 200:
            raise RuntimeError(
                f"regenerate_foundry_account_key {resp.status_code}: "
                f"{_safe_body(resp, max_len=400)}",
            )
        data = resp.json()
        return {
            "key1": str(data.get("key1") or ""),
            "key2": str(data.get("key2") or ""),
        }

    async def list_foundry_deployments(
        self,
        *,
        subscription_id: str,
        resource_group: str,
        account_name: str,
    ) -> list[dict[str, str]]:
        """List the model deployments on a Foundry / AOAI account.

        Model deployments are what the ``AZURE_OPENAI_DEPLOYMENT_NAME`` /
        OpenAI-SDK ``model=`` argument references — they are created by IT
        in the Azure portal (or via ARM), NOT by the employee. This
        read-only ARM ``GET .../accounts/{name}/deployments`` surfaces
        the deployment *names* + their backing model so the employee's
        "how to use this key" panel can show the real deployment to call
        instead of telling them to go deploy one themselves.

        Returns an ordered list of ``{"name", "model", "version",
        "sku", "state"}`` (only ``succeeded`` deployments are usable, but
        we return all states so the caller can surface an in-progress
        deployment too). Empty list = the account has no deployments yet
        → the caller should route the employee to IT.
        """
        from urllib.parse import quote as _q

        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{_q(subscription_id, safe='')}"
            f"/resourceGroups/{_q(resource_group, safe='')}/providers/"
            f"Microsoft.CognitiveServices/accounts/"
            f"{_q(account_name, safe='')}/deployments"
            f"?api-version={self._FOUNDRY_API_VERSION}"
        )
        async with self._http() as client:
            resp = await client.get(
                url,
                headers={"Authorization": f"Bearer {token}"},
            )
        if resp.status_code != 200:
            raise RuntimeError(
                f"list_foundry_deployments {resp.status_code}: "
                f"{_safe_body(resp, max_len=400)}",
            )
        out: list[dict[str, str]] = []
        for item in resp.json().get("value", []) or []:
            props = item.get("properties") or {}
            model = props.get("model") or {}
            sku = item.get("sku") or {}
            out.append({
                "name": str(item.get("name") or ""),
                "model": str(model.get("name") or ""),
                "version": str(model.get("version") or ""),
                "sku": str(sku.get("name") or ""),
                "state": str(props.get("provisioningState") or ""),
            })
        return out

    # ── Billing (Cost Management Query) ──────────────────────────

    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]:
        # ``bq_billing`` is AWS-driver-specific (CUR-in-BigQuery for
        # the prod main Org). Accept for signature parity with the
        # base class + service-layer fanout, then ignore.
        del bq_billing
        if scope is None:
            return []  # caller iterates; Azure cost is subscription-scoped
        # Azure Cost Management's POST query endpoint takes a
        # subscription GUID in the URL path. Sub-tenant rows (e.g.
        # ``azure_resource_group`` minted by the Azure OpenAI auto-
        # grant flow) carry an RG name in ``external_id`` and would
        # produce an ``InvalidSubscriptionId`` 400. The cost-sync
        # crons filter to ``TENANT_LEVEL_SCOPE_TYPES`` so they don't
        # call us here — but raise ``NotImplementedError`` for any
        # future caller that forgets, matching the base contract's
        # "unsupported scope" path.
        _require_subscription_scope(scope, "cost")
        # 成本口径改造 PR6:Exports 白名单路径(opt-in)。默认继续走
        # Cost Management(折后实付,list 恒 NULL)。
        if (self.config.get("billing_source") or "cost_management") == "exports":
            return await self._list_billing_daily_exports(
                start_date, end_date, scope=scope,
            )
        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{scope.external_id}/providers/"
            "Microsoft.CostManagement/query?api-version=2023-11-01"
        )
        body = {
            "type": "ActualCost",
            "timeframe": "Custom",
            "timePeriod": {
                "from": start_date.isoformat() + "T00:00:00Z",
                # ``to`` is INCLUSIVE (Cost Management, timeframe=Custom)
                # — the verified statement lives in
                # ``providers/azure.py::get_monthly_billing``, which passes
                # the month's LAST day for exactly that reason. This driver
                # used to send ``end_date + 1`` at T00:00:00Z with no
                # explanation, so under the same semantics it pulled one
                # extra, still-incomplete calendar day and upserted half of
                # "today" into the daily table (#2154). Harmless in practice
                # (the next run overwrites it, and most consumers filter
                # ``<= end``) — but the two halves of the repo cannot both
                # be right. Aligned on the documented one, pinned to the
                # last instant of ``end_date`` so the request is
                # unambiguous whichever way Azure reads the boundary.
                "to": end_date.isoformat() + "T23:59:59Z",
            },
            "dataset": {
                "granularity": "Daily",
                "aggregation": {
                    "totalCost": {"name": "Cost", "function": "Sum"},
                },
                "grouping": [{"type": "Dimension", "name": "ServiceName"}],
            },
        }
        # Cost Management Query paginates via ``properties.nextLink`` —
        # subsequent pages are POSTed to the returned URL with the
        # SAME body, not to the original endpoint. A single-subscription
        # × short-window × group-by-service call usually fits in one
        # page, but larger windows or many services spill and silently
        # truncate without this loop.
        rows: list[list[Any]] = []
        cols: list[str] = []
        next_url: str | None = url
        # Use ``_post_with_retry`` for combined transport + 429/5xx retry
        # (Azure Cost Management's per-tenant + per-scope rate limiter +
        # transport-level flakes that bit ``list_cost_facts`` on
        # 7bc8ab17 — same backend, same failure modes). Pre-PR this
        # method had NO retry at all; cron just hadn't exercised
        # ``list_billing_daily``'s transport flake yet.
        async with self._http() as client:
            while next_url:
                resp = await _post_with_retry(
                    client, next_url,
                    json=body,
                    headers={"Authorization": f"Bearer {token}"},
                    endpoint_label=(
                        f"list_billing_daily sub={scope.external_id}"
                    ),
                    parse_retry_after=_parse_azure_cost_retry_after,
                    pacer=self._cm_pacer,
                    throttle_budget_s=_CM_THROTTLE_BUDGET_SEC,
                )
                if resp.status_code != 200:
                    # Raise instead of returning ``[]`` — silently
                    # substituting "no spend this period" on 401/403/429/
                    # 5xx lets ``last_seen_billing_daily_at`` advance and
                    # spike-detection then runs on stale 7-day averages
                    # that don't include the missing days. ``list_cost_facts``
                    # already raises on the same condition; daily was the
                    # holdout (audited 2026-05-10). The lifecycle service
                    # catches and records the per-scope error so other
                    # subscriptions still sync.
                    raise RuntimeError(
                        f"Azure cost query sub={scope.external_id} "
                        f"status={resp.status_code} "
                        f"body={_safe_body(resp, max_len=300)!r}",
                    )
                data = resp.json().get("properties") or {}
                if not cols:
                    cols = [c["name"] for c in data.get("columns") or []]
                rows.extend(data.get("rows") or [])
                next_url = data.get("nextLink") or None
        if not cols:
            return []
        try:
            ci = (
                cols.index("PreTaxCost") if "PreTaxCost" in cols
                else cols.index("Cost")
            )
            di = cols.index("UsageDate")
            si = cols.index("ServiceName")
            cur_i = cols.index("Currency") if "Currency" in cols else None
        except ValueError:
            return []
        by_day: dict[date, dict[str, Any]] = {}
        for row in rows:
            d_raw = str(row[di])
            # Azure returns UsageDate as ``20260422`` int-like OR an ISO
            # date — normalize both.
            if len(d_raw) == 8 and d_raw.isdigit():
                d = date(int(d_raw[:4]), int(d_raw[4:6]), int(d_raw[6:8]))
            else:
                try:
                    d = date.fromisoformat(d_raw[:10])
                except ValueError:
                    continue
            cost = float(row[ci] or 0)
            svc = str(row[si] or "unknown")
            cur = str(row[cur_i]) if cur_i is not None else "USD"
            entry = by_day.setdefault(
                d, {"total": 0.0, "services": {}, "currency": cur},
            )
            entry["total"] += cost
            entry["services"][svc] = entry["services"].get(svc, 0.0) + cost
        return [
            CloudDailyCost(
                date=d,
                scope=scope,
                total_cost_usd=entry["total"],
                currency=entry["currency"],
                cost_by_service=entry["services"],
            )
            for d, entry in sorted(by_day.items())
        ]

    # ── Cost Management custom-dimension query ───────────────────
    #
    # Cost Management's POST query body accepts an arbitrary
    # ``grouping`` array (no hard limit like AWS's 2-dimension cap)
    # and supports both Dimension and Tag groupings. The Helix API
    # surfaces canonical lowercase aliases:
    #
    #     subscription / service / resource_group / resource_type /
    #     resource_location / meter_category / meter_subcategory /
    #     resource / tag:KEY
    #
    # mapped to Cost Management's PascalCase native names below.
    _CM_DIMENSION_ALIASES: dict[str, str] = {
        "subscription": "SubscriptionId",
        "service": "ServiceName",
        "resource_group": "ResourceGroup",
        "resource_type": "ResourceType",
        "resource_location": "ResourceLocation",
        "region": "ResourceLocation",     # convenience alias
        "meter_category": "MeterCategory",
        "meter_subcategory": "MeterSubCategory",
        "meter": "Meter",
        "resource": "ResourceId",
    }

    @classmethod
    def supported_billing_dimensions(cls) -> frozenset[str]:
        return frozenset({*cls._CM_DIMENSION_ALIASES.keys(), "tag"})

    def _cm_grouping_spec(self, dim: str) -> dict[str, str]:
        if dim.startswith("tag:"):
            tag_key = dim.split(":", 1)[1].strip()
            if not tag_key:
                raise ValueError(
                    "tag dimension requires a key (e.g. 'tag:Project')",
                )
            return {"type": "Tag", "name": tag_key}
        cm = self._CM_DIMENSION_ALIASES.get(dim.lower())
        if cm is None:
            raise ValueError(
                f"Azure does not support dimension {dim!r} "
                f"(supported: {sorted(self._CM_DIMENSION_ALIASES)} + tag:KEY)",
            )
        return {"type": "Dimension", "name": cm}

    def _cm_filter_expr(
        self, filters: dict[str, list[str]],
    ) -> dict[str, Any] | None:
        """Build a ``filter`` clause AND-ing each dimension's allow
        list. Cost Management's filter shape is recursive (And/Or/Not
        + leaf dimensions / tags); we keep things flat — top-level
        And of leaf clauses is enough for the tag/subscription
        slicing the API surfaces.
        """
        clauses: list[dict[str, Any]] = []
        for dim, values in (filters or {}).items():
            if not values:
                continue
            if dim.startswith("tag:"):
                tag_key = dim.split(":", 1)[1].strip()
                if tag_key:
                    clauses.append({"tags": {
                        "name": tag_key,
                        "operator": "In",
                        "values": list(values),
                    }})
                continue
            cm = self._CM_DIMENSION_ALIASES.get(dim.lower())
            if cm is None:
                raise ValueError(
                    f"Azure filter dimension {dim!r} not supported",
                )
            clauses.append({"dimensions": {
                "name": cm,
                "operator": "In",
                "values": list(values),
            }})
        if not clauses:
            return None
        if len(clauses) == 1:
            return clauses[0]
        return {"and": clauses}

    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]:
        del bq_billing  # AWS-driver-specific; accept for signature parity
        if not group_by:
            raise ValueError("group_by must contain at least one dimension")
        # Azure cost is naturally subscription-scoped: the query URL
        # carries the subscription id. Caller should pass scope per
        # subscription. ``subscription`` in group_by is technically
        # redundant when scope is set but harmless — we keep it so
        # the cross-subscription tenant-wide path (scope=None) can
        # still emit a subscription column.
        if scope is None:
            return []
        # See ``list_billing_daily`` — Azure CM Query needs a
        # subscription GUID in the URL; sub-tenant scope types (RG)
        # are rejected here so a 400 from Azure doesn't masquerade as
        # an "unsupported dimension" error.
        _require_subscription_scope(scope, "cost")
        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{scope.external_id}/providers/"
            "Microsoft.CostManagement/query?api-version=2023-11-01"
        )
        groupings = [self._cm_grouping_spec(d) for d in group_by]
        body: dict[str, Any] = {
            "type": "ActualCost",
            "timeframe": "Custom",
            "timePeriod": {
                "from": start_date.isoformat() + "T00:00:00Z",
                # See #2154: Cost Management's ``to`` is inclusive.
                "to": end_date.isoformat() + "T23:59:59Z",
            },
            "dataset": {
                "granularity": "Daily",
                "aggregation": {
                    "totalCost": {"name": "Cost", "function": "Sum"},
                },
                "grouping": groupings,
            },
        }
        cm_filter = self._cm_filter_expr(filters or {})
        if cm_filter:
            body["dataset"]["filter"] = cm_filter

        rows: list[list[Any]] = []
        cols: list[str] = []
        next_url: str | None = url
        # Same combined transport + 429/5xx retry as ``list_billing_daily``
        # above — same endpoint, same failure modes. This path is admin /
        # dashboard driven (not cron) so it doesn't fire daily, but a
        # transport flake here still surfaces as "dashboard charts empty"
        # rather than the helpful "Azure transport error" the failure-
        # notifier can dedup. Audit 2026-05-23.
        async with self._http() as client:
            while next_url:
                resp = await _post_with_retry(
                    client, next_url,
                    json=body,
                    headers={"Authorization": f"Bearer {token}"},
                    endpoint_label=(
                        f"list_billing_by_dimension sub={scope.external_id}"
                    ),
                    parse_retry_after=_parse_azure_cost_retry_after,
                    pacer=self._cm_pacer,
                    throttle_budget_s=_CM_THROTTLE_BUDGET_SEC,
                )
                if resp.status_code != 200:
                    # Symmetric with ``list_billing_daily`` /
                    # ``list_cost_facts`` (PR #234 fixed those to raise).
                    # Pre-fix this path returned [] on 401/403/429/5xx,
                    # silently masking a real Azure outage as "no spend
                    # on this dimension" — the dashboard / MCP caller
                    # then reported zeros and operators couldn't tell
                    # the difference between a healthy zero and an
                    # auth break. Raise so the caller surfaces it.
                    # Sanitise the response body before including in
                    # the error: (1) redact credentials — a misconfigured
                    # proxy WAF on a 401 page can echo Authorization tokens
                    # back in the body, which would land in
                    # ``scheduled_job_runs.errors`` and the failure-notifier
                    # Feishu payload; (2) strip control chars so the audit
                    # log stays single-line and readable.
                    body_redacted = redact_provider_error(resp.text or "", max_len=200)
                    body_safe = "".join(
                        c if c.isprintable() and c not in ("\t", "\r", "\n")
                        else " "
                        for c in body_redacted
                    ).strip()
                    raise RuntimeError(
                        f"Azure cost query (by-dimension) "
                        f"sub={scope.external_id} status={resp.status_code} "
                        f"body={body_safe}",
                    )
                data = resp.json().get("properties") or {}
                if not cols:
                    cols = [c["name"] for c in data.get("columns") or []]
                rows.extend(data.get("rows") or [])
                next_url = data.get("nextLink") or None
        if not cols or not rows:
            return []
        try:
            ci = (
                cols.index("PreTaxCost") if "PreTaxCost" in cols
                else cols.index("Cost")
            )
            di = cols.index("UsageDate")
            cur_i = cols.index("Currency") if "Currency" in cols else None
        except ValueError:
            return []
        # The grouping columns appear after the cost+date columns.
        # Map Helix-canonical aliases to the column index Cost
        # Management actually returned (PascalCase native names).
        col_idx_for_alias: dict[str, int] = {}
        missing_aliases: list[str] = []
        for alias, gspec in zip(group_by, groupings, strict=False):
            target_name = gspec["name"]
            try:
                col_idx_for_alias[alias] = cols.index(target_name)
            except ValueError:
                # Tag groupings come back under the tag key directly
                # (e.g. ``Project`` not ``Tag``). Try the bare key.
                try:
                    col_idx_for_alias[alias] = cols.index(
                        target_name.lower(),
                    )
                except ValueError:
                    missing_aliases.append(alias)
        if missing_aliases:
            # Pre-fix this branch silently dropped the missing alias
            # from the per-row output schema — the caller asked for
            # ``["service", "tag:Project"]`` and got rows missing the
            # tag key. The MCP tool / dashboard treated the absence as
            # "row has no Project tag" leading to silent allocation
            # drift. Raise so the caller knows the request shape can't
            # be honoured (mis-typed tag, removed tag, or Cost
            # Management not yet exposing the dimension on this
            # subscription).
            raise RuntimeError(
                f"Azure cost query (by-dimension) "
                f"sub={scope.external_id} response missing requested "
                f"columns {missing_aliases!r} — got cols={cols!r}",
            )
        out: list[BillingByDimensionRow] = []
        for row in rows:
            d_raw = str(row[di])
            if len(d_raw) == 8 and d_raw.isdigit():
                d = date(int(d_raw[:4]), int(d_raw[4:6]), int(d_raw[6:8]))
            else:
                try:
                    d = date.fromisoformat(d_raw[:10])
                except ValueError:
                    continue
            cost = float(row[ci] or 0)
            currency = str(row[cur_i]) if cur_i is not None else "USD"
            dims_out: dict[str, str] = {}
            for alias, idx in col_idx_for_alias.items():
                v = row[idx] if idx < len(row) else None
                dims_out[alias] = str(v) if v is not None else ""
            out.append(BillingByDimensionRow(
                date=d,
                dimensions=dims_out,
                cost=cost,
                currency=currency,
                raw={"row": row, "cols": cols},
            ))
        return out

    async def _fetch_retail_prices_usd(
        self, meter_ids: set[str],
    ) -> dict[str, float]:
        """Resolve PAYG (list / 原价) USD unit prices for ``meter_ids``.

        Hits the public, unauthenticated Azure Retail Prices API and
        returns ``{meterId: retailPrice_usd_per_unit}`` for the base
        (``tierMinimumUnits == 0``) ``Consumption`` price of each meter.
        Meters with no Consumption price (Marketplace / support /
        reservation-only / refund meters) are simply absent from the
        map — the caller leaves their gross NULL (honest coverage).

        Best-effort by design: ANY failure (transport, non-200, malformed
        body) degrades to "no gross for this run" — it MUST NOT break net
        cost ingestion. Returns whatever was resolved before a mid-flight
        error so a single bad page doesn't discard healthy batches.
        """
        # Only GUID-shaped meter ids reach the OData ``$filter`` — guards
        # against filter injection AND skips the synthetic ids the cost
        # path mints for resource-less rows (``azure-sub:...``), which
        # are not real meters and would never match the catalogue.
        ids = sorted(m for m in meter_ids if m and _OID_GUID_RE.match(m))
        out: dict[str, float] = {}
        if not ids:
            return out
        try:
            async with self._http() as client:
                for i in range(0, len(ids), _RETAIL_PRICE_BATCH):
                    batch = ids[i:i + _RETAIL_PRICE_BATCH]
                    clause = " or ".join(
                        f"meterId eq '{m}'" for m in batch
                    )
                    params: dict[str, str] = {
                        "api-version": _RETAIL_PRICES_API_VERSION,
                        "currencyCode": "USD",
                        "$filter": f"priceType eq 'Consumption' and ({clause})",
                    }
                    # First page via params; subsequent pages via the
                    # server-supplied (already-encoded) NextPageLink.
                    url: str | None = _RETAIL_PRICES_URL
                    use_params: dict[str, str] | None = params
                    while url:
                        resp = await client.get(
                            url,
                            params=use_params if use_params else None,
                        )
                        if resp.status_code != 200:
                            logger.warning(
                                "Azure retail-prices lookup HTTP %s "
                                "(gross stays NULL for this batch): %s",
                                resp.status_code,
                                _safe_body(resp, max_len=200),
                            )
                            break
                        data = resp.json()
                        for item in data.get("Items") or []:
                            mid = str(item.get("meterId") or "")
                            if not mid:
                                continue
                            # Base PAYG tier only; higher tiers are
                            # volume-discount steps that don't represent
                            # the headline list price.
                            try:
                                tier = float(item.get("tierMinimumUnits") or 0)
                            except (TypeError, ValueError):
                                continue
                            if tier != 0:
                                continue
                            price = item.get("retailPrice")
                            if price is None:
                                continue
                            try:
                                price_f = float(price)
                            except (TypeError, ValueError):
                                continue
                            # First Consumption/tier-0 hit wins; a meter
                            # should resolve to exactly one such row.
                            out.setdefault(mid, price_f)
                        url = data.get("NextPageLink") or None
                        use_params = None
        except Exception as exc:  # noqa: BLE001 — never break net ingest
            logger.warning(
                "Azure retail-prices lookup failed (gross stays NULL): %s",
                format_exception_for_log(exc, max_len=200),
            )
        return out

    # ── Line-item cost facts (production-grade FinOps ingestion) ────
    #
    # Cost Management Query supports up to 20 ``grouping`` entries in
    # one POST, so we can ask for the full multi-dim shape (resource
    # id + service + meter + RG + region + resource type) in a single
    # round-trip per subscription per window. Tags are a separate
    # ``Tag`` grouping that the API rejects mixing with Dimension
    # groupings beyond a low cap; we therefore make a second pass per
    # tag key the integration is configured to surface (config
    # ``cost_tag_keys: ["Owner", "Project"]``) and merge by ResourceId.
    #
    # Window cap: Cost Management hard-limits ``Custom`` timeframe to
    # 365 days and degrades response size aggressively past 90. The
    # caller (cron / backfill) is expected to chunk; we don't second-
    # guess here.

    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]:
        # ``principal_tag_keys`` is AWS-specific (IAM principal tag
        # cost-allocation). Azure Cost Management surfaces resource
        # / tag dimensions but no IAM-principal dimension; accept
        # for signature uniformity and ignore.
        # ``bq_billing`` likewise AWS-specific (CUR-in-BigQuery routing).
        del principal_tag_keys, bq_billing
        # Azure: the driver queries Cost Management's ``CostUSD``
        # column which Microsoft populates from their own FX
        # (independent of Helix's CNY/USD oracle); only when both
        # ``CostUSD`` is absent AND the row currency is non-USD does
        # the driver flag ``amount_usd=0`` (see below). Either way,
        # the Helix-resolved CNY→USD rate is irrelevant. We accept
        # the kwarg to match the base-class signature so the cost-
        # ingestion service can pass it uniformly to every driver.
        del cny_per_usd
        if scope is None:
            # Tenant-wide ingestion fans out per subscription via the
            # cron handler. Returning empty here keeps the contract
            # honest: this driver requires a subscription pin.
            return []
        # See ``list_billing_daily`` — sub-tenant scope types (e.g.
        # ``azure_resource_group`` rows minted by the Azure OpenAI
        # auto-grant flow) carry RG names not subscription GUIDs and
        # would produce an ``InvalidSubscriptionId`` 400. RG-bound
        # spend is captured by the parent subscription's facts and
        # attributed to the right cost owner via the allocation rule
        # the auto-grant flow creates alongside the RG row.
        _require_subscription_scope(scope, "cost")

        # 成本口径改造 PR6:Exports 白名单路径(opt-in)。默认继续走
        # Cost Management(折后实付,list 恒 NULL)。
        if (self.config.get("billing_source") or "cost_management") == "exports":
            return await self._list_cost_facts_exports(
                start_date, end_date, scope=scope,
            )

        token = await self._token(_ARM_BASE)
        url = (
            f"{_ARM_BASE}/subscriptions/{scope.external_id}/providers/"
            "Microsoft.CostManagement/query?api-version=2023-11-01"
        )

        # 折扣前(原价)enrichment: ActualCost metrics are all
        # POST-discount (Cost / PreTaxCost / CostUSD = EA/MCA negotiated).
        # The query API has NO retail/list metric, so we ADD a ``MeterId``
        # grouping + ``UsageQuantity`` aggregation to the SAME query (no
        # extra round-trip on the heavily-throttled CM endpoint) and join
        # per-meter usage to the public Retail Prices catalogue below:
        # gross_usd = Σ(meter qty × meter PAYG unit price). The finer
        # MeterId grain is collapsed back to the canonical fact grain
        # BEFORE emitting, so net facts (and their fact_keys) are
        # byte-identical to the un-augmented path. Operator kill-switch:
        # set config ``cost_gross_pricing: false`` to disable.
        gross_enabled = bool(self.config.get("cost_gross_pricing", True))

        def _build_body(with_gross: bool) -> dict[str, Any]:
            aggregation: dict[str, Any] = {
                "totalCost": {"name": "Cost", "function": "Sum"},
                "totalCostUSD": {"name": "CostUSD", "function": "Sum"},
            }
            grouping: list[dict[str, str]] = [
                {"type": "Dimension", "name": "ResourceId"},
                {"type": "Dimension", "name": "ServiceName"},
                {"type": "Dimension", "name": "MeterCategory"},
                {"type": "Dimension", "name": "MeterSubCategory"},
                {"type": "Dimension", "name": "ResourceLocation"},
            ]
            if with_gross:
                aggregation["totalQuantity"] = {
                    "name": "UsageQuantity", "function": "Sum",
                }
                grouping.append({"type": "Dimension", "name": "MeterId"})
            return {
                "type": "ActualCost",
                "timeframe": "Custom",
                "timePeriod": {
                    "from": start_date.isoformat() + "T00:00:00Z",
                    # See the twin above (#2154): ``to`` is inclusive.
                    "to": end_date.isoformat() + "T23:59:59Z",
                },
                "dataset": {
                    "granularity": "Daily",
                    "aggregation": aggregation,
                    "grouping": grouping,
                },
            }

        body: dict[str, Any] = _build_body(gross_enabled)

        rows: list[list[Any]] = []
        cols: list[str] = []
        next_url: str | None = url
        # Azure Cost Management is one of the most aggressively throttled
        # ARM endpoints — the query API documents a per-tenant + per-scope
        # rate limiter that 429s freely even on sequential calls across
        # subscriptions in the same tenant. The server tells us EXACTLY
        # how long to back off via ``Retry-After`` (canonical) and the
        # entity-/clienttype-scoped ``x-ms-ratelimit-microsoft.costmanagement-*-retry-after``
        # headers. Without honouring those, fan-out across N
        # subscriptions trips throttle on sub 1 and every subsequent sub
        # fails too — exactly the prod failure mode logged at 03:35 /
        # 08:15 / 09:17 on 2026-05-11 (3 subs all 429ing in one run).
        #
        # ``_post_with_retry`` now ALSO covers transport-level errors
        # (httpx.ConnectError / ReadTimeout / RemoteProtocolError) —
        # the prod 2026-05-21 + 2026-05-22 incidents on subscription
        # 7bc8ab17-... ("PingCAP Azure OpenAI") were a TCP-level flake
        # the pre-PR retry loop did not catch (it only handled status
        # codes, never reached on transport exception). Same retry
        # budget (5 attempts) — a permanently transport-broken scope
        # surfaces as a per-scope error so the partial-failure
        # notifier fires once and the other 37 scopes complete.
        async with self._http() as client:
            while next_url:
                resp = await _post_with_retry(
                    client, next_url,
                    json=body,
                    headers={"Authorization": f"Bearer {token}"},
                    endpoint_label=(
                        f"list_cost_facts sub={scope.external_id}"
                    ),
                    parse_retry_after=_parse_azure_cost_retry_after,
                    pacer=self._cm_pacer,
                    throttle_budget_s=_CM_THROTTLE_BUDGET_SEC,
                )
                if (
                    resp.status_code == 400
                    and gross_enabled
                    and not rows
                ):
                    # The gross augmentation (MeterId grouping +
                    # UsageQuantity metric) may be rejected on some
                    # tenant agreements / api-versions. NEVER let
                    # best-effort gross break net ingestion: drop the
                    # augmentation and re-issue the base query. Only
                    # before any page is consumed (``not rows``) so a
                    # mid-pagination 400 still surfaces as a real error.
                    code, _msg = _parse_arm_error(resp)
                    logger.warning(
                        "Azure cost-facts gross augmentation rejected "
                        "sub=%s code=%s status=%s — falling back to "
                        "net-only query (gross stays NULL)",
                        scope.external_id, code or "?", resp.status_code,
                    )
                    gross_enabled = False
                    body = _build_body(False)
                    continue
                if resp.status_code != 200:
                    # Surface as a real error so the cron handler
                    # records the failure against this integration in
                    # ``scheduled_job_runs.errors``. Returning [] would
                    # be indistinguishable from "no spend this period"
                    # — masks 401/403/429/5xx and lets stale forecasts
                    # tick along while the operator stays in the
                    # dark. Same posture as the AWS / Aliyun /
                    # Tencent / GCP drivers in this PR.
                    # On 429 the body is Azure's content-free "Too many
                    # requests. Please retry." — it names neither which
                    # limiter fired nor how long it wanted. Carry the
                    # throttle headers into the message so the row that
                    # lands in ``scheduled_job_runs.errors`` is
                    # self-diagnosing (the 2026-09-04 one was not, and
                    # answering "did the server even tell us to wait?"
                    # took a container-log dig).
                    throttle_ctx = (
                        f" throttle_headers={_throttle_headers_diag(resp.headers)}"
                        if resp.status_code == 429 else ""
                    )
                    raise RuntimeError(
                        f"Azure Cost Management query failed "
                        f"sub={scope.external_id} "
                        f"status={resp.status_code}"
                        f"{throttle_ctx} "
                        f"body={_safe_body(resp, max_len=300)!r}",
                    )
                data = resp.json().get("properties") or {}
                if not cols:
                    cols = [c["name"] for c in data.get("columns") or []]
                rows.extend(data.get("rows") or [])
                next_url = data.get("nextLink") or None

        if not cols or not rows:
            return []

        # Resolve column indices once; tolerate either ``Cost`` or
        # ``PreTaxCost`` (cost type varies by tenant agreement).
        try:
            ci_native = (
                cols.index("PreTaxCost") if "PreTaxCost" in cols
                else cols.index("Cost")
            )
        except ValueError:
            return []
        ci_usd = cols.index("CostUSD") if "CostUSD" in cols else None
        ci_date = cols.index("UsageDate") if "UsageDate" in cols else None
        ci_currency = cols.index("Currency") if "Currency" in cols else None
        ci_resource = cols.index("ResourceId") if "ResourceId" in cols else None
        ci_service = cols.index("ServiceName") if "ServiceName" in cols else None
        ci_meter_cat = (
            cols.index("MeterCategory") if "MeterCategory" in cols else None
        )
        ci_meter_sub = (
            cols.index("MeterSubCategory") if "MeterSubCategory" in cols else None
        )
        ci_location = (
            cols.index("ResourceLocation") if "ResourceLocation" in cols else None
        )
        if ci_date is None:
            return []

        # 折扣前(原价)collapse: when the query was augmented (MeterId
        # grouping + UsageQuantity present), rows are split one-per-meter.
        # We (1) price each meter via the Retail Prices catalogue and
        # accumulate per-fact gross USD, then (2) collapse the meter-split
        # rows back to the canonical fact grain (summing net cost) so the
        # emit loop below — and every downstream fact_key — is identical
        # to the un-augmented path. Skipped entirely (and rows left as-is)
        # when the augmentation was rejected / disabled / absent.
        ci_meterid = cols.index("MeterId") if "MeterId" in cols else None
        ci_qty = (
            cols.index("UsageQuantity") if "UsageQuantity" in cols else None
        )
        gross_active = (
            gross_enabled and ci_meterid is not None and ci_qty is not None
        )
        # USD gross per canonical fact grain; ``None`` value means
        # "incomplete — at least one contributing meter had no list
        # price", which leaves that fact's gross NULL (honest coverage).
        gross_by_key: dict[tuple[Any, ...], float | None] = {}

        def _base_key(r: list[Any]) -> tuple[Any, ...]:
            # Raw dimension cells (pre-coercion) defining the fact grain,
            # MINUS MeterId. Computed identically here and in the emit
            # loop so the gross lookup lines up exactly.
            return (
                r[ci_date],
                r[ci_resource] if ci_resource is not None else None,
                r[ci_service] if ci_service is not None else None,
                r[ci_meter_cat] if ci_meter_cat is not None else None,
                r[ci_meter_sub] if ci_meter_sub is not None else None,
                r[ci_location] if ci_location is not None else None,
            )

        if gross_active:
            assert ci_meterid is not None and ci_qty is not None
            meter_ids = {
                str(r[ci_meterid]) for r in rows if r[ci_meterid]
            }
            prices = await self._fetch_retail_prices_usd(meter_ids)
            collapsed: dict[tuple[Any, ...], list[Any]] = {}
            for r in rows:
                key = _base_key(r)
                # Collapse net cost across the key's meter rows.
                stored = collapsed.get(key)
                if stored is None:
                    collapsed[key] = list(r)
                else:
                    stored[ci_native] = (
                        float(stored[ci_native] or 0) + float(r[ci_native] or 0)
                    )
                    if ci_usd is not None:
                        stored[ci_usd] = (
                            float(stored[ci_usd] or 0) + float(r[ci_usd] or 0)
                        )
                # Accumulate gross USD for the key.
                acc = gross_by_key.get(key, 0.0)
                if acc is None:
                    continue  # already marked incomplete
                mid = str(r[ci_meterid]) if r[ci_meterid] else None
                unit_price = prices.get(mid) if mid else None
                if unit_price is None:
                    gross_by_key[key] = None
                else:
                    gross_by_key[key] = acc + float(r[ci_qty] or 0) * unit_price
            rows = list(collapsed.values())

        out: list[CostFact] = []
        integration_id = self.integration_id
        unparseable_dates = 0
        for row in rows:
            d_raw = str(row[ci_date])
            if len(d_raw) == 8 and d_raw.isdigit():
                d = date(int(d_raw[:4]), int(d_raw[4:6]), int(d_raw[6:8]))
            else:
                try:
                    d = date.fromisoformat(d_raw[:10])
                except ValueError:
                    # Don't silently swallow — surface in logs so the
                    # operator notices the gap before finance does.
                    # Bound the per-call log volume so a malformed
                    # batch can't flood the worker logs.
                    if unparseable_dates < 5:
                        logger.warning(
                            "Azure cost-facts: dropping row with "
                            "unparseable UsageDate=%r sub=%s",
                            d_raw, scope.external_id,
                        )
                    unparseable_dates += 1
                    continue

            native = float(row[ci_native] or 0)
            currency = (
                str(row[ci_currency]) if ci_currency is not None else "USD"
            )
            # CostUSD is populated by Microsoft when the billing
            # currency differs from USD; falls back to native when
            # the tenant is already USD-billed.
            if ci_usd is not None and row[ci_usd] is not None:
                amount_usd = float(row[ci_usd] or 0)
            elif currency.upper() == "USD":
                amount_usd = native
            else:
                # No USD column AND non-USD billing — driver returns
                # the row but flags it; the ingestor will count it for
                # native-currency aggregation only.
                amount_usd = 0.0

            resource_id = (
                str(row[ci_resource]) if ci_resource is not None else None
            )
            service = (
                str(row[ci_service]) if ci_service is not None else "Other"
            ) or "Other"
            meter_cat = (
                str(row[ci_meter_cat]) if ci_meter_cat is not None else None
            )
            meter_sub = (
                str(row[ci_meter_sub]) if ci_meter_sub is not None else None
            )
            location = (
                str(row[ci_location]) if ci_location is not None else None
            )

            # Azure ResourceId carries the resource group as the 4th
            # segment: ``/subscriptions/{sub}/resourceGroups/{rg}/...``.
            # Resource type is ``providers/{ns}/{type}`` segments 6+7.
            rg, rtype, rname = _parse_azure_resource_id(resource_id)

            # Round-22 audit: Azure cost rows with ``resource_id IS
            # NULL`` are common — Marketplace fees, support charges,
            # tenant-wide discounts/refunds, EA-roll-ups. Without a
            # synthetic id, ``compute_fact_key((date, integration,
            # scope_external_id, None, usage_type))`` collapses every
            # NULL-resource row on the same (date, sub, usage_type) to
            # a single fact_key — the upsert then overwrites and
            # silently drops every additional Marketplace SKU on that
            # day. AWS BQ-path already synthesises ``aws-account:
            # {acc}:{product_code}`` for this exact case (line 3239);
            # parity here so monthly totals don't undercount when an
            # account has multiple Marketplace offers.
            if resource_id is None:
                resource_id = (
                    f"azure-sub:{scope.external_id}:{service}:"
                    f"{meter_cat or 'none'}:{meter_sub or 'none'}"
                )

            usage_type = meter_sub or meter_cat

            # 折扣前(原价): list-price USD from the meter-join above,
            # plus a hard honesty guard. List ≥ contracted always, so a
            # gross that came out BELOW net (qty/unit mismatch on an
            # exotic meter) is untrustworthy → drop to NULL rather than
            # surface a wrong, too-low gross. Also meaningless on
            # zero/negative net (credits, refunds) → NULL there too.
            gross_usd_value: float | None = (
                gross_by_key.get(_base_key(row)) if gross_active else None
            )
            if gross_usd_value is not None and (
                amount_usd <= 0 or gross_usd_value < amount_usd - 1e-6
            ):
                gross_usd_value = None
            # Native gross via the row's own effective FX so it stays
            # comparable to ``amount_native`` (USD-billed tenants — the
            # Azure Foundry case — collapse to the identity branch).
            if gross_usd_value is None:
                gross_native_value: float | None = None
            elif currency.upper() == "USD" or amount_usd <= 0:
                gross_native_value = gross_usd_value
            else:
                gross_native_value = gross_usd_value * (native / amount_usd)

            out.append(CostFact(
                date=d,
                integration_id=integration_id,
                service=service,
                amount_usd=amount_usd,
                amount_native=native,
                amount_usd_gross=gross_usd_value,
                amount_native_gross=gross_native_value,
                currency=currency,
                scope=scope,
                resource_group=rg,
                resource_id=resource_id,
                resource_name=rname,
                resource_type=rtype,
                region=location,
                usage_type=usage_type,
                tags={},
                raw={"row": row, "cols": cols},
            ))
        if unparseable_dates:
            logger.warning(
                "Azure cost-facts: %d row(s) dropped due to "
                "unparseable UsageDate sub=%s",
                unparseable_dates, scope.external_id,
            )
        return out

    # ── Cost Exports 数据面（billing_source="exports"，PR6）───────
    #
    # 规则/变换/聚合已就绪（模块顶 _classify_exports_bucket 一族）；
    # 唯一缺口是真实拉取器 —— Exports 未开通，storage account 读取
    # 方式待云管理定，_fetch_exports_rows 留显式 NotImplementedError。

    async def _fetch_exports_rows(
        self, start_date: date, end_date: date,
        *, scope: CloudScopeRef,
    ) -> list[dict[str, Any]]:
        """拉取窗口内的 Exports 行（snake_case 契约，见模块顶）。

        Exports 尚未开通:数据将落在 storage account 的 CSV/parquet,
        读取方式(SAS vs RBAC、目录布局、拉取节奏)等云管理开通导出
        后才能确定。届时在此实现:按 ``exports_storage_account`` /
        ``exports_container`` / ``exports_path_prefix`` 配置定位当期
        导出文件,逐行解析后过 :func:`_normalize_exports_row` 归一,
        直接喂给已就绪的 ``_list_cost_facts_exports`` /
        ``_list_billing_daily_exports`` 变换层。
        """
        raise NotImplementedError(
            "Azure Cost Exports data plane is not wired yet — the export "
            "lands in a storage account (CSV/parquet) and the access "
            "method is decided when cloud-mgmt enables the export. "
            "Whitelist rules / transforms are implemented; only this "
            "fetcher is pending. Keep billing_source unset (or "
            "'cost_management') until then.",
        )

    def _exports_fact_from_row(
        self, row: dict[str, Any], *, scope: CloudScopeRef,
    ) -> tuple[CostFact | None, bool]:
        """单行 → CostFact。返回 (fact, payg_missing_on_whitelist_row)。

        金额语义:
          * ``amount_usd``（发票口径）= cost_in_usd,缺列时 USD 计费
            回落 cost_in_billing_currency,非 USD 无换算则 0.0+native
            （driver_base 契约,ingestor 只入 native 聚合）。
          * ``amount_usd_list``（研发口径）= 白名单行的
            payg_cost_in_billing_currency;非 USD 计费用行内
            cost_in_usd/cost_in_billing_currency 纯汇率比换算（两列
            同额两币,比值即 FX,不引入折扣信息）;PAYG 缺失**绝不用
            实际成本兜底** → None + 上层计数告警。
          * ``amount_usd_public`` 恒 None —— PAYG 价即近似牌价,是否
            拆独立 public 度量待云管理评估（见成本口径说明页）。
        """
        d_raw = _exports_str(row, "date")
        d: date | None = None
        if len(d_raw) == 8 and d_raw.isdigit():
            d = date(int(d_raw[:4]), int(d_raw[4:6]), int(d_raw[6:8]))
        else:
            try:
                d = date.fromisoformat(d_raw[:10])
            except ValueError:
                # Exports 老 schema 用 MM/DD/YYYY。
                try:
                    m, dd, y = d_raw.split("/")
                    d = date(int(y), int(m), int(dd))
                except (ValueError, AttributeError):
                    d = None
        if d is None:
            return None, False

        def _num(key: str) -> float | None:
            value = row.get(key)
            if value is None or (isinstance(value, str) and not value.strip()):
                return None
            try:
                return float(value)
            except (TypeError, ValueError):
                return None

        native = _num("cost_in_billing_currency") or 0.0
        currency = (_exports_str(row, "billing_currency") or "USD").upper()
        usd_actual = _num("cost_in_usd")
        if usd_actual is not None:
            amount_usd = usd_actual
        elif currency == "USD":
            amount_usd = native
        else:
            amount_usd = 0.0  # 无换算不伪造;ingestor 只入 native 聚合

        bucket = _classify_exports_bucket(row)
        in_whitelist = bucket == COST_BUCKET_USAGE
        payg_native = _exports_payg_amount(row) if in_whitelist else None
        payg_missing = in_whitelist and payg_native is None
        usd_list: float | None = None
        if payg_native is not None:
            if currency == "USD":
                usd_list = payg_native
            elif usd_actual is not None and native:
                # 行内两列同额两币 → 比值是纯 FX,无折扣泄露。
                usd_list = payg_native * (usd_actual / native)
            # 非 USD 且无 cost_in_usd:换不了 → None(不伪造),
            # native list 仍落盘。

        resource_id = _exports_str(row, "resource_id") or None
        rg, rtype, rname = _parse_azure_resource_id(resource_id)
        rg = rg or (_exports_str(row, "resource_group") or None)
        service = (
            _exports_str(row, "meter_category")
            or _exports_str(row, "consumed_service")
            or "Other"
        )
        meter_sub = _exports_str(row, "meter_sub_category") or None
        if resource_id is None:
            # 与 Cost Management 路径同款合成 id(Marketplace/税费等无
            # 资源行不得在 fact_key 上互相覆盖)。
            resource_id = (
                f"azure-sub:{scope.external_id}:{service}:"
                f"{_exports_str(row, 'charge_type') or 'none'}:"
                f"{meter_sub or 'none'}"
            )

        return CostFact(
            date=d,
            integration_id=self.integration_id,
            service=service,
            amount_usd=amount_usd,
            amount_native=native,
            amount_usd_list=usd_list,
            amount_native_list=payg_native,
            amount_usd_public=None,
            cost_bucket=bucket,
            currency=currency,
            scope=scope,
            resource_group=rg,
            resource_id=resource_id,
            resource_name=rname,
            resource_type=rtype,
            region=_exports_str(row, "resource_location") or None,
            usage_type=meter_sub,
            usage_quantity=_num("quantity"),
            usage_unit=_exports_str(row, "unit_of_measure") or None,
            tags={},
            raw={"exports_row": row},
        ), payg_missing

    async def _list_cost_facts_exports(
        self, start_date: date, end_date: date,
        *, scope: CloudScopeRef,
    ) -> list[CostFact]:
        rows = await self._fetch_exports_rows(
            start_date, end_date, scope=scope,
        )
        out: list[CostFact] = []
        payg_missing = 0
        dropped_dates = 0
        for raw_row in rows:
            fact, missing = self._exports_fact_from_row(
                raw_row, scope=scope,
            )
            if fact is None:
                dropped_dates += 1
                continue
            if missing:
                payg_missing += 1
            out.append(fact)
        if payg_missing:
            # 云管理规则:PAYG 缺失行单独告警核查(绝不兜底)。
            logger.warning(
                "Azure exports cost-facts: %d whitelist row(s) missing "
                "payg_cost_in_billing_currency sub=%s window=%s..%s — "
                "list left NULL (never backfilled from actual cost); "
                "flag to cloud-mgmt for review",
                payg_missing, scope.external_id, start_date, end_date,
            )
        if dropped_dates:
            logger.warning(
                "Azure exports cost-facts: %d row(s) dropped due to "
                "unparseable date sub=%s",
                dropped_dates, scope.external_id,
            )
        return out

    async def _list_billing_daily_exports(
        self, start_date: date, end_date: date,
        *, scope: CloudScopeRef,
    ) -> list[CloudDailyCost]:
        facts = await self._list_cost_facts_exports(
            start_date, end_date, scope=scope,
        )
        by_day: dict[date, dict[str, Any]] = {}
        for fact in facts:
            # ``currency`` is "USD", NOT the billing currency (#2176).
            # Everything summed below is ``fact.amount_usd`` /
            # ``fact.amount_usd_list`` — already-converted USD — so
            # tagging the row with the subscription's billing currency
            # made ``CloudCostUsdConverter`` divide by the FX rate a
            # SECOND time on read (a CNY-billed subscription would read
            # ~7× low). The CM path next door stores native amounts with
            # a native label and is self-consistent; this path stores
            # USD, so it must say USD.
            entry = by_day.setdefault(fact.date, {
                "total": 0.0, "list": 0.0, "services": {},
                "currency": "USD",
                "payg_seen": False, "usage_missing_payg": False,
            })
            entry["total"] += fact.amount_usd
            entry["services"][fact.service] = (
                entry["services"].get(fact.service, 0.0) + fact.amount_usd
            )
            if fact.amount_usd_list is not None:
                entry["list"] += fact.amount_usd_list
                entry["payg_seen"] = True
            elif fact.cost_bucket == COST_BUCKET_USAGE:
                # 白名单行但无可用的 USD list 值(缺 PAYG 列,或非 USD
                # 计费且无行内 FX)——当日 list 口径不完整。
                entry["usage_missing_payg"] = True
        return [
            CloudDailyCost(
                date=d,
                scope=scope,
                total_cost_usd=entry["total"],
                # 白名单日合计,三种情形分开:
                #   * 当天有行但全出局(无白名单 usage)→ 0.0 是真实
                #     口径,非伪造;
                #   * 白名单行存在且至少一行带 PAYG → 合计(部分缺失
                #     由 cost-facts 路径 WARN 计数);
                #   * 白名单行存在但**没有任何一行**带 PAYG → None
                #     (口径未接入),绝不伪造 0 —— 0.0 会让读侧把
                #     「数据缺失」当成「当日白名单支出为零」。
                # public 待评估,None。
                total_cost_usd_list=(
                    None
                    if (
                        entry["usage_missing_payg"]
                        and not entry["payg_seen"]
                    )
                    else entry["list"]
                ),
                total_cost_usd_public=None,
                currency=entry["currency"],
                cost_by_service=entry["services"],
            )
            for d, entry in sorted(by_day.items())
        ]

    # ── Resource-creator lookup via Activity Log ─────────────────
    #
    # Azure Activity Log captures every ARM write operation
    # (``Microsoft.<Provider>/<resource-type>/write``) with the
    # ``caller`` field set to the AAD UPN that performed the
    # operation. Default retention is 90 days; Diagnostic Settings
    # can export to a Storage Account for longer retention.
    #
    # API: ``GET /subscriptions/{sub}/providers/microsoft.insights/
    # eventtypes/management/values?api-version=2015-04-01&$filter=...``
    # Filter syntax (Azure Monitor):
    #   ``eventTimestamp ge '<start>' and eventTimestamp le '<end>'``
    #   `` and resourceUri eq '<resource-id>'``
    #   `` and operationName eq '<provider>/<rt>/write'``
    #
    # We could narrow further by ``operationName`` but the resource
    # ID is already specific enough — every write to that ID lands
    # in the result set. Post-filter the EARLIEST event whose
    # operation name ends with ``/write`` (the create event is
    # always a write; updates also use write but we pick the
    # oldest, which IS the create for the standard ARM lifecycle).

    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" via Activity Log.

        90-day retention default. Returns the EARLIEST ``/write``
        event for the resource within the window.
        """
        import datetime as _dt

        if not resource_id:
            return None
        # Synthetic ids from the cost-facts ingest path don't have
        # an ARM resourceUri to look up.
        if resource_id.startswith((
            "azure-aggregated:", "azure-subscription:",
        )):
            return None
        if not resource_id.startswith("/subscriptions/"):
            # Activity Log queries require the ARM resourceUri,
            # which always starts with ``/subscriptions/``.
            return None

        # Extract subscription id from the resource_id —
        # ``/subscriptions/<sub-id>/...``
        parts = resource_id.split("/")
        if len(parts) < 3:
            return None
        subscription_id = parts[2]
        if not subscription_id:
            return None
        # Defence-in-depth: Azure subscription ids are UUIDs. Reject
        # anything else to keep the URL-path interpolation tight (the
        # subscription_id is interpolated into the request URL below).
        if not re.match(
            r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
            r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$",
            subscription_id,
        ):
            return None

        end_time = _dt.datetime.now(tz=_dt.UTC)
        start_time = end_time - _dt.timedelta(days=90)

        token = await self._token(_ARM_BASE)

        # Activity Log filter; date format must be ISO with Z. The
        # OData ``$filter`` syntax escapes single quotes by doubling
        # them (RFC 7-style). Azure ARM ids legitimately never
        # contain ``'`` but a malicious caller could pass one
        # through the path param; the doubled-quote escape blocks
        # the predicate-break.
        safe_resource_id = resource_id.replace("'", "''")
        filter_str = (
            f"eventTimestamp ge '{start_time.strftime('%Y-%m-%dT%H:%M:%SZ')}'"
            f" and eventTimestamp le '{end_time.strftime('%Y-%m-%dT%H:%M:%SZ')}'"
            f" and resourceUri eq '{safe_resource_id}'"
        )

        url = (
            f"{_ARM_BASE}/subscriptions/{subscription_id}/providers/"
            f"microsoft.insights/eventtypes/management/values"
        )
        params = {
            "api-version": "2015-04-01",
            "$filter": filter_str,
        }

        # We want the EARLIEST /write event = the original create.
        # Activity Log's default ordering is newest-first; the API
        # does NOT support ``$orderby``. For a heavily-mutated
        # resource (e.g. a VM with 200+ activity log entries in 90
        # days) the create event lands on the LAST page.
        #
        # Strategy: walk forward through pages but ALSO track if
        # we've reached the end of the result set (no ``nextLink``).
        # Cap at 50 pages (50 × ~50 events ≈ 2500 events / 90 days
        # — generous for "VM lifecycle" workloads; OS-image-update
        # event spam doesn't typically exceed this). When the cap
        # is hit we return the oldest event seen so far — biased
        # newer than the true create but still useful attribution,
        # and the cron sweep can re-attempt with ``force=True``.
        earliest_event: dict[str, Any] | None = None
        earliest_at: datetime | None = None
        pages_walked = 0
        _PAGE_CAP = 50

        async with self._http() as client:
            next_url: str | None = url
            current_params: dict[str, Any] | None = params
            while next_url and pages_walked < _PAGE_CAP:
                resp = await client.get(
                    next_url,
                    params=current_params,
                    headers={"Authorization": f"Bearer {token}"},
                )
                pages_walked += 1
                if resp.status_code == 400:
                    # Activity Log returns 400 on filter-too-wide
                    # (e.g. window exceeds retention).
                    body = resp.text
                    safe = _safe_body(resp, max_len=300)
                    if "OutsideAllowedRange" in body or "OutOfRange" in body:
                        raise ResourceCreatorWindowExceeded(
                            f"Azure Activity Log rejected window for "
                            f"{resource_id!r}: {safe}",
                        )
                    raise RuntimeError(
                        f"Azure Activity Log 400: {safe}",
                    )
                if resp.status_code != 200:
                    raise RuntimeError(
                        f"Azure Activity Log {resp.status_code}: "
                        f"{_safe_body(resp, max_len=300)}",
                    )
                data = resp.json()
                for ev in data.get("value") or []:
                    op_name = (
                        (ev.get("operationName") or {}).get("value") or ""
                    )
                    if not op_name.lower().endswith("/write"):
                        continue
                    ts_str = ev.get("eventTimestamp")
                    if not ts_str:
                        continue
                    try:
                        ts = _dt.datetime.fromisoformat(
                            ts_str.replace("Z", "+00:00"),
                        )
                    except ValueError:
                        continue
                    if earliest_at is None or ts < earliest_at:
                        earliest_event = ev
                        earliest_at = ts
                next_url = data.get("nextLink")
                current_params = None  # nextLink carries params

        if earliest_event is None:
            return None
        info = self._azure_event_to_creator_info(earliest_event, earliest_at)
        # SP-caller enrichment: when the static parse couldn't recover
        # an email (the caller was a Service Principal, not a user), try
        # Graph to resolve the SP's owner — that's the closest human
        # who can be held accountable for the resource. Falls through
        # silently on Graph permission denied / object not found; the
        # row still gets stamped with whatever the static parse
        # produced (display_name = SP's displayName when we can, or
        # raw GUID otherwise) so the operator at least sees something
        # actionable.
        if info.email is None and _OID_GUID_RE.match(info.principal_id):
            enriched = await self._enrich_sp_creator(info)
            if enriched is not None:
                info = enriched
        return info

    async def _enrich_sp_creator(
        self, info: ResourceCreatorInfo,
    ) -> ResourceCreatorInfo | None:
        """Best-effort SP→owner-user enrichment via Microsoft Graph.

        Pre-fix prod state (verified 2026-05-26 against the prod
        ``cmdb_resource_creators`` table): 5169 / 5193 resolved Azure
        rows had ``creator_email IS NULL`` because the
        :meth:`get_resource_creator` Activity-Log walk found a
        ``/write`` event whose ``caller`` was a Service Principal GUID
        (Terraform / AKS managed identities / GitHub Actions). The
        static parse has no way to map that GUID → a Helix Person.

        Graph closes the gap:
          1. ``GET /directoryObjects/{oid}`` resolves whichever kind
             the GUID is (user / SP / group). Cheaper than guessing
             the endpoint.
          2. If user → use its ``mail`` / ``userPrincipalName``.
          3. If SP → fetch ``/servicePrincipals/{oid}/owners``; take
             the first user owner's mail/UPN. Also enrich
             ``display_name`` with the SP's ``displayName`` so the row
             stops showing a raw GUID in the admin UI.
          4. On Graph 403 / 404 / non-user-non-SP / no owners → leave
             ``email`` None but at least populate ``display_name`` when
             we got the SP's name from step 1.

        Cached at process level by ``(integration_id, oid)`` — see
        :data:`_GRAPH_OID_CACHE`. None-results are also cached
        (terminal "can't resolve") to bound Graph QPS.
        """
        oid = info.principal_id
        cache_key = (self.integration_id, oid)
        hit, cached = _graph_oid_cache_get(cache_key)
        if hit:
            resolved = cached
        else:
            try:
                resolved = await self._graph_resolve_oid(oid)
            except Exception as exc:  # noqa: BLE001
                # Graph transport / auth failure on this one call.
                # Don't poison the entire ``get_resource_creator`` —
                # the static-parse info is still useful. Log once and
                # cache the failure so we don't retry every row.
                logger.info(
                    "Azure SP enrichment %s/%s graph lookup failed (%s) — "
                    "row keeps static parse",
                    self.integration_id, oid, exc,
                )
                resolved = None
            _graph_oid_cache_put(cache_key, resolved)

        if resolved is None:
            return None

        new_email = info.email
        new_display = info.display_name
        new_raw = dict(info.raw or {})

        if resolved.get("kind") == "user":
            user_email = resolved.get("email")
            if isinstance(user_email, str) and "@" in user_email:
                new_email = user_email.lower()
            user_display = resolved.get("display_name")
            if user_display:
                new_display = user_display
        elif resolved.get("kind") == "sp":
            sp_display = resolved.get("sp_display_name") or oid
            owner_email = resolved.get("owner_email")
            owner_display = resolved.get("owner_display_name")
            if isinstance(owner_email, str) and "@" in owner_email:
                new_email = owner_email.lower()
                # Render display_name as "owner via SP-name" so the
                # admin can see both who's responsible AND which SP
                # actually performed the write.
                new_display = (
                    f"{owner_display or owner_email} (via {sp_display})"
                    if owner_display else f"{owner_email} (via {sp_display})"
                )
            else:
                # No human owner — still upgrade display_name from
                # raw GUID to "<sp-name> (SP, owner unknown)" so the
                # row is actionable in the UI.
                new_display = f"{sp_display} (SP, owner unknown)"
            # Stash the Graph-discovered metadata in raw for ops
            # forensics.
            graph_meta = {
                k: resolved.get(k)
                for k in ("sp_display_name", "sp_app_id",
                          "owner_email", "owner_display_name",
                          "owner_count")
                if resolved.get(k) is not None
            }
            if graph_meta:
                new_raw["graph_sp_enrichment"] = graph_meta

        if (
            new_email == info.email
            and new_display == info.display_name
            and new_raw == (info.raw or {})
        ):
            return None

        return ResourceCreatorInfo(
            principal_id=info.principal_id,
            email=new_email,
            display_name=new_display,
            created_at_cloud=info.created_at_cloud,
            source=info.source,
            raw=new_raw,
        )

    async def _graph_resolve_oid(
        self, oid: str,
    ) -> dict[str, Any] | None:
        """Resolve an AAD object id via Microsoft Graph.

        Returns one of:
          * ``{"kind": "user", "email": str, "display_name": str}``
          * ``{"kind": "sp", "sp_display_name": str, "sp_app_id": str,
               "owner_email": str | None, "owner_display_name": str | None,
               "owner_count": int}``
          * ``None`` — Graph 403 / 404 / unsupported odata type / no
            owner found.

        Defence-in-depth: ``oid`` is interpolated into the URL path,
        so reject anything that isn't a canonical AAD GUID. The
        regex matches the same shape as the existing subscription-id
        guard.
        """
        if not _OID_GUID_RE.match(oid):
            return None
        try:
            token = await self._token("https://graph.microsoft.com")
        except Exception as exc:  # noqa: BLE001
            logger.info(
                "Azure SP enrichment: graph token fetch failed (%s)", exc,
            )
            return None
        headers = {"Authorization": f"Bearer {token}"}
        async with self._http() as client:
            try:
                obj_resp = await client.get(
                    f"{_GRAPH_BASE}/directoryObjects/{oid}",
                    headers=headers,
                )
            except (httpx.TransportError, OSError) as exc:
                logger.info(
                    "Azure SP enrichment %s: transport error on "
                    "directoryObjects (%s)", oid, exc,
                )
                return None
            if obj_resp.status_code in (401, 403):
                # Permission missing — Directory.Read.All not granted.
                # Operator-visible via the failure-notifier (cron
                # surfaces this via the per-row last_error trail when
                # we eventually capture it). Return None and cache so
                # we don't burn Graph QPS retrying.
                logger.info(
                    "Azure SP enrichment %s/%s: graph %d on "
                    "directoryObjects — Directory.Read.All needed",
                    self.integration_id, oid, obj_resp.status_code,
                )
                return None
            if obj_resp.status_code == 404:
                # Object deleted in AAD between the Activity-Log event
                # and our lookup. Terminal "can't resolve".
                return None
            if obj_resp.status_code != 200:
                logger.info(
                    "Azure SP enrichment %s/%s: graph %d on "
                    "directoryObjects — %s",
                    self.integration_id, oid, obj_resp.status_code,
                    _safe_body(obj_resp, max_len=200),
                )
                return None
            obj = obj_resp.json()
            odata_type = str(obj.get("@odata.type") or "").lower()
            if odata_type == "#microsoft.graph.user":
                mail = obj.get("mail") or obj.get("userPrincipalName")
                display = obj.get("displayName")
                if isinstance(mail, str) and "@" in mail:
                    return {
                        "kind": "user",
                        "email": mail,
                        "display_name": display,
                    }
                return None
            if odata_type != "#microsoft.graph.serviceprincipal":
                # Group / device / orgcontact — no human-mappable
                # creator. Bail.
                return None

            sp_display = obj.get("displayName") or ""
            sp_app_id = obj.get("appId") or ""

            # Owner lookup. ``$select`` keeps the payload tiny.
            try:
                owners_resp = await client.get(
                    f"{_GRAPH_BASE}/servicePrincipals/{oid}/owners",
                    params={
                        "$select": "id,mail,userPrincipalName,"
                                   "displayName",
                        "$top": "20",
                    },
                    headers=headers,
                )
            except (httpx.TransportError, OSError) as exc:
                logger.info(
                    "Azure SP enrichment %s: transport error on "
                    "owners (%s)", oid, exc,
                )
                # Return SP info without owner — still upgrades display_name.
                return {
                    "kind": "sp",
                    "sp_display_name": sp_display,
                    "sp_app_id": sp_app_id,
                    "owner_email": None,
                    "owner_display_name": None,
                    "owner_count": 0,
                }
            if owners_resp.status_code in (401, 403, 404):
                # 403 typically means Application.Read.All not granted
                # (Directory.Read.All sometimes lacks the
                # ``servicePrincipals/{id}/owners`` action depending
                # on tenant policy). Still useful: we got the SP name.
                logger.info(
                    "Azure SP enrichment %s/%s: owners %d — "
                    "returning SP-name only",
                    self.integration_id, oid, owners_resp.status_code,
                )
                return {
                    "kind": "sp",
                    "sp_display_name": sp_display,
                    "sp_app_id": sp_app_id,
                    "owner_email": None,
                    "owner_display_name": None,
                    "owner_count": 0,
                }
            if owners_resp.status_code != 200:
                logger.info(
                    "Azure SP enrichment %s/%s: owners %d — %s",
                    self.integration_id, oid, owners_resp.status_code,
                    _safe_body(owners_resp, max_len=200),
                )
                return {
                    "kind": "sp",
                    "sp_display_name": sp_display,
                    "sp_app_id": sp_app_id,
                    "owner_email": None,
                    "owner_display_name": None,
                    "owner_count": 0,
                }
            owners = owners_resp.json().get("value") or []
            owner_email: str | None = None
            owner_display: str | None = None
            user_owner_count = 0
            for owner in owners:
                if str(
                    owner.get("@odata.type") or "",
                ).lower() != "#microsoft.graph.user":
                    continue
                user_owner_count += 1
                if owner_email is not None:
                    continue
                mail = owner.get("mail") or owner.get("userPrincipalName")
                if isinstance(mail, str) and "@" in mail:
                    owner_email = mail
                    owner_display = owner.get("displayName")
            return {
                "kind": "sp",
                "sp_display_name": sp_display,
                "sp_app_id": sp_app_id,
                "owner_email": owner_email,
                "owner_display_name": owner_display,
                "owner_count": user_owner_count,
            }

    @classmethod
    def _azure_event_to_creator_info(
        cls, event: dict[str, Any], event_time: datetime | None,
    ) -> ResourceCreatorInfo:
        """Parse one Activity Log event → ResourceCreatorInfo."""
        # ``caller`` is the AAD UPN that performed the operation;
        # ``claims.{name,upn,oid}`` carry richer identity.
        caller = event.get("caller") or ""
        claims = event.get("claims") or {}

        email: str | None = None
        if isinstance(caller, str) and "@" in caller:
            email = caller
        else:
            upn_claim = claims.get(
                "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/upn",
            ) or claims.get("upn")
            if isinstance(upn_claim, str) and "@" in upn_claim:
                email = upn_claim

        principal_id = (
            claims.get(
                "http://schemas.microsoft.com/identity/claims/objectidentifier",
            )
            or claims.get("oid")
            or caller
            or "<unknown>"
        )

        display_name: str | None = None
        if email:
            display_name = email.split("@", 1)[0].replace(".", " ").title()
        elif isinstance(caller, str) and caller:
            display_name = caller

        _RAW_CAP_BYTES = 16 * 1024
        raw_to_store: dict[str, Any] = {
            "operation_name": (
                event.get("operationName") or {}
            ).get("value"),
            "event_timestamp": event.get("eventTimestamp"),
            "caller": caller,
            "correlation_id": event.get("correlationId"),
            "resource_provider_name": (
                event.get("resourceProviderName") or {}
            ).get("value"),
            "status": (event.get("status") or {}).get("value"),
        }
        # Trim claims to the most useful keys to stay under cap.
        slim_claims = {
            k: v for k, v in claims.items()
            if k in {"name", "upn", "oid", "appid", "iss"}
        }
        if slim_claims:
            raw_to_store["claims"] = slim_claims
        if len(json.dumps(raw_to_store, default=str)) > _RAW_CAP_BYTES:
            raw_to_store.pop("claims", None)

        return ResourceCreatorInfo(
            principal_id=str(principal_id),
            email=email,
            display_name=display_name,
            created_at_cloud=event_time,
            source="azure_activity",
            raw=raw_to_store,
        )


# ── HTTP retry primitive ─────────────────────────────────────────
#
# Combined transport-level + HTTP-status retry for every outbound POST
# the Azure driver does that lives on a cron path (cost ingestion +
# token endpoint). Other Azure paths (IAM grant/revoke, foundry
# provisioning) deliberately stay outside this primitive — they're
# wrapped by their own orchestrators (provisioning queue / offboarding
# transfer / Temporal) which retry the whole flow with richer context.
#
# Why both layers in one helper:
#   * 429 / 5xx — covered before this PR for ``list_cost_facts`` only.
#     Azure Cost Management's per-tenant + per-scope rate limiter (see
#     ``_parse_azure_cost_retry_after`` below) fires aggressively when
#     a tenant has many subs — bit prod 2026-05-11 (3-of-4 subs 429'd
#     in one tick).
#   * Transport (httpx.ConnectError / ReadTimeout / RemoteProtocolError
#     / WriteError / PoolTimeout) — what bit prod 2026-05-21 AND
#     2026-05-22 on the SAME Azure subscription (PingCAP Azure OpenAI,
#     7bc8ab17-...): TCP-level flake against the Cost Management
#     backend (resource_sync on the same sub succeeded that same night
#     — not auth, not quota, transport-only). Pre-fix the retry loop
#     only catches status codes; a transport exception bypassed it and
#     killed one scope per cron, dropping a day of cost coverage.
#
# Backoff semantics:
#   * Transport: exponential ``min(2 ** attempt, 60)``, no Retry-After
#     (the server hasn't responded yet to give us one).
#   * HTTP-status: honour ``parse_retry_after(headers)`` if provided
#     (Azure Cost Management has 4 candidate headers; ``_parse_azure_
#     cost_retry_after`` returns the LARGEST), else exponential as
#     above. Cap any sleep at 90s so a misbehaving server can't peg
#     a worker for the worker's entire cron window.
#
# Retry budget:
#   * 5 attempts default — same as the existing ``list_cost_facts`` loop.
#     A scope that's permanently failing after 5 transport retries is
#     a real outage, surface as a per-scope error so the partial-failure
#     notifier fires and the other 37 scopes still complete.

_TRANSPORT_RETRY_EXCEPTIONS: tuple[type[BaseException], ...] = (
    # ConnectError covers DNS failure + TCP RST + TLS handshake.
    # The prod 2026-05-22 incident raised this exact class.
    httpx.ConnectError,
    httpx.ConnectTimeout,
    # ReadError + ReadTimeout cover mid-stream connection loss after
    # the request was already sent — Azure ARM endpoints occasionally
    # 504 silently by dropping the socket instead of returning a status.
    httpx.ReadError,
    httpx.ReadTimeout,
    httpx.WriteError,
    httpx.WriteTimeout,
    # PoolTimeout fires when our local connection pool can't lease a
    # socket within the timeout — usually a sign upstream is slow and
    # earlier requests are still in flight. Treating it as transient
    # gives the pool time to drain.
    httpx.PoolTimeout,
    # RemoteProtocolError covers HTTP/1.1 + HTTP/2 framing violations
    # we've seen from misconfigured intermediate proxies.
    httpx.RemoteProtocolError,
)

_DEFAULT_MAX_HTTP_ATTEMPTS = 5

# Hard ceiling on how long we'll honour a server-stated ``Retry-After``
# before declaring the throttle terminal and surfacing the 429/5xx to
# the caller. Code-review pass 1 (2026-05-23) flagged that capping a
# server's "wait 600s" instruction at 90s GUARANTEES another 429 on
# the retry (the load shedder hasn't moved its window) — so the 5-
# attempt budget would burn deterministically with no chance of
# success. Better posture: if the server says "wait longer than I
# can plausibly hold a worker on this scope", give up immediately so
# the failure-notifier fires once with the real "Retry-After=600s"
# context. The cron retries tomorrow when the throttle window has
# rolled. 90s matches:
#   * cost_service.py's per-scope timeout (_COST_FACTS_PER_SCOPE_TIMEOUT_SEC
#     = 300s) — one 90s sleep + one retry round still fits inside the
#     timeout with headroom for the actual query
#   * Azure CM's typical entity-retry-after values (1-60s in healthy
#     load, 90s+ only when a tenant is severely throttled)
_MAX_HONOURED_RETRY_AFTER_SEC = 90

# ── Cost Management client-side pacing ───────────────────────────
#
# Measured on prod 2026-09-04/05 (the monthly ``cmdb.cost_facts_
# monthclose`` fan-out, 5 chunks × 36 subscriptions). Worker log
# excerpt, one line per subscription, for 77 straight minutes:
#
#     23:02:41 429 sub=fc0a85ea attempt=1/5 sleeping 45s
#     23:03:48 429 sub=f0c99419 attempt=1/5 sleeping 38s
#     23:04:47 429 sub=7bc8ab17 attempt=1/5 sleeping 39s
#     …  (83 such lines; server-stated cooldowns 26–58s)
#
# i.e. **every** subscription 429s on its FIRST Cost Management query
# and only succeeds after sleeping out the server-stated cooldown.
# That is not an incident — it is the steady state: the driver
# discovers the rate limit by tripping it, once per scope, forever.
# ~80% of a 36-scope chunk's wall clock is throttle sleep, and the
# limiter is per-tenant/per-client-type so the burst also throttles
# any interactive dashboard query issued during the run.
#
# Two concrete defects fell out of that trace, both fixed here:
#
# 1. **Header-less 429s burn retry attempts.** Azure does not always
#    set a Retry-After header on a CM 429. The old fallback was
#    ``min(2 ** attempt, 60)`` starting at 1s, so the sub that finally
#    failed (7f070020, 00:04:34) slept 56s → 2s → 57s → 8s: the 2s and
#    8s rounds could not possibly clear a ~60s window, so 2 of its 5
#    attempts were spent on requests guaranteed to 429. Post-fix the
#    fallback is the LAST server-stated cooldown this pacer learned
#    (``suggested_backoff_s``), floored at ``_CM_MIN_THROTTLE_BACKOFF_SEC``.
#
# 2. **No client-side pacing.** Nothing carries "the server just told
#    us to wait 45s" from one scope to the next, so scope N+1 fires
#    immediately into a limiter we already know is closed. The pacer
#    below re-arms the gate after every CM response and halves the
#    spacing on each un-retried success (AIMD): the driver converges on
#    the real rate from below instead of rediscovering it by 429ing.
#
# Constants are grounded in the measurement above, not guessed:
#   * ``_CM_PACE_CAP_SEC`` = 60 — the top of the observed 26–58s band.
#   * ``_CM_MIN_THROTTLE_BACKOFF_SEC`` = 20 — conservatively under the
#     bottom of that band; still 20× the old 1s first retry.
#   * ``_CM_THROTTLE_BUDGET_SEC`` = 200 — cumulative retry sleep per
#     call. Keeps a throttled scope inside ``cost_service``'s 300s
#     ``_COST_FACTS_PER_SCOPE_TIMEOUT_SEC`` (200 budget + ≤60 pacing
#     wait + ~10s of real query time observed in the trace) so it
#     surfaces as an honest "throttled" error instead of the useless
#     ``status='timeout'`` the old unbounded 4 × 90s could produce.
_CM_PACE_CAP_SEC = 60.0
_CM_MIN_THROTTLE_BACKOFF_SEC = 20.0
_CM_THROTTLE_BUDGET_SEC = 200.0
# Multiplicative decrease applied to the learned spacing after a
# request that succeeded WITHOUT needing a retry — probes back toward
# full speed so a single hostile minute can't pace the rest of the run
# at 60s/scope.
_CM_PACE_DECAY = 0.5
# Hard ceiling on how much PROACTIVE pacing one driver instance (= one
# cron run) may add. The daily ``cmdb.cost_facts.azure-primary`` job has
# no ``timeout_secs`` override, so it inherits the executor's 1800s
# default, and prod runs finish in 587-775s. Pacing is expected to be
# wall-clock-neutral (it replaces 429 round-trips that were sleeping the
# same cooldown anyway, and the AIMD decay converges on the real rate) —
# but "expected" is a model of Azure's limiter, and the model being
# wrong must not turn a 700s run into a timed-out one. 600s of pacing
# keeps the worst case at ~1300s, inside 1800s, and far inside the 7200s
# the month-close fan-out passes. Past the budget the driver reverts to
# the pre-pacing reactive behaviour (learn cooldowns, retry on 429) and
# says so once in the log.
_CM_PACE_RUN_BUDGET_SEC = 600.0


class _AzureThrottlePacer:
    """Carries an observed rate-limit cooldown across requests.

    One instance per driver (``AzureCloudDriver._cm_pacer``) is shared
    by every Cost Management call in a run, so the cooldown learned on
    subscription N is honoured before subscription N+1's first request
    instead of being rediscovered with another 429.

    The gate is *consuming*: :meth:`wait` sleeps out any pending
    deadline and then clears it, so a spacing deadline is honoured
    exactly once. It covers proactive spacing only — a request's own
    retry backoff is slept by the retry loop, so a cooldown can never
    be paid twice (once as backoff, then again as pace).

    ``pacing_enabled=False`` (the default, used for the throwaway
    per-call instance on non-cost endpoints) makes this a pure
    backoff-carrier with no proactive spacing — byte-for-byte the
    pre-pacer behaviour.
    """

    def __init__(
        self,
        *,
        pacing_enabled: bool = False,
        min_backoff_s: float = 0.0,
        pace_cap_s: float = _CM_PACE_CAP_SEC,
        pace_run_budget_s: float = _CM_PACE_RUN_BUDGET_SEC,
    ) -> None:
        self._pacing_enabled = pacing_enabled
        self._min_backoff_s = min_backoff_s
        self._pace_cap_s = pace_cap_s
        self._pace_run_budget_s = pace_run_budget_s
        # Monotonic deadline before which the next request must not be
        # issued. 0.0 = gate open.
        self._not_before: float = 0.0
        # Learned inter-request spacing, seeded from server-stated
        # cooldowns. 0.0 = nothing learned yet.
        self._pace_s: float = 0.0
        # Cumulative proactive pacing spent by this instance, against
        # ``_CM_PACE_RUN_BUDGET_SEC``.
        self._paced_total_s: float = 0.0
        self._budget_exhausted_logged = False

    @property
    def suggested_backoff_s(self) -> float:
        """Backoff to use for a 429 that carried no Retry-After header.

        Returns 0.0 when nothing has been learned and no floor is
        configured, so the caller falls back to its exponential.
        """
        return max(self._pace_s, self._min_backoff_s)

    def arm(self, seconds: float) -> None:
        """Hold the next request for at least ``seconds``."""
        if seconds <= 0:
            return
        self._not_before = max(
            self._not_before, time.monotonic() + seconds,
        )

    def observe_throttle(self, retry_after_s: float | None) -> None:
        """Learn the spacing from a 429 (``None`` = no header present)."""
        learned = retry_after_s if retry_after_s else self.suggested_backoff_s
        if learned > 0:
            self._pace_s = min(float(learned), self._pace_cap_s)

    def arm_learned(self) -> None:
        """Hold the next request for the cooldown already learned.

        Used when a request GIVES UP on a 429: the limiter is still
        closed, so the next caller through this pacer must not fire
        straight into it. No decay — nothing succeeded. Bounded by
        ``_pace_cap_s`` because :meth:`observe_throttle` clamps there,
        so a hostile ``Retry-After: 600`` cannot park the next scope
        past its own timeout.
        """
        self._arm_pace()

    def _arm_pace(self) -> None:
        """Arm the learned spacing, unless this run's pacing budget is
        spent (then behave exactly as the pre-pacing driver did)."""
        if not self._pacing_enabled or self._pace_s <= 0:
            return
        if self._paced_total_s >= self._pace_run_budget_s:
            if not self._budget_exhausted_logged:
                self._budget_exhausted_logged = True
                logger.warning(
                    "Azure Cost Management pacing budget spent "
                    "(%.0fs >= %.0fs) — reverting to reactive 429 "
                    "handling for the rest of this run",
                    self._paced_total_s, self._pace_run_budget_s,
                )
            return
        self.arm(self._pace_s)

    def observe_success(self, *, retried: bool) -> None:
        """Record a 2xx and re-arm proactive spacing for the next call."""
        if not self._pacing_enabled:
            return
        if not retried and self._pace_s > 0:
            self._pace_s *= _CM_PACE_DECAY
            # Below a second the spacing is noise — reopen fully so a
            # healthy tenant runs at full speed.
            if self._pace_s < 1.0:
                self._pace_s = 0.0
        self._arm_pace()

    async def wait(self) -> float:
        """Sleep out any pending deadline, then clear it.

        Returns the number of seconds actually slept (0.0 when open).
        """
        deadline = self._not_before
        self._not_before = 0.0
        if deadline <= 0:
            return 0.0
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return 0.0
        await asyncio.sleep(remaining)
        self._paced_total_s += remaining
        return remaining


def _throttle_headers_diag(headers: httpx.Headers) -> str:
    """Render the throttle headers present on a response, for the error
    text and log line.

    The prod 429 that surfaced to ``scheduled_job_runs.errors`` on
    2026-09-04 carried only Azure's generic body
    (``{"error":{"code":"429","message":"Too many requests. Please
    retry."}}``), which says nothing about WHICH limiter fired or how
    long it wanted — so the on-call could not tell "no header, we
    retried after 2s" from "server said 600s, we gave up". Naming the
    headers (and their absence) makes the next one self-diagnosing.
    """
    seen = []
    for key, label in (
        ("Retry-After", "retry_after"),
        (
            "x-ms-ratelimit-microsoft.costmanagement-entity-retry-after",
            "entity",
        ),
        (
            "x-ms-ratelimit-microsoft.costmanagement-clienttype-retry-after",
            "clienttype",
        ),
        (
            "x-ms-ratelimit-microsoft.costmanagement-tenant-retry-after",
            "tenant",
        ),
    ):
        raw = headers.get(key)
        if raw:
            seen.append(f"{label}={raw}")
    return ",".join(seen) if seen else "none"


async def _post_with_retry(
    client: httpx.AsyncClient,
    url: str,
    *,
    json: Any = None,
    data: Any = None,
    headers: dict[str, str] | None = None,
    endpoint_label: str,
    max_attempts: int = _DEFAULT_MAX_HTTP_ATTEMPTS,
    parse_retry_after: Any = None,
    pacer: _AzureThrottlePacer | None = None,
    throttle_budget_s: float | None = None,
) -> httpx.Response:
    """POST with combined transport + 429/5xx retry.

    Returns the final ``httpx.Response`` (which may still be 4xx/5xx if
    the retry budget is exhausted — the caller raises with its own
    sanitised error body). Re-raises the last transport exception if
    every attempt failed with transport errors (preserves the original
    ``ConnectError`` / ``ReadTimeout`` traceback for failure-notifier
    via ``format_exception_for_log``).

    Parameters
    ----------
    endpoint_label
        Short human-readable label for log lines — typically the
        method name plus scope id (e.g. ``"cost_facts sub=7bc8ab17-..."``
        or ``"_token resource=https://management.azure.com"``).
    parse_retry_after
        Optional callable ``(httpx.Headers) -> int | None`` for
        vendor-specific Retry-After header parsing. Defaults to canonical
        ``Retry-After`` header if absent.
    pacer
        Optional :class:`_AzureThrottlePacer` shared across calls to the
        same rate-limited endpoint family (the driver passes its
        ``_cm_pacer`` for every Cost Management call so a cooldown
        learned on one subscription is honoured before the next one's
        first request). Omitted → a throwaway non-pacing instance, which
        behaves exactly as this helper did before pacing existed.
    throttle_budget_s
        Optional cumulative ceiling on retry sleep within THIS call.
        When the next backoff would push the total past it we stop
        retrying and return the throttled response, so the caller raises
        a truthful "throttled" error instead of being killed by an outer
        timeout mid-sleep. ``None`` → no cumulative ceiling (each
        individual sleep is still capped at
        ``_MAX_HONOURED_RETRY_AFTER_SEC``).
    """
    attempt = 0
    # Non-pacing throwaway keeps the single code path below honest for
    # callers that don't share a pacer.
    gate = pacer if pacer is not None else _AzureThrottlePacer()
    slept_total = 0.0
    # Build kwargs without ``data=None`` / ``json=None`` — some tests
    # mock ``client.post`` with a narrower signature
    # (``post(self, url, json, headers)``) and would TypeError if we
    # always passed the alternate body kwarg. Real httpx.AsyncClient
    # accepts None for both, but the fakes don't.
    post_kwargs: dict[str, Any] = {"headers": headers}
    if json is not None:
        post_kwargs["json"] = json
    if data is not None:
        post_kwargs["data"] = data
    while True:
        # Proactive spacing only: sleeps out a deadline armed by an
        # EARLIER request through a shared pacer, then reopens. This
        # call's own retry backoff is slept inline in the branches
        # below — keeping the two sources separate is what stops a
        # cooldown being slept twice (once as backoff, once as pace).
        await gate.wait()
        try:
            resp = await client.post(url, **post_kwargs)
        except _TRANSPORT_RETRY_EXCEPTIONS as exc:
            if attempt >= max_attempts - 1:
                # Exhausted — re-raise the ORIGINAL exception so the
                # caller's ``format_exception_for_log(exc)`` surfaces
                # the real httpx exception type in the failure-notifier
                # (was the gap fixed by PR #465's observability sweep —
                # don't reintroduce by wrapping in a generic RuntimeError
                # here).
                raise
            delay = min(2 ** attempt, 60)
            logger.warning(
                "Azure HTTP transport error %s attempt=%d/%d "
                "delay=%ds: %s",
                endpoint_label, attempt + 1, max_attempts, delay,
                format_exception_for_log(exc, max_len=200),
            )
            await asyncio.sleep(delay)
            attempt += 1
            continue

        # Got a response — apply HTTP-status retry policy.
        if resp.status_code == 429 or resp.status_code >= 500:
            retry_after_s: int | None = None
            if parse_retry_after is not None:
                try:
                    retry_after_s = parse_retry_after(resp.headers)
                except Exception:  # noqa: BLE001
                    # A buggy parser must not break retry — fall through
                    # to exponential. Bound the log volume because this
                    # would otherwise spam at every retry.
                    if attempt == 0:
                        logger.exception(
                            "Azure retry-after parser raised for %s; "
                            "falling back to exponential",
                            endpoint_label,
                        )
                    retry_after_s = None
            if retry_after_s is None:
                # Canonical Retry-After (integer seconds) — covers
                # non-cost AAD/Graph endpoints whose throttle response
                # only sets the standard header.
                raw = resp.headers.get("Retry-After")
                if raw:
                    try:
                        v = int(raw)
                        if v > 0:
                            retry_after_s = v
                    except (TypeError, ValueError):
                        pass
            backoff_source = "server" if retry_after_s is not None else None
            if resp.status_code == 429:
                # Teach the pacer what the server just said (``None``
                # when header-less — it then keeps whatever it already
                # learned), so both this call's next backoff and the
                # NEXT call's proactive spacing use a real cooldown.
                gate.observe_throttle(retry_after_s)
            if retry_after_s is None and resp.status_code == 429:
                # Header-less 429: prefer the last cooldown this pacer
                # learned over the exponential ramp. Azure CM's stated
                # cooldowns run 26-58s (prod 2026-09-04), so the old
                # ``min(2 ** attempt, 60)`` spent attempts 2 and 4 on
                # 2s / 8s sleeps that could not clear the window —
                # exactly how sub 7f070020 burned its retry budget.
                # 429 ONLY: a 5xx is not a throttle, and pacing a
                # transient server error behind a 45s cooldown would
                # slow recovery for no reason. ``suggested_backoff_s``
                # is 0.0 on a pacer that has learned nothing and has no
                # floor (every non-cost endpoint), which preserves the
                # exponential there too.
                suggested = gate.suggested_backoff_s
                if suggested > 0:
                    retry_after_s = int(round(suggested))
                    backoff_source = "learned"
            if retry_after_s is None:
                retry_after_s = min(2 ** attempt, 60)
                backoff_source = "exponential"
            # Code-review #1 fix: if the server-stated cooldown exceeds
            # our honour ceiling, retrying inside the ceiling is
            # guaranteed to 429 again (the load shedder won't have
            # moved its window) — burning the full 5-attempt budget on
            # known-to-fail attempts is strictly worse than surfacing
            # the throttle immediately so the failure-notifier fires
            # once with the real "Retry-After=600s" context. Caller's
            # ``status_code != 200`` check then raises with the
            # sanitised body. Checked BEFORE the plain attempt-budget
            # exhaustion so the log line attributes the give-up to the
            # actionable reason ("server wants 600s") rather than the
            # generic one, when both are true on the last attempt.
            give_up: str | None = None
            if retry_after_s > _MAX_HONOURED_RETRY_AFTER_SEC:
                give_up = (
                    f"server stated Retry-After={retry_after_s}s > honour "
                    f"ceiling {_MAX_HONOURED_RETRY_AFTER_SEC}s"
                )
            elif (
                throttle_budget_s is not None
                and slept_total + retry_after_s > throttle_budget_s
            ):
                # Same posture, cumulative axis: a scope that has already
                # slept out its whole budget is being hard-throttled, and
                # the caller (``cost_service``) has a 300s per-scope
                # timeout. Stopping here yields "throttled after Ns" in
                # ``scheduled_job_runs.errors``; sleeping on would get the
                # coroutine cancelled mid-sleep and recorded as the far
                # less useful ``status='timeout'``.
                give_up = (
                    f"throttle sleep budget exhausted "
                    f"({slept_total:.0f}s slept, next backoff "
                    f"{retry_after_s}s > budget {throttle_budget_s:.0f}s)"
                )
            elif attempt >= max_attempts - 1:
                # Exhausted on status — return the final response and
                # let the caller raise with its own sanitised error
                # (call sites have different body-redaction policies).
                give_up = f"retry budget exhausted after {max_attempts} attempts"
            if give_up is not None:
                if resp.status_code == 429:
                    # We are giving up while the limiter is still
                    # closed. Hold the gate so the NEXT caller — the
                    # next SUBSCRIPTION in the same cron run, sharing
                    # this driver's pacer — does not fire straight into
                    # it and turn one throttled scope into a cascade.
                    # Bounded by ``_pace_cap_s``, so a hostile
                    # "Retry-After: 600" cannot park the next scope past
                    # its own timeout.
                    gate.arm_learned()
                logger.warning(
                    "Azure HTTP %s %s giving up: %s (attempt %d/%d, "
                    "throttle_headers=%s)",
                    resp.status_code, endpoint_label, give_up,
                    attempt + 1, max_attempts,
                    _throttle_headers_diag(resp.headers),
                )
                return resp
            logger.warning(
                "Azure HTTP %s %s attempt=%d/%d sleeping %ds before "
                "retry (source=%s, throttle_headers=%s)",
                resp.status_code, endpoint_label,
                attempt + 1, max_attempts, retry_after_s,
                backoff_source,
                _throttle_headers_diag(resp.headers),
            )
            await asyncio.sleep(retry_after_s)
            slept_total += retry_after_s
            attempt += 1
            continue

        # 2xx or non-retriable 4xx — return to caller.
        if resp.status_code < 400:
            gate.observe_success(retried=attempt > 0)
        return resp


def _parse_azure_cost_retry_after(
    headers: httpx.Headers,
) -> int | None:
    """Extract the server-stated cooldown (in seconds) from an Azure
    Cost Management 429/5xx response.

    Azure Cost Management exposes the throttle decision via multiple
    headers — listed in spec preference order:

    * ``Retry-After`` — canonical HTTP header, integer seconds.
    * ``x-ms-ratelimit-microsoft.costmanagement-entity-retry-after``
      — per-scope (per-subscription) limit, the one most likely set
      when one sub is hot but the tenant isn't.
    * ``x-ms-ratelimit-microsoft.costmanagement-clienttype-retry-after``
      — per-client-type (per-SP) limit, set when our SP is hammering
      the API across many scopes.
    * ``x-ms-ratelimit-microsoft.costmanagement-tenant-retry-after``
      — tenant-wide limit, the most aggressive.

    We honour the LARGEST of any present value: backing off less than
    what the server told us guarantees another 429 (Microsoft's load
    shedder does not get nicer if you ignore it). Returns ``None`` if
    no header is present, so the caller can fall back to exponential.
    """
    candidates: list[int] = []
    for key in (
        "Retry-After",
        "x-ms-ratelimit-microsoft.costmanagement-entity-retry-after",
        "x-ms-ratelimit-microsoft.costmanagement-clienttype-retry-after",
        "x-ms-ratelimit-microsoft.costmanagement-tenant-retry-after",
    ):
        raw = headers.get(key)
        if not raw:
            continue
        try:
            v = int(raw)
        except (TypeError, ValueError):
            continue
        if v > 0:
            candidates.append(v)
    if not candidates:
        return None
    return max(candidates)


def _parse_azure_resource_id(
    rid: str | None,
) -> tuple[str | None, str | None, str | None]:
    """Pull (resource_group, resource_type, resource_name) out of an
    Azure resource id.

    Azure resource ids have the canonical shape::

        /subscriptions/{sub}/resourceGroups/{rg}/providers/{ns}/{type}/{name}

    With nested children there can be additional ``/subtype/subname``
    segments — we only return the top-level type/name pair, since
    that's what the dashboard pivots on.

    Returns ``(None, None, None)`` for tenant-overhead lines whose
    ResourceId is empty or lacks the expected segments.
    """
    if not rid:
        return None, None, None
    parts = rid.strip("/").split("/")
    rg: str | None = None
    rtype: str | None = None
    rname: str | None = None
    # Walk the segments looking for ``resourceGroups`` (case-
    # insensitive — Azure normalises to lowercase but we don't trust
    # the cloud to be consistent here).
    for i, seg in enumerate(parts[:-1]):
        if seg.lower() == "resourcegroups" and i + 1 < len(parts):
            rg = parts[i + 1]
            break
    # type/name pair: after ``providers/{ns}/{type}/{name}``
    for i, seg in enumerate(parts[:-1]):
        if seg.lower() == "providers" and i + 3 < len(parts):
            rtype = f"{parts[i + 1]}/{parts[i + 2]}"
            rname = parts[i + 3]
            break
    return rg, rtype, rname