"""Tencent cloud driver.

CMDB discovery via :class:`TencentProvider` (servers / domains / DNS / LB /
CDN / COS / IAM).

Account lifecycle: **IAM mode** — Tencent is typically a single-tenant
shop (similar to AWS SG). A grant = a CAM sub-account (``CreateUser``)
+ a policy attachment (``AttachUserPolicy``). Multi-account via
Resource Directory exists but is rare in practice; if the admin
wants it they can flip ``account_mode=sso`` and we'll degrade to
manual mode (recording intent, IT executes).

Role key convention: the driver uses the CAM **PolicyName** (e.g.
``AdministratorAccess``, ``QCloudReadOnlyAccess``, or a customer-
defined policy's name) as the ``role_key`` everywhere — same as
Aliyun RAM. CAM's ``AttachUserPolicy`` / ``DetachUserPolicy`` require
the numeric ``PolicyId``, so the driver resolves name → id via
``ListPolicies`` on the fly (cached per call).

Billing: daily per-day cost via ``DescribeBillDetail`` with client-side
aggregation (Tencent doesn't expose day-granular totals natively).
Scope is tenant-wide — Tencent has no sub-account concept that maps to
a CmdbCloudScope.external_id.

Offboarding: ``revoke_account`` deletes the CAM user in the correct
order (detach policies, leave groups, delete login profile, delete
access keys, DeleteUser).
"""
from __future__ import annotations

import asyncio
import logging
from collections.abc import Callable, Hashable
from datetime import UTC, date, datetime, timedelta
from typing import Any

try:
    from tencentcloud.billing.v20180709 import billing_client
    from tencentcloud.billing.v20180709 import models as billing_models
    from tencentcloud.cam.v20190116 import cam_client
    from tencentcloud.cam.v20190116 import models as cam_models
    from tencentcloud.common import credential as tc_credential
    from tencentcloud.common.exception.tencent_cloud_sdk_exception import (
        TencentCloudSDKException,
    )
    from tencentcloud.common.profile.client_profile import ClientProfile
    from tencentcloud.common.profile.http_profile import HttpProfile
    _TC_AVAILABLE = True
except ImportError:  # pragma: no cover
    billing_client = None  # type: ignore[assignment]
    billing_models = None  # type: ignore[assignment]
    cam_client = None  # type: ignore[assignment]
    cam_models = None  # type: ignore[assignment]
    tc_credential = None  # type: ignore[assignment]
    TencentCloudSDKException = Exception  # type: ignore[assignment, misc]
    ClientProfile = None  # type: ignore[assignment]
    HttpProfile = None  # type: ignore[assignment]
    _TC_AVAILABLE = False

from app.core.phone import country_dialing_code, to_e164
from app.domains.cmdb.providers.exceptions import (
    ProviderAuthError,
    redact_provider_error,
)
from app.domains.cmdb.providers.tencent import TencentProvider
from app.integrations._tencent_throttle import (
    tencent_call_sync as _tc_call,
)
from app.integrations.driver_base import (
    CloseScopeResult,
    CloudAccessContext,
    CloudAccountRevokeContext,
    CloudDailyCost,
    CloudGrantResult,
    CloudPrincipalNotFoundError,
    CloudRoleInfo,
    CloudScopeInfo,
    CloudScopeRef,
    CostFact,
    DiscoveredGrant,
    ProvisioningResult,
)
from app.integrations.drivers._cloud_shared import (
    _CloudDriverBase,
    account_id_candidates,
)
from app.integrations.registry import register_driver
from app.integrations.schemas import (
    IntegrationFieldInfo,
    IntegrationTestResult,
)

logger = logging.getLogger(__name__)


# ── Default-promoted Tencent CAM preset policies ────────────────────
# The narrow set surfaced in the request modal by default. Every other
# preset returned by ``ListPolicies`` is still synced (so the admin
# promote/demote UI can pull it in) but starts unpromoted. Override
# via the ``cloud.tencent.default_policies`` platform setting.
_TENCENT_DEFAULT_FALLBACK: list[dict[str, Any]] = [
    {
        "role_key": "ReadOnlyAccess",
        "display_name": "ReadOnlyAccess",
        "description": "只读全部资源",
    },
    {
        "role_key": "AdministratorAccess",
        "display_name": "AdministratorAccess",
        "description": "全部资源管理(慎授)",
    },
    {
        "role_key": "QCloudCVMFullAccess",
        "display_name": "QCloudCVMFullAccess",
        "description": "CVM 管理",
    },
    {
        "role_key": "QCloudCVMReadOnlyAccess",
        "display_name": "QCloudCVMReadOnlyAccess",
        "description": "CVM 只读",
    },
    {
        "role_key": "QCloudCDBFullAccess",
        "display_name": "QCloudCDBFullAccess",
        "description": "云数据库 MySQL 管理",
    },
    {
        "role_key": "QCloudCOSFullAccess",
        "display_name": "QCloudCOSFullAccess",
        "description": "对象存储 COS 管理",
    },
    {
        "role_key": "QCloudCOSReadOnlyAccess",
        "display_name": "QCloudCOSReadOnlyAccess",
        "description": "对象存储 COS 只读",
    },
]

# 100 × 200 = 20k CAM policies, far above a realistic tenant catalog.
_MAX_CAM_POLICY_PAGES = 100
_MAX_CAM_ATTACHED_POLICY_PAGES = 10


def _collect_authoritative_cam_pages(
    *,
    fetch_page: Callable[[int], Any],
    items_attr: str,
    page_size: int,
    max_pages: int,
    item_key: Callable[[Any], Hashable | None],
    item_identity: str,
    operation: str,
    allow_same_page_duplicates: bool = False,
) -> list[Any]:
    """Collect a CAM snapshot only when its pagination is self-consistent.

    ``ListAttachedUserAllPolicies(AttachType=0)`` may legitimately contain
    the same effective policy twice in one response when CAM reports its
    direct and group-derived attachment paths separately.  Its ``TotalNum``
    counts those raw rows.  That is different from a repeated *page*, which
    would make the snapshot non-authoritative and must still fail closed.
    """
    collected: list[Any] = []
    seen: set[Hashable] = set()
    expected_total: int | None = None
    raw_observed = 0

    for page in range(1, max_pages + 1):
        try:
            response = fetch_page(page)
        except Exception as exc:  # noqa: BLE001
            raise RuntimeError(
                f"Tencent {operation} page={page} failed: "
                f"{redact_provider_error(exc)}",
            ) from exc

        total_raw = getattr(response, "TotalNum", None)
        try:
            if total_raw is None:
                raise TypeError("TotalNum is missing")
            total = int(total_raw)
        except (TypeError, ValueError) as exc:
            raise RuntimeError(
                f"Tencent {operation} returned an invalid TotalNum at "
                f"page={page}; refusing a non-authoritative snapshot",
            ) from exc
        if total < 0:
            raise RuntimeError(
                f"Tencent {operation} returned a negative TotalNum at "
                f"page={page}; refusing a non-authoritative snapshot",
            )
        if expected_total is None:
            expected_total = total
        elif total != expected_total:
            raise RuntimeError(
                f"Tencent {operation} changed TotalNum during pagination "
                f"({expected_total} -> {total} at page={page}); refusing "
                "a non-atomic snapshot",
            )

        batch = list(getattr(response, items_attr, None) or [])
        page_seen: set[Hashable] = set()
        for item in batch:
            key = item_key(item)
            if key is None or (isinstance(key, str) and not key.strip()):
                raise RuntimeError(
                    f"Tencent {operation} returned an item without "
                    f"{item_identity} at page={page}; refusing a partial "
                    "snapshot",
                )
            if key in page_seen and allow_same_page_duplicates:
                # CAM's aggregate attachment response counts a policy once
                # per attachment path.  Return a single effective policy,
                # while ``raw_observed`` below remains aligned to TotalNum.
                continue
            if key in seen:
                raise RuntimeError(
                    f"Tencent {operation} returned an overlapping/repeated "
                    f"page at page={page}; refusing a partial snapshot",
                )
            page_seen.add(key)
            seen.add(key)
            collected.append(item)

        raw_observed += len(batch)
        if raw_observed >= total:
            if raw_observed != total:
                observed_label = "rows" if allow_same_page_duplicates else "unique rows"
                raise RuntimeError(
                    f"Tencent {operation} returned {raw_observed} {observed_label} "
                    f"for TotalNum={total} at page={page}; refusing a "
                    "non-authoritative snapshot",
                )
            return collected
        if not batch:
            raise RuntimeError(
                f"Tencent {operation} returned an empty page before "
                f"TotalNum was exhausted (page={page}, total={total}); "
                "refusing a partial snapshot",
            )
        if len(batch) < page_size:
            raise RuntimeError(
                f"Tencent {operation} returned a short page before "
                f"TotalNum was exhausted (page={page}, rows={len(batch)}, "
                f"total={total}); refusing a partial snapshot",
            )

    raise RuntimeError(
        f"Tencent {operation} exceeded the {max_pages}-page safety ceiling "
        "before TotalNum was exhausted; refusing a partial snapshot",
    )


def _bill_cycle_months(start_date: date, end_date: date) -> list[str]:
    """Every billing-cycle month (``YYYY-MM``) the window touches.

    Tencent attributes cost by **billing cycle** (``BillMonth``), and
    ``DescribeBillDetail`` accepts a ``Month`` parameter that returns
    exactly one cycle's line items — the figure that matches the Tencent
    console's 费用趋势 / 按计费周期 view (verified in prod: ``Month=2026-05``
    sums to the console's May total to the fen, ¥12,782.47).

    We page one cycle per month and attribute each row to its usage day
    (``_bill_detail_usage_day``) rather than its ``PayTime`` (扣费时间):
    monthly settlements (CDN / COS) are *paid* on the 1st of the FOLLOWING
    month, so grouping by ``PayTime`` dumped a whole month's settlement
    onto day 1 of the next month (e.g. May's ¥5,390 CDN settlement showed
    up under 2026-06-01, 35× over-stating June). ``Month`` + usage-day
    attribution keeps every charge in its own cycle and matches the bill.

    A ``DescribeBillDetail`` ``Month`` query supersedes ``BeginTime`` /
    ``EndTime`` (the two are mutually exclusive in the API), so the old
    ``_split_by_calendar_month`` cross-month workaround is no longer needed.
    """
    out: list[str] = []
    y, m = start_date.year, start_date.month
    while (y, m) <= (end_date.year, end_date.month):
        out.append(f"{y:04d}-{m:02d}")
        m += 1
        if m > 12:
            m = 1
            y += 1
    return out


def _bill_detail_usage_day(
    row: Any, cycle_year: int, cycle_month: int,
) -> date:
    """The day a ``BillDetail`` row is attributed to: its billing-cycle
    usage start (``FeeBeginTime``), matching the console 按计费周期 view.

    Falls back to ``BillDay`` then ``PayTime`` if ``FeeBeginTime`` is
    absent/unparseable. The result is clamped into the queried cycle month
    (first..last day) so the per-day rollup always sums to that cycle's
    ``Month`` total — a settlement whose ``FeeBeginTime`` lands a hair
    outside the cycle (a fee period spanning a month boundary) would
    otherwise leak into an adjacent month and break console parity.
    """
    parsed: date | None = None
    for attr in ("FeeBeginTime", "BillDay", "PayTime"):
        v = str(getattr(row, attr, "") or "")
        if len(v) >= 10:
            try:
                parsed = datetime.strptime(v[:10], "%Y-%m-%d").date()
                break
            except ValueError:
                continue
    cycle_first = date(cycle_year, cycle_month, 1)
    if cycle_month == 12:
        cycle_last = date(cycle_year, 12, 31)
    else:
        cycle_last = date(cycle_year, cycle_month + 1, 1) - timedelta(days=1)
    if parsed is None or parsed < cycle_first or parsed > cycle_last:
        return cycle_first
    return parsed


def _bill_detail_real_cost_cny(row: Any) -> float:
    """Sum the real (discount-applied) CNY cost of one ``DescribeBillDetail``
    ``BillDetail`` row.

    Tencent's ``BillDetail`` has **no** top-level ``RealTotalCost`` /
    ``TotalCost`` field — those belong to a *different* API
    (``DescribeBillSummaryByProduct`` → ``SummaryOverview``, used by
    ``providers/tencent.py`` for monthly rollups). On ``DescribeBillDetail``
    the billable amount lives one level down, in
    ``BillDetail.ComponentSet[]`` — each ``BillDetailComponent`` carries
    ``RealCost`` (组件优惠后总价 = post-discount real cost, CNY string). The
    row's real cost is the sum of its components' ``RealCost``.

    We deliberately read ONLY ``RealCost`` and never fall back to the
    component's ``Cost`` (官网价/list price): ``RealCost`` is the
    authoritative discount-applied amount the customer is actually billed
    and Tencent always populates it (as a string, ``"0"`` for a zero-cost
    component). Substituting list price for a (vanishingly rare) missing
    ``RealCost`` would over-report spend on any discounted account by
    blending pre- and post-discount numbers in one row — so an unparseable
    component simply contributes 0.

    Why this helper exists: both billing methods previously read the
    non-existent ``getattr(row, "RealTotalCost") or getattr(row,
    "TotalCost") or "0"`` — the field name was copied from the
    SummaryOverview API. ``getattr`` returned ``None`` for every row, so it
    fell through to ``"0"``: ``billing_daily`` wrote all-zero days and
    ``cost_facts`` dropped every line via ``if native == 0``. Result: the
    Tencent账单/概览 hero, by-account ranking and cost facts were silently
    empty even though the connector authenticated and the API returned full
    line items (service names parsed fine — those are real top-level fields).
    """
    total = 0.0
    for comp in getattr(row, "ComponentSet", None) or []:
        try:
            total += float(getattr(comp, "RealCost", None))
        except (TypeError, ValueError):
            continue
    return total


@register_driver
class TencentCloudDriver(_CloudDriverBase):
    provider = "tencent"
    display_name = "腾讯云"
    description = (
        "腾讯云(CVM / DNSPod / CLB / CDN / COS / 域名注册 + CAM 子账号 + 账单)。"
        "默认 IAM 模式 — 单租户下建 CAM 子账号 + 挂策略 + 离职删除账号。"
    )
    provider_cls = TencentProvider
    supported_capabilities = frozenset({
        "servers", "domains", "dns_records", "load_balancers",
        "cdn", "object_storage", "iam", "user_revoke",
    })
    capabilities = [
        "servers", "domains", "dns_records", "load_balancers",
        "cdn", "object_storage", "iam", "user_revoke",
    ]
    # ``account_mode`` is computed dynamically from config via the
    # @property below (defaults to ``iam``). No class-level literal.
    # Tencent CAM is single-tenant: CAM sub-accounts and their attached
    # policies are tenant-wide (no multi-account concept here). Run
    # grants-sync once per integration.
    grants_scope_wide = False
    # Tencent CAM's system policy for full access. Custom admin-equivalent
    # policies should be added by the admin via platform_settings if a
    # tenant renames it.
    owner_role_keys = frozenset({"AdministratorAccess"})

    # Role governance (see app/integrations/cloud_role_governance.py).
    # AdministratorAccess (above) is owner-equivalent. The CAM (IAM)
    # role/policy-management presets below all grant privilege escalation
    # — the holder can mint a CAM role, attach AdministratorAccess to it,
    # then assume it (or authorize it onto a sub-account). None of them
    # carry the "admin"/"管理员" keyword, so WITHOUT this explicit set the
    # central classifier would only see the "FullAccess" token and land
    # them at ``elevated`` (employee-visible + self-requestable, just
    # admin-approved). Listing them here promotes them to ``restricted``
    # (hidden from the employee dropdown, admin-only) — the correct posture
    # for an escalation primitive. Confirmed present in the prod Tencent
    # CAM preset-policy catalog (2026-06-08 audit). Other Qcloud*FullAccess
    # (CVM/COS/...) are bounded service roles and correctly stay elevated.
    restricted_role_keys = frozenset({
        "QcloudCamFullAccess",                       # full CAM (IAM) control
        "QcloudCamRoleFullAccess",                   # create/manage CAM roles → escalation
        "QcloudCamSubaccountsAuthorizeRoleFullAccess",  # authorize roles onto sub-accounts → escalation
    })
    # CNY-native billing: list_cost_facts needs a CNY→USD rate.
    requires_cny_fx = True
    config_schema = [
        IntegrationFieldInfo(
            key="secret_id", label="SecretId",
            is_secret=True, required=True,
        ),
        IntegrationFieldInfo(
            key="secret_key", label="SecretKey",
            is_secret=True, required=True,
        ),
        IntegrationFieldInfo(
            key="regions", label="Regions(逗号分隔,CVM/CLB 多 Region 同步用)",
            is_secret=False, required=False,
            default=(
                "ap-guangzhou,ap-shanghai,ap-beijing,ap-chengdu,ap-nanjing,"
                "ap-hongkong,ap-singapore,ap-tokyo,ap-seoul,na-siliconvalley,"
                "na-ashburn,eu-frankfurt"
            ),
            placeholder="ap-shanghai,ap-beijing",
        ),
        IntegrationFieldInfo(
            key="account_mode", label="账号模式(iam / sso)",
            is_secret=False, required=False,
            default="iam",
            placeholder="iam",
        ),
        # Optional per-integration FX override. Normally the platform
        # uses ``platform_settings.ai_spending.cny_to_usd_rate``
        # (refreshed weekly from frankfurter.dev / ECB). Set this only
        # if finance needs a fixed monthly rate that's independent of
        # the live rate — leaving it blank routes through the
        # platform value. See app.core.fx_rate.resolve_cny_per_usd.
        IntegrationFieldInfo(
            key="cny_per_usd",
            label="CNY→USD 汇率覆盖 (可选,留空走平台值)",
            is_secret=False, required=False,
            placeholder="7.20",
        ),
    ]

    @property
    def account_mode(self) -> str:  # type: ignore[override]
        return str(self.config.get("account_mode") or "iam").strip() or "iam"

    # ── Throttle helper ───────────────────────────────────────
    #
    # All sync-path SDK calls funnel through ``_call`` so the per-
    # credential token bucket in ``_tencent_throttle`` governs every
    # CAM / billing API hit. This is the difference between
    # ``cmdb.grants_sync`` running cleanly on a 50-user tenant vs the
    # 2026-05-09 incident where the per-user
    # ``ListAttachedUserAllPolicies`` loop trivially exceeded the
    # documented 5 RPS CAM ceiling and aborted mid-snapshot (which a
    # mark-and-sweep grants_sync interprets as "cloud removed
    # everything → revoke" — a mass-revoke hazard).

    def _secret_id(self) -> str:
        return str(self.config.get("secret_id") or "")

    def _call(self, fn: Any, *args: Any, service: str = "cam", **kwargs: Any) -> Any:
        return _tc_call(
            fn, *args,
            secret_id=self._secret_id(),
            service=service,
            **kwargs,
        )

    # ── Health probe ──────────────────────────────────────────

    async def test_connection(self) -> IntegrationTestResult:
        """Lightweight auth probe.

        ``_CloudDriverBase.test_connection`` would call
        ``provider.list_servers()`` which iterates every configured
        region (15 by default) — too heavy for an admin-UI "test
        connection" click. We probe CAM ``GetUserAppId`` instead: one
        call, instant response, same credential surface as the sync
        path. Falls through to the base implementation if the SDK is
        somehow unavailable.
        """
        if not _TC_AVAILABLE:
            return IntegrationTestResult(
                ok=False,
                message="tencentcloud-sdk-python is not installed",
            )
        return await asyncio.to_thread(self._test_connection_sync)

    def _test_connection_sync(self) -> IntegrationTestResult:
        try:
            uin, owner_uin, app_id = self._fetch_account_identity_sync()
            return IntegrationTestResult(
                ok=True,
                message=f"OK · OwnerUin={owner_uin} AppId={app_id}",
                detail={
                    "uin": uin,
                    "owner_uin": owner_uin,
                    "app_id": app_id,
                },
            )
        except Exception as exc:  # noqa: BLE001
            code = str(getattr(exc, "code", "") or "")
            return IntegrationTestResult(
                ok=False,
                message=(
                    f"{code or type(exc).__name__}: "
                    f"{redact_provider_error(exc)}"
                ),
                detail={"code": code},
            )

    # ── Client factories ──────────────────────────────────────

    def _tc_credential(self) -> Any:
        if not _TC_AVAILABLE:
            raise RuntimeError("tencentcloud-sdk-python is not installed")
        secret_id = str(self.config.get("secret_id") or "").strip()
        secret_key = str(self.config.get("secret_key") or "").strip()
        # Fail-closed on missing creds so we never produce an empty
        # Credential that would either fall through to the SDK's
        # ambient credential chain or hit Tencent with anonymous
        # request signing — both result in opaque errors and, in the
        # ambient case, can target the wrong account. (Round-17 audit.)
        if not secret_id or not secret_key:
            raise RuntimeError(
                "Tencent integration is missing secret_id / secret_key "
                "— refusing to construct a Credential with empty values."
            )
        return tc_credential.Credential(  # type: ignore[union-attr]
            secret_id, secret_key,
        )

    def _cam_client(self) -> Any:
        cred = self._tc_credential()
        http = HttpProfile()  # type: ignore[misc]
        http.endpoint = "cam.tencentcloudapi.com"
        http.reqTimeout = 30
        profile = ClientProfile()  # type: ignore[misc]
        profile.httpProfile = http
        return cam_client.CamClient(cred, "", profile)  # type: ignore[union-attr]

    def _billing_client(self) -> Any:
        cred = self._tc_credential()
        http = HttpProfile()  # type: ignore[misc]
        http.endpoint = "billing.tencentcloudapi.com"
        http.reqTimeout = 30
        profile = ClientProfile()  # type: ignore[misc]
        profile.httpProfile = http
        return billing_client.BillingClient(cred, "", profile)  # type: ignore[union-attr]

    # ── Scopes ────────────────────────────────────────────────

    async def list_scopes(self) -> list[CloudScopeInfo]:
        """Return the single tenant-scope row for this Tencent integration.

        ``external_id`` MUST be the real cloud-side account identifier
        (Tencent ``OwnerUin`` — the numeric tenant id) rather than the Helix
        ``Integration.id`` UUID. AWS and Aliyun fixed exactly this in
        2026-04-23 ("ghost scope rows that collected stale grants until
        manual cleanup"); without the same fix, every Tencent grant /
        billing row joined on a Helix UUID, downstream auto-adoption of
        owners + cross-account integration silently targeted the wrong
        identifier, and a credential rotation that recreated the
        Integration row produced a fresh UUID that disagreed with every
        prior grant — orphan rows.

        On STS-style failure we refuse to write a scope row (mirror
        AWS / Aliyun's ``ProviderAuthError`` shape) so a misconfigured
        AK doesn't silently bind to the integration UUID and re-create
        the original bug.
        """
        if not _TC_AVAILABLE:
            raise RuntimeError("tencentcloud-sdk-python is not installed")
        credential_uin, owner_uin, _app_id = await asyncio.to_thread(
            self._fetch_account_identity_sync,
        )
        alias = str(self.config.get("_alias") or f"tencent-{owner_uin}")
        return [CloudScopeInfo(
            scope_type="standalone_iam",
            external_id=owner_uin,
            display_name=alias,
            status="active",
            raw={
                "mode": "iam",
                "owner_uin": owner_uin,
                "credential_uin": credential_uin,
            },
        )]

    def _fetch_account_identity_sync(self) -> tuple[str, str, int | None]:
        """Resolve current credential and owning tenant via GetUserAppId.

        ``Uin`` identifies the current credential account; ``OwnerUin`` is
        the stable tenant identity.  They differ for CAM sub-user keys, so
        only OwnerUin may key the Helix scope.
        """
        try:
            client = self._cam_client()
            req = cam_models.GetUserAppIdRequest()  # type: ignore[misc]
            resp = self._call(client.GetUserAppId, req)
            credential_uin = str(getattr(resp, "Uin", "") or "").strip()
            owner_uin = str(getattr(resp, "OwnerUin", "") or "").strip()
            app_id_raw = getattr(resp, "AppId", None)
            app_id = int(app_id_raw) if app_id_raw is not None else None
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "Tencent GetUserAppId failed (cannot derive tenant identity): %s",
                redact_provider_error(exc),
            )
            raise ProviderAuthError(
                "Tencent CAM GetUserAppId failed — cannot resolve the "
                "owning tenant; verify cam:GetUserAppId and credentials: "
                f"{redact_provider_error(exc)}",
            ) from exc
        if not credential_uin or not owner_uin:
            raise ProviderAuthError(
                "Tencent CAM GetUserAppId omitted Uin or OwnerUin; refusing "
                "to key a tenant scope from an ambiguous account identity",
            )
        return credential_uin, owner_uin, app_id

    # ── Role catalog: CAM policies ───────────────────────────

    async def list_scope_roles(
        self, scope: CloudScopeRef,
    ) -> list[CloudRoleInfo]:
        from app.integrations.drivers._cloud_defaults import (
            load_default_roles,
        )
        defaults = await load_default_roles(
            "cloud.tencent.default_policies", _TENCENT_DEFAULT_FALLBACK,
        )
        promoted_keys: set[str] = {
            str(e.get("role_key") or "").strip()
            for e in defaults
            if (e.get("role_key") or "").strip()
        }
        return await asyncio.to_thread(
            self._list_policies_sync, scope, promoted_keys,
        )

    def _list_policies_sync(
        self,
        scope: CloudScopeRef,
        promoted_keys: set[str],
    ) -> list[CloudRoleInfo]:
        """Curated presets (promoted) + every Custom CAM policy
        (promoted) + every other preset (NOT promoted).

        Tencent's CAM has roughly ~200 ``Type=2`` (preset) policies
        that historically all dumped into the request modal. Now only
        the curated subset is promoted; Custom (admin-defined) policies
        are always promoted. Everything stays in the catalog so the
        admin promote/demote UI can later widen the curated set.
        """
        client = self._cam_client()
        page_size = 200  # CAM ListPolicies max Rp

        def _fetch(page: int) -> Any:
            req = cam_models.ListPoliciesRequest()  # type: ignore[misc]
            req.Rp = page_size
            req.Page = page
            return self._call(client.ListPolicies, req)

        policies = _collect_authoritative_cam_pages(
            fetch_page=_fetch,
            items_attr="List",
            page_size=page_size,
            max_pages=_MAX_CAM_POLICY_PAGES,
            item_key=lambda policy: str(
                getattr(policy, "PolicyName", "") or "",
            ).strip() or None,
            item_identity="PolicyName",
            operation="ListPolicies",
        )
        out: list[CloudRoleInfo] = []
        for p in policies:
            # Tencent CAM ``ListPolicies`` Type field:
            #   1 = 自定义 (custom, admin-created in console)
            #   2 = 预设 (system-managed preset)
            # CreateMode: 1 = 可视化编辑创建, 2 = 按产品或项目权限创建.
            # ``Type`` is the source-of-truth; default to 1 if absent
            # so an unknown policy surfaces as custom (safer — admin
            # sees it in the role catalog).
            policy_type_raw = getattr(p, "Type", 1)
            is_custom = int(policy_type_raw) == 1
            policy_name = str(getattr(p, "PolicyName", "") or "")
            out.append(CloudRoleInfo(
                role_key=policy_name,
                display_name=policy_name,
                description=str(getattr(p, "Description", "") or "") or None,
                is_custom=is_custom,
                source="tencent_custom" if is_custom else "tencent_preset",
                scope=scope,
                raw={"policy_id": getattr(p, "PolicyId", None)},
                is_promoted=is_custom or policy_name in promoted_keys,
            ))
        return out

    def _resolve_policy_id(self, client: Any, policy_name: str) -> int | None:
        """Look up CAM PolicyId for a given PolicyName.

        CAM enforces name uniqueness within a tenant (presets share the
        namespace with custom, first-writer wins on conflicts), so a
        name-only lookup is safe. Called on grant/revoke where the
        role_key is the PolicyName.
        """
        page_size = 200

        def _fetch(page: int) -> Any:
            req = cam_models.ListPoliciesRequest()  # type: ignore[misc]
            req.Rp = page_size
            req.Page = page
            req.Keyword = policy_name  # server-side filter to short-circuit
            return self._call(client.ListPolicies, req)

        policies = _collect_authoritative_cam_pages(
            fetch_page=_fetch,
            items_attr="List",
            page_size=page_size,
            max_pages=10,
            item_key=lambda policy: str(
                getattr(policy, "PolicyName", "") or "",
            ).strip() or None,
            item_identity="PolicyName",
            operation=f"ListPolicies Keyword={policy_name!r}",
        )
        for policy in policies:
            name = str(getattr(policy, "PolicyName", "") or "")
            if name != policy_name:
                continue
            policy_id = getattr(policy, "PolicyId", None)
            try:
                if policy_id is None:
                    raise TypeError("PolicyId is missing")
                return int(policy_id)
            except (TypeError, ValueError) as exc:
                raise RuntimeError(
                    "Tencent ListPolicies returned the requested policy "
                    f"{policy_name!r} without a valid PolicyId",
                ) from exc
        return None

    def _list_attached_policies_authoritative(
        self,
        client: Any,
        *,
        target_uin: int,
        attach_type: int,
        context: str,
    ) -> list[Any]:
        page_size = 200

        def _fetch(page: int) -> Any:
            request = cam_models.ListAttachedUserAllPoliciesRequest()  # type: ignore[misc]
            request.TargetUin = target_uin
            request.AttachType = attach_type
            request.Page = page
            request.Rp = page_size
            return self._call(client.ListAttachedUserAllPolicies, request)

        def _policy_key(policy: Any) -> tuple[int, str] | None:
            try:
                policy_id = int(getattr(policy, "PolicyId", 0) or 0)
            except (TypeError, ValueError):
                policy_id = 0
            policy_name = str(getattr(policy, "PolicyName", "") or "").strip()
            if not policy_name:
                return None
            return policy_id, policy_name

        return _collect_authoritative_cam_pages(
            fetch_page=_fetch,
            items_attr="PolicyList",
            page_size=page_size,
            max_pages=_MAX_CAM_ATTACHED_POLICY_PAGES,
            item_key=_policy_key,
            item_identity="PolicyId and PolicyName",
            operation=f"ListAttachedUserAllPolicies {context}",
            allow_same_page_duplicates=attach_type == 0,
        )

    def _resolve_user_ids(
        self, client: Any, user_name: str,
    ) -> tuple[int, int] | None:
        """Look up (Uid, Uin) for a CAM sub-account by Name.

        Tencent CAM exposes two numeric identifiers per sub-account:

          * ``Uin`` — sub-account UIN, plain-number identifier used by
            ``DetachUserPolicy.DetachUin`` / ``AttachUserPolicy.AttachUin``
            / ``ListAttachedUserAllPolicies.TargetUin``.
          * ``Uid`` — CAM-internal identifier (different from Uin) used
            by ``ListGroupsForUser.Uid`` / ``RemoveUserFromGroup.Info[].Uid``.

        Both come back on the ``GetUser`` response. Returns ``None`` only
        when Tencent explicitly reports that the user is absent
        (idempotent success for the caller); raises for malformed identity
        responses and every other SDK error so failures cannot be confused
        with a completed revoke. Background: a Uid→Uin mix-up
        ate the 2026-06-01 audit of lifu.wu@'s revoke attempt — the
        first patch passed Uin where the API wanted Uid, ``ListGroupsForUser``
        returned ``ResourceNotFound.UserNotExist`` (lying!) and the
        loop appeared to have nothing to remove.
        """
        try:
            req = cam_models.GetUserRequest()  # type: ignore[misc]
            req.Name = user_name
            resp = self._call(client.GetUser, req)
        except TencentCloudSDKException as exc:
            code = str(getattr(exc, "code", "") or "")
            if "NotExist" in code or "UserNotExist" in code:
                return None
            raise
        try:
            uid = int(getattr(resp, "Uid", 0) or 0)
            uin = int(getattr(resp, "Uin", 0) or 0)
        except (TypeError, ValueError) as exc:
            raise RuntimeError(
                f"Tencent GetUser returned invalid identifiers for {user_name!r}",
            ) from exc
        if not uid or not uin:
            raise RuntimeError(
                f"Tencent GetUser returned missing identifiers for {user_name!r}",
            )
        return uid, uin

    def _resolve_uin(self, client: Any, user_name: str) -> int | None:
        """Legacy single-value UIN lookup. New code should use
        :meth:`_resolve_user_ids` so it has Uid for group ops too."""
        ids = self._resolve_user_ids(client, user_name)
        return ids[1] if ids else None

    @staticmethod
    def _coerce_group_dict(item: Any) -> dict[str, Any] | None:
        """``ListAttachedUserAllPolicies`` returns ``Groups`` as a list
        whose element types vary across SDK versions / response paths —
        sometimes a JSON-string-encoded dict (older SDKs), sometimes a
        plain dict (deserialized), sometimes an SDK model object.
        Coerce all three to a plain dict so the caller can read GroupId.
        """
        if item is None:
            return None
        if isinstance(item, dict):
            return item
        if isinstance(item, str):
            try:
                import json
                return json.loads(item)
            except (ValueError, TypeError):
                return None
        gid = getattr(item, "GroupId", None)
        gname = getattr(item, "GroupName", None)
        if gid is None and gname is None:
            return None
        return {"GroupId": gid, "GroupName": gname}

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

    def _cam_user_name_for(self, email: str) -> str:
        local = email.split("@", 1)[0]
        sanitized = "".join(
            c for c in local
            if c.isascii() and (c.isalnum() or c in "-._")
        )
        if sanitized:
            return sanitized[:64]
        # A fully non-ASCII local-part (e.g. ``张三@x.com``) collapses to
        # empty after the ASCII filter. Mirror the Aliyun driver's sha1-
        # based fallback so the mapping stays 1:1 and deterministic across
        # runs — returning the literal ``helix-user`` would collide when
        # two employees share the same non-ASCII pattern.
        import hashlib
        # Compatibility identity, not secret protection: this value is part
        # of already-provisioned CAM usernames and must remain stable.
        digest = hashlib.sha1(
            email.encode("utf-8"),
            usedforsecurity=False,
        ).hexdigest()[:12]
        return f"helix-{digest}"

    @staticmethod
    def _cam_phone_fields(ctx: CloudAccessContext) -> tuple[str, str] | None:
        """Return CAM's national-number + country-code fields when safe.

        CAM's sensitive operations (including access-key creation) require a
        phone number on the sub-user.  Helix keeps the HRIS value in its
        identity record, but CAM expects it split into ``PhoneNum`` and
        ``CountryCode`` rather than E.164.  Missing or ambiguous HRIS data is
        deliberately omitted: an incorrect verification number is worse than
        asking the user to have IT correct the directory record.
        """
        raw_phone = str(ctx.extra.get("mobile_phone") or "")
        country = str(ctx.extra.get("mobile_country_code") or "")
        country_code = country_dialing_code(country)
        e164 = to_e164(raw_phone, country)
        if not country_code or not e164:
            return None
        prefix = f"+{country_code}"
        if not e164.startswith(prefix):
            # A manually-entered international number whose country does not
            # match the HRIS country is ambiguous. Never attach it to CAM.
            return None
        national_number = e164.removeprefix(prefix)
        return (national_number, country_code) if national_number else None

    async def grant_access(
        self, ctx: CloudAccessContext,
    ) -> CloudGrantResult:
        return await asyncio.to_thread(self._grant_sync, ctx)

    async def revoke_access(
        self, ctx: CloudAccessContext,
    ) -> CloudGrantResult:
        return await asyncio.to_thread(self._revoke_sync, ctx)

    def _grant_sync(self, ctx: CloudAccessContext) -> CloudGrantResult:
        try:
            client = self._cam_client()
        except RuntimeError as e:
            return CloudGrantResult(
                ok=False, error_message=redact_provider_error(e),
            )
        user_name = self._cam_user_name_for(ctx.principal_email)
        policy_name = ctx.role_key

        # Legacy: pre-2026-04 rows stored PolicyId-as-string in role_key.
        # Try to int-parse first; if it works treat it as an id directly,
        # otherwise fall back to the PolicyName → id lookup below.
        policy_id: int | None
        try:
            policy_id = int(policy_name)
        except (TypeError, ValueError):
            try:
                policy_id = self._resolve_policy_id(client, policy_name)
            except Exception as exc:  # noqa: BLE001
                return CloudGrantResult(
                    ok=False, error_message=redact_provider_error(exc),
                )
            if policy_id is None:
                return CloudGrantResult(
                    ok=False,
                    error_message=(
                        f"Tencent: CAM policy {policy_name!r} not found in "
                        "this tenant (run roles_sync first)"
                    ),
                )

        # Create user (idempotent). Capture the freshly-allocated UIN
        # from the response when the user is brand new — saves a GetUser
        # round-trip on the AttachUserPolicy call below.
        created_user = False
        uin: int | None = None
        try:
            add_req = cam_models.AddUserRequest()  # type: ignore[misc]
            add_req.Name = user_name
            add_req.Remark = f"helix managed — {ctx.principal_email}"
            add_req.ConsoleLogin = 1
            # A role grant needs console access, not an unused long-lived
            # API credential. Tencent returns a generated initial password
            # when omitted; it is passed ephemerally to secure delivery.
            add_req.UseApi = 0
            add_req.NeedResetPassword = 1
            phone_fields = self._cam_phone_fields(ctx)
            if phone_fields is not None:
                add_req.PhoneNum, add_req.CountryCode = phone_fields
            add_resp = self._call(client.AddUser, add_req)
            created_user = True
            add_uin = getattr(add_resp, "Uin", None) or getattr(add_resp, "Uid", None)
            if add_uin is not None:
                try:
                    uin = int(add_uin)
                except (TypeError, ValueError):
                    uin = None
        except TencentCloudSDKException as exc:
            code = str(getattr(exc, "code", "") or "")
            # CAM currently reports duplicate sub-account names as
            # ``InvalidParameter.SubUserNameInUse``. Treat it exactly
            # like the older ExistedUser/UserExist variants: resolve the
            # existing account and continue attaching the requested policy.
            if (
                "ExistedUser" in code
                or "UserExist" in code
                or "SubUserNameInUse" in code
            ):
                created_user = False
            else:
                return CloudGrantResult(
                    ok=False, error_message=redact_provider_error(exc),
                )

        # Resolve UIN for the existing-user path (AddUser short-circuited
        # via ExistedUser). SDK 3.1.79's ``AttachUserPolicyRequest``
        # carries only ``PolicyId`` + ``AttachUin`` — never set
        # ``AttachUserName`` (becomes an ad-hoc attribute that serializes
        # mangled as ``EtachUserName`` → API rejects with
        # ``UnknownParameter``, parity with the Detach bug fixed in the
        # 2026-06-01 incident).
        if uin is None:
            try:
                uin = self._resolve_uin(client, user_name)
            except Exception as exc:  # noqa: BLE001
                return CloudGrantResult(
                    ok=False, error_message=redact_provider_error(exc),
                )
        if uin is None:
            return CloudGrantResult(
                ok=False,
                error_message=(
                    f"Tencent: cannot resolve Uin for CAM user "
                    f"{user_name!r} after AddUser — refusing to attach "
                    "policy without identifier."
                ),
            )

        # Attach policy
        try:
            at = cam_models.AttachUserPolicyRequest()  # type: ignore[misc]
            at.PolicyId = policy_id
            at.AttachUin = uin
            self._call(client.AttachUserPolicy, at)
        except TencentCloudSDKException as exc:
            code = str(getattr(exc, "code", "") or "")
            if "PolicyExist" in code or "AlreadyAttached" in str(exc):
                return CloudGrantResult(
                    ok=True, already_exists=True,
                    external_ref=f"{user_name}::{policy_id}::{policy_name}",
                    detail={
                        "user_name": user_name,
                        "policy_id": policy_id,
                        "policy_name": policy_name,
                        "created": created_user,
                    },
                )
            return CloudGrantResult(
                ok=False, error_message=redact_provider_error(exc),
            )
        detail: dict[str, Any] = {
            "user_name": user_name,
            "policy_id": policy_id,
            "policy_name": policy_name,
            "created": created_user,
        }
        if created_user:
            initial_password = str(getattr(add_resp, "Password", "") or "")
            if initial_password:
                detail["initial_console_password"] = initial_password
        return CloudGrantResult(
            ok=True,
            external_ref=f"{user_name}::{policy_id}::{policy_name}",
            detail=detail,
        )

    def _revoke_sync(self, ctx: CloudAccessContext) -> CloudGrantResult:
        try:
            client = self._cam_client()
        except RuntimeError as e:
            return CloudGrantResult(
                ok=False, error_message=redact_provider_error(e),
            )

        # Parse existing_external_ref. Supported shapes:
        #   user_name::policy_id::policy_name      — current convention
        #   user_name::policy_id                   — legacy (pre-rename)
        user_name: str
        policy_id: int | None = None
        if ctx.existing_external_ref and ctx.existing_external_ref.count("::") >= 1:
            parts = ctx.existing_external_ref.split("::")
            user_name = parts[0]
            if len(parts) >= 2:
                try:
                    policy_id = int(parts[1])
                except ValueError:
                    policy_id = None
        else:
            user_name = self._cam_user_name_for(ctx.principal_email)

        if policy_id is None:
            # Resolve from role_key (PolicyName) or int string
            try:
                policy_id = int(ctx.role_key)
            except (TypeError, ValueError):
                try:
                    policy_id = self._resolve_policy_id(client, ctx.role_key)
                except Exception as exc:  # noqa: BLE001
                    return CloudGrantResult(
                        ok=False, error_message=redact_provider_error(exc),
                    )
                if policy_id is None:
                    # Policy already gone → the attachment is effectively
                    # revoked (CAM removes attachments when policies die).
                    return CloudGrantResult(ok=True, already_absent=True)

        # Group-aware revoke. Tencent CAM policies attach to users two
        # ways:
        #   (1) DIRECT — user-policy pair via AttachUserPolicy
        #   (2) GROUP-INHERITED — user joins a group, group has policies
        #
        # Helix's grants_sync flattens both into per-(user, policy)
        # ``cmdb_cloud_access_grants`` rows. Revoke must pick the right
        # detach API or the cloud-side attachment lingers forever:
        #
        #   * ``DetachUserPolicy`` only undoes DIRECT attachments. Called
        #     against a group-inherited attachment, it returns RequestId
        #     (looks successful) but silently does nothing — the policy
        #     stays attached via the group. This is the 2026-06-01 prod
        #     incident: lifu.wu@'s 9 grants and chujie@'s 4 grants were
        #     ALL group-inherited (管理员 / 账单计费 / EE groups), but
        #     T+0 cascade called DetachUserPolicy → silent no-op →
        #     cloud-side權限 残留 for 2 months until manual cleanup.
        #
        #   * ``RemoveUserFromGroup`` is the only API that revokes a
        #     group-inherited policy. It removes the user from the
        #     ENTIRE group — every policy that group confers on the
        #     user goes away with it. For offboarding the side effect
        #     is correct (a leaver should lose all group-conferred
        #     access); for admin single-grant revoke the UI must warn.
        #
        # Algorithm:
        #   1. Resolve (Uid, Uin) via GetUser. User-absent → idempotent
        #      ``already_absent``.
        #   2. ``ListAttachedUserAllPolicies(TargetUin=uin, AttachType=2)``
        #      to find the source group(s) for this policy. If found →
        #      ``RemoveUserFromGroup(Uid, GroupId)`` per group.
        #   3. Otherwise try ``AttachType=1`` (direct attach) → if
        #      this policy is in the list, ``DetachUserPolicy(Uin, PolicyId)``.
        #   4. Policy absent from both lists → ``already_absent``.
        try:
            ids = self._resolve_user_ids(client, user_name)
        except Exception as exc:  # noqa: BLE001
            return CloudGrantResult(
                ok=False, error_message=redact_provider_error(exc),
            )
        if ids is None:
            return CloudGrantResult(ok=True, already_absent=True)
        uid, uin = ids

        # Step 2: group-inherited path
        try:
            group_policies = self._list_attached_policies_authoritative(
                client,
                target_uin=uin,
                attach_type=2,
                context=f"during group revoke for user={user_name!r}",
            )
        except Exception as exc:  # noqa: BLE001
            return CloudGrantResult(
                ok=False, error_message=redact_provider_error(exc),
            )
        source_group_ids: list[int] = []
        for p in group_policies:
            try:
                pid = int(getattr(p, "PolicyId", 0) or 0)
            except (TypeError, ValueError):
                continue
            if pid != policy_id:
                continue
            for raw_group in (getattr(p, "Groups", None) or []):
                g = self._coerce_group_dict(raw_group)
                if not g:
                    continue
                try:
                    gid = int(g.get("GroupId") or 0)
                except (TypeError, ValueError):
                    continue
                if gid:
                    source_group_ids.append(gid)
        if source_group_ids:
            for gid in source_group_ids:
                try:
                    rm = cam_models.RemoveUserFromGroupRequest()  # type: ignore[misc]
                    rm.Info = [{"Uid": uid, "GroupId": gid}]
                    self._call(client.RemoveUserFromGroup, rm)
                except TencentCloudSDKException as exc:
                    code = str(getattr(exc, "code", "") or "")
                    if (
                        "NotExist" in code
                        or "NotInGroup" in code
                        or "UserNotInGroup" in str(exc)
                    ):
                        continue
                    return CloudGrantResult(
                        ok=False, error_message=redact_provider_error(exc),
                    )
            return CloudGrantResult(
                ok=True,
                external_ref=ctx.existing_external_ref,
                detail={
                    "revoke_path": "group_remove",
                    "groups_removed": source_group_ids,
                    "side_effect_warning": (
                        "user was removed from "
                        f"{len(source_group_ids)} group(s) — every other "
                        "policy those groups confer on this user is also "
                        "now revoked. This is correct for offboarding; "
                        "for admin single-grant revoke surface this on "
                        "the confirmation dialog."
                    ),
                },
            )

        # Step 3: direct-attach path
        try:
            direct_policies = self._list_attached_policies_authoritative(
                client,
                target_uin=uin,
                attach_type=1,
                context=f"during direct revoke for user={user_name!r}",
            )
        except Exception as exc:  # noqa: BLE001
            return CloudGrantResult(
                ok=False, error_message=redact_provider_error(exc),
            )
        direct_ids: set[int] = set()
        for p in direct_policies:
            try:
                pid = int(getattr(p, "PolicyId", 0) or 0)
            except (TypeError, ValueError):
                continue
            direct_ids.add(pid)
        if policy_id not in direct_ids:
            # Step 4: policy not attached at all → idempotent success
            return CloudGrantResult(ok=True, already_absent=True)
        try:
            de = cam_models.DetachUserPolicyRequest()  # type: ignore[misc]
            de.PolicyId = policy_id
            de.DetachUin = uin
            # NOTE: NOT setting ``DetachUserName`` — SDK 3.1.79 doesn't
            # accept it as a real field; assigning to it becomes an
            # ad-hoc attribute that serializes mangled (``EtachUserName``,
            # leading-char drop bug) and the API rejects with
            # ``UnknownParameter``. ``DetachUin`` alone is the correct
            # identifier for direct-attach revoke.
            self._call(client.DetachUserPolicy, de)
        except TencentCloudSDKException as exc:
            code = str(getattr(exc, "code", "") or "")
            if (
                "NotExist" in code
                or "NotAttached" in str(exc)
                or "UserNotExist" in code
            ):
                return CloudGrantResult(ok=True, already_absent=True)
            return CloudGrantResult(
                ok=False, error_message=redact_provider_error(exc),
            )
        return CloudGrantResult(
            ok=True,
            external_ref=ctx.existing_external_ref,
            detail={"revoke_path": "direct_detach"},
        )

    # ── Last login: CAM GetUser ────────────────────────────

    async def get_last_login(
        self, ctx: CloudAccessContext,
    ) -> datetime | None:
        return await asyncio.to_thread(
            self._get_last_login_sync,
            self._cam_user_name_for(ctx.principal_email),
        )

    def _get_last_login_sync(self, user_name: str) -> datetime | None:
        try:
            client = self._cam_client()
        except RuntimeError:
            return None
        try:
            req = cam_models.GetUserRequest()  # type: ignore[misc]
            req.Name = user_name
            resp = self._call(client.GetUser, req)
        except TencentCloudSDKException:
            return None
        # GetUser returns ``LastLoginTime`` (console login) — the access-key
        # last-used timestamp lives on the individual keys (separate API).
        # We use console login as the primary signal; if a CAM user only
        # uses programmatic access their last_login will be None, which
        # the inactivity sweep treats as "unknown" → skip (safe default).
        iso = getattr(resp, "LastLoginTime", None)
        if not iso:
            return None
        try:
            parsed = datetime.fromisoformat(str(iso).replace("Z", "+00:00"))
        except ValueError:
            for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
                try:
                    parsed = datetime.strptime(str(iso), fmt)
                    break
                except ValueError:
                    continue
            else:
                return None
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=UTC)
        # Sentinel epoch-zero = "never logged in" on some SDK variants.
        if parsed.year <= 1970:
            return None
        return parsed

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

    async def list_scope_grants(
        self, scope: CloudScopeRef,
    ) -> list[DiscoveredGrant]:
        """Tenant-wide (CAM sub-account × attached policy) enumeration.

        Each returned grant uses ``role_key = PolicyName`` so it matches
        the role catalog and the owner-role-keys owner-auto-adoption
        logic in ``cmdb.grants_sync``.
        """
        return await asyncio.to_thread(self._list_scope_grants_sync, scope)

    def _list_scope_grants_sync(
        self, scope: CloudScopeRef,
    ) -> list[DiscoveredGrant]:
        # Pre-launch P0 fix (mass-revoke safety): a CAM auth flap or
        # 5xx MUST raise, never return ``[]``. `grants_sync` treats an
        # empty list as "no grants exist on cloud" and revokes every
        # Helix-tracked grant. Same class as the 8th-audit Azure fix
        # and the matching Aliyun fix in this PR.
        client = self._cam_client()
        out: list[DiscoveredGrant] = []
        # CAM ``ListUsers`` does NOT accept ``Page`` / ``Rp`` parameters
        # despite the SDK exposing them on the request class. Passing
        # them returns ``UnknownParameter `P``` (Tencent's error parser
        # mangles unknown field names by dropping leading characters,
        # same shape as the ``Month` → `Onth`` bug we hit on the billing
        # API). The endpoint returns the full tenant user list in a
        # single response — single call, no pagination.
        req = cam_models.ListUsersRequest()  # type: ignore[misc]
        # Defensive: older SDK versions defaulted ``_Page`` / ``_Rp``
        # in ``__init__``, which makes the API reject with the
        # ``UnknownParameter `P``` shape (see ``providers/tencent.py``
        # for the same fix). Strip them before the request goes out.
        for _attr in ("_Page", "_Rp"):
            if hasattr(req, _attr):
                setattr(req, _attr, None)
        resp = self._call(client.ListUsers, req)
        users = list(getattr(resp, "Data", None) or [])
        for u in users:
            user_name = str(getattr(u, "Name", "") or "")
            email_attr = str(getattr(u, "Email", "") or "").strip().lower()
            if not user_name:
                raise RuntimeError(
                    "Tencent ListUsers returned a user without Name; "
                    "refusing a partial grants snapshot",
                )
            # Identity key, in the order the ``DiscoveredGrant`` contract
            # specifies for IAM-mode clouds: the CAM ``Email`` attribute
            # when the admin set it, else the CAM user name itself — which
            # on this tenant is usually already an address
            # (``lilin@pingcap.com``) and sometimes a bare handle
            # (``zhuoran.li``, ``tiup-server``). The resolver in
            # ``_apply_discovered_grants`` matches bare handles against the
            # work_email local-part and counts whatever is left as
            # unresolved.
            #
            # This used to raise for the bare-handle case ("refusing a
            # partial grants snapshot"). That guard was aimed at the real
            # hazard — a snapshot that OMITS a user, which the drift sweep
            # then reads as "their access is gone" — but a user with no
            # email attribute is not an omission, and raising took the
            # WHOLE tenant's snapshot down with it: prod
            # ``cmdb.grants.tencent-primary`` failed on every run from
            # 2026-08-03 to 2026-08-12 (18 consecutive) on the service
            # account ``tiup-server``, freezing the Tencent members list at
            # its 2026-08-01 state. Emitting the handle keeps the snapshot
            # COMPLETE (which is what protects the sweep) and downgrades an
            # unmappable principal to what it actually is: one row nobody
            # can attribute, counted and named in the sync summary.
            email = email_attr or user_name.lower()
            uin = getattr(u, "Uin", None)
            # Pre-launch P0 fix (per-user partial-snapshot safety):
            # a transient SDK exception MUST raise to the caller. The
            # previous ``except → continue`` produced a snapshot that
            # silently omitted user X's grants, which `grants_sync`
            # treats as "user X has no policies on cloud → revoke
            # every Helix-tracked grant for X". Same mass-revoke risk
            # class as the outer ListUsers fix above.
            # CAM ``ListAttachedUserAllPolicies`` REQUIRES ``Page`` (≥1)
            # and ``Rp`` (≤200) — see Tencent CAM API docs and the
            # production error code ``MissingParameter / 请求缺少必传参数
            # Page`` (incident 2026-05-16, request id
            # 93e4e7a1-26d9-4475-82ea-631a8f96ac7a). A previous code
            # comment claimed the API rejected Page with ``UnknownParameter
            # P`` — that error shape belongs to ``ListUsers`` (which
            # really does reject Page); the symmetric assumption was
            # mistakenly carried to this endpoint and the prod log
            # is authoritative. Page through with Rp=200 until either
            # the returned PolicyList is short OR we've collected
            # TotalNum entries; this covers tenants with >200 attached
            # policies that the old single-call path silently truncated
            # (mass-revoke risk if grants_sync interpreted the
            # truncation as drift). ``TotalNum`` is returned alongside
            # ``PolicyList`` per the SDK response shape.
            try:
                if uin is None:
                    raise TypeError("Uin is missing")
                tc_uin = int(uin)
            except (TypeError, ValueError) as exc:
                raise RuntimeError(
                    "Tencent ListUsers returned a user without a valid Uin "
                    f"(Name={user_name!r}); refusing a partial grants "
                    "snapshot",
                ) from exc
            if tc_uin <= 0:
                raise RuntimeError(
                    "Tencent ListUsers returned a non-positive Uin "
                    f"(Name={user_name!r}); refusing a partial grants "
                    "snapshot",
                )
            collected = self._list_attached_policies_authoritative(
                client,
                target_uin=tc_uin,
                attach_type=0,
                context=f"during grants sync for user={user_name!r}",
            )
            for p in collected:
                pname = str(getattr(p, "PolicyName", "") or "")
                pid = getattr(p, "PolicyId", None)
                out.append(DiscoveredGrant(
                    principal_email=email,
                    principal_type="user",
                    role_key=pname,  # matches grant_access + role catalog
                    role_display_name=pname,
                    external_ref=f"{user_name}::{pid}::{pname}",
                    scope=scope,
                    granted_at=None,
                    raw={
                        "user_name": user_name,
                        "uin": uin,
                        "policy_id": pid,
                        "policy_name": pname,
                        "policy_type": getattr(p, "PolicyType", None),
                    },
                ))
        return out

    # ── Account revocation (offboarding kill-switch) ──────────

    async def revoke_account(
        self, ctx: CloudAccountRevokeContext,
    ) -> ProvisioningResult:
        """Delete a CAM sub-account cleanly.

        ``account_id`` is the CAM ``Name``. For convenience, if an email
        is passed we derive the user name via the same rule as
        :meth:`_cam_user_name_for` (same logic as grant_access).

        Deletion order (CAM refuses ``DeleteUser`` while any of these
        exist):

          1. Detach every attached policy
          2. Remove from every group
          3. Delete all AK/SK access keys
          4. Delete console login profile (password)
          5. DeleteUser (with ``Force=1`` to remove MFA atomically)

        Each step tolerates ``NotExist`` / ``UserNotExist`` so a re-run
        on a partially-cleaned account still succeeds.
        """
        if not _TC_AVAILABLE:
            return ProvisioningResult(
                ok=False,
                error_message="tencentcloud-sdk-python not installed",
            )
        raw_id = (ctx.account_id or "").strip()
        if not raw_id:
            return ProvisioningResult(
                ok=False,
                error_message="Tencent revoke_account requires account_id (CAM user Name or email)",
            )
        return await asyncio.to_thread(self._revoke_account_sync, raw_id)

    def _revoke_account_sync(self, raw_id: str) -> ProvisioningResult:
        try:
            client = self._cam_client()
        except RuntimeError as exc:
            return ProvisioningResult(
                ok=False, error_message=redact_provider_error(exc),
            )

        steps: list[str] = []

        def _swallow_nse(label: str, fn, *args, **kwargs) -> None:
            try:
                self._call(fn, *args, **kwargs)
                steps.append(label)
            except TencentCloudSDKException as exc:
                code = str(getattr(exc, "code", "") or "")
                if (
                    "NotExist" in code
                    or "UserNotExist" in code
                    or "NotAttached" in str(exc)
                ):
                    return
                raise

        # Helper for the outer per-step try/except. We tolerate
        # arbitrary SDK errors per step (the next step is independent
        # work — leaving groups doesn't depend on detaching policies),
        # but throttle exhaustion MUST propagate so the operator sees
        # the real cause instead of a downstream "DeleteUser failed:
        # residual attachments" symptom (which is what this helper
        # avoided pre-throttle by silently leaving residue and letting
        # step 5 fail with a misleading message).
        def _is_throttle_exhausted(exc: TencentCloudSDKException) -> bool:
            code = str(getattr(exc, "code", "") or "")
            return code.startswith("RequestLimitExceeded")

        # Resolve which CAM user we are actually deleting, then harvest
        # BOTH ids (Uid for group ops, Uin for policy ops — see
        # ``_resolve_user_ids`` for the 2026-06-01 mix-up that hid this
        # delta for a release cycle).
        #
        # The handle is tried LITERALLY first. Deriving the local part
        # from an ``@``-shaped id unconditionally is what made this whole
        # chain a no-op on 17 of 22 live users of the prod tenant: the
        # derived name did not exist, ``UserNotExist`` was read as
        # "already absent", and the caller was told ok=True while the
        # real sub-user kept AdministratorAccess. Only when NEITHER
        # candidate resolves may we report already-absent.
        candidates = account_id_candidates(raw_id, self._cam_user_name_for)
        user_name = ""
        uin = 0
        uid = 0
        probe_error: TencentCloudSDKException | None = None
        for candidate in candidates:
            try:
                get_req = cam_models.GetUserRequest()  # type: ignore[misc]
                get_req.Name = candidate
                user_resp = self._call(client.GetUser, get_req)
                uin = int(getattr(user_resp, "Uin", 0) or 0)
                uid = int(getattr(user_resp, "Uid", 0) or 0)
                user_name = candidate
                break
            except TencentCloudSDKException as exc:
                code = str(getattr(exc, "code", "") or "")
                if "NotExist" in code or "UserNotExist" in code:
                    continue
                probe_error = exc
                break
        if not user_name:
            if probe_error is not None:
                return ProvisioningResult(
                    ok=False,
                    error_message=redact_provider_error(probe_error),
                )
            return ProvisioningResult(
                ok=True,
                external_ref=raw_id,
                raw_response={
                    "note": "user already absent",
                    "tried": candidates,
                },
            )

        # 1. Detach every policy. CAM's ``ListAttachedUserAllPolicies``
        # REQUIRES ``Page`` (≥1) and ``Rp`` (≤200) — the prior comment
        # claimed the opposite based on a mistaken transfer of the
        # ``ListUsers`` rejection shape. Production error
        # ``MissingParameter Page`` (2026-05-16) is authoritative.
        # Page through with Rp=200 until TotalNum is exhausted so we
        # detach EVERY policy, not just the first page — a partial
        # detach leaves CAM refusing DeleteUser at step 5 with
        # misleading "residual attachments" symptoms. Hard 10-page
        # ceiling (2000 policies) bounds the loop against a malformed
        # API response.
        try:
            items = self._list_attached_policies_authoritative(
                client,
                target_uin=uin,
                attach_type=0,
                context=f"during account revoke for user={user_name!r}",
            )
            for p in items:
                pid = getattr(p, "PolicyId", None)
                if pid is None:
                    continue
                # ``DetachUserPolicy`` only undoes DIRECT attachments. For
                # GROUP-INHERITED policies (which this list mixes in) the
                # call returns RequestId but silently does nothing —
                # step 2 below (``RemoveUserFromGroup``) is what
                # actually revokes those. Calling DetachUserPolicy here
                # is harmless on group-inherited attachments (no-op) and
                # correct on direct ones. SDK 3.1.79's
                # ``DetachUserPolicyRequest`` only carries
                # ``PolicyId`` + ``DetachUin`` — never set
                # ``DetachUserName`` (becomes an ad-hoc attribute and
                # serializes mangled as ``EtachUserName`` → API rejects
                # with ``UnknownParameter``, 2026-06-01 prod incident).
                detach = cam_models.DetachUserPolicyRequest()  # type: ignore[misc]
                detach.PolicyId = int(pid)
                detach.DetachUin = uin
                _swallow_nse(
                    f"detach_policy:{getattr(p, 'PolicyName', '') or pid}",
                    client.DetachUserPolicy, detach,
                )
        except Exception as exc:  # noqa: BLE001
            if (
                isinstance(exc, TencentCloudSDKException)
                and _is_throttle_exhausted(exc)
            ):
                # Surface the real cause — letting it fall through to
                # the DeleteUser step would mask "throttled" as a
                # misleading "residual attachments" failure.
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        "Tencent CAM throttle exhausted detaching policies "
                        f"for {user_name}: {redact_provider_error(exc)}"
                    ),
                    raw_response={"steps": steps},
                )
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Tencent CAM could not collect an authoritative policy "
                    f"snapshot for {user_name}: {redact_provider_error(exc)}"
                ),
                raw_response={"steps": steps},
            )

        # 2. Remove from every group. Same Page/Rp prohibition applies
        # to ``ListGroupsForUser`` — drop pagination params and rely on
        # the default response covering up to 200 groups (well above any
        # tenant we sync). DeleteUser refuses while group memberships
        # exist; pre-fix loop silently failed for the same reason as
        # the policy detach loop above.
        try:
            lg = cam_models.ListGroupsForUserRequest()  # type: ignore[misc]
            # ``Uid`` is the CAM-internal id, NOT the sub-account UIN —
            # passing Uin where Uid is wanted returns
            # ``ResourceNotFound.UserNotExist`` even when the user
            # plainly exists (lying!). Old code threaded ``uin`` here
            # because nobody noticed the field-name mismatch; the
            # offboarding kill-switch hit this in the 2026-06-01 audit.
            lg.Uid = uid
            lg_resp = self._call(client.ListGroupsForUser, lg)
            groups = getattr(lg_resp, "GroupInfo", None) or []
            for g in groups:
                gid = getattr(g, "GroupId", None)
                if gid is None:
                    continue
                rm = cam_models.RemoveUserFromGroupRequest()  # type: ignore[misc]
                # Same Uid-vs-Uin distinction. ``Info`` is a list of
                # ``{Uid, GroupId}`` records; ``Uid`` MUST be the CAM
                # internal id from GetUser, NOT the sub-account UIN.
                rm.Info = [{"Uid": uid, "GroupId": int(gid)}]
                _swallow_nse(
                    f"remove_from_group:{getattr(g, 'GroupName', '') or gid}",
                    client.RemoveUserFromGroup, rm,
                )
            total = int(getattr(lg_resp, "TotalNum", 0) or 0)
            if total > len(groups):
                # Round-22 audit: previously this just warned and let
                # the loop continue, then DeleteUser at step 5 failed
                # with "residual group memberships" — the leaver still
                # belongs to N - returned_count groups. Surface the
                # truncation as ok=False so the operator sees the
                # honest reason (truncated CAM ListGroupsForUser
                # response) instead of the misleading DeleteUser
                # error. Recommend manual cleanup of overflow groups.
                logger.warning(
                    "Tencent CAM user %s belongs to %d groups but "
                    "ListGroupsForUser returned only %d; refusing "
                    "DeleteUser to avoid leaving the leaver with "
                    "residual group memberships",
                    user_name, total, len(groups),
                )
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        f"Tencent CAM user {user_name} belongs to "
                        f"{total} groups but the single-call "
                        f"ListGroupsForUser response only returned "
                        f"{len(groups)} — manual cleanup of the "
                        f"overflow groups is required before this "
                        f"revoke can complete safely."
                    ),
                    raw_response={"steps": steps, "groups_total": total,
                                  "groups_seen": len(groups)},
                )
        except TencentCloudSDKException as exc:
            if _is_throttle_exhausted(exc):
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        "Tencent CAM throttle exhausted leaving groups for "
                        f"{user_name}: {redact_provider_error(exc)}"
                    ),
                    raw_response={"steps": steps},
                )
            logger.warning(
                "Tencent leave groups for %s failed: %s",
                user_name, redact_provider_error(exc),
            )

        # 3. Delete access keys
        try:
            lk = cam_models.ListAccessKeysRequest()  # type: ignore[misc]
            lk.TargetUin = uin
            lk_resp = self._call(client.ListAccessKeys, lk)
            for k in getattr(lk_resp, "AccessKeys", None) or []:
                ak_id = getattr(k, "AccessKeyId", None)
                if not ak_id:
                    continue
                dk = cam_models.DeleteAccessKeyRequest()  # type: ignore[misc]
                dk.TargetUin = uin
                dk.AccessKeyId = str(ak_id)
                _swallow_nse(
                    f"delete_access_key:{ak_id}",
                    client.DeleteAccessKey, dk,
                )
        except TencentCloudSDKException as exc:
            if _is_throttle_exhausted(exc):
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        "Tencent CAM throttle exhausted deleting access keys "
                        f"for {user_name}: {redact_provider_error(exc)}"
                    ),
                    raw_response={"steps": steps},
                )
            logger.warning(
                "Tencent delete access keys for %s failed: %s",
                user_name, redact_provider_error(exc),
            )

        # 4. Delete console login profile
        try:
            # CAM has no standalone DeleteLoginProfile; DeleteUser with
            # Force=1 handles the console password atomically. Still,
            # explicitly wiping it first is defense-in-depth: if the
            # Force delete later fails, at least the user can't log in.
            up = cam_models.UpdateUserRequest()  # type: ignore[misc]
            up.Name = user_name
            up.ConsoleLogin = 0
            up.Password = ""
            up.NeedResetPassword = 0
            _swallow_nse(
                "disable_console_login",
                client.UpdateUser, up,
            )
        except TencentCloudSDKException as exc:
            if _is_throttle_exhausted(exc):
                return ProvisioningResult(
                    ok=False,
                    error_message=(
                        "Tencent CAM throttle exhausted disabling console "
                        f"login for {user_name}: {redact_provider_error(exc)}"
                    ),
                    raw_response={"steps": steps},
                )
            logger.warning(
                "Tencent UpdateUser(disable console) for %s: %s",
                user_name, redact_provider_error(exc),
            )

        # 5. DeleteUser with Force=1 (removes MFA + residual associations)
        try:
            du = cam_models.DeleteUserRequest()  # type: ignore[misc]
            du.Name = user_name
            du.Force = 1
            self._call(client.DeleteUser, du)
            steps.append("delete_user")
        except TencentCloudSDKException as exc:
            code = str(getattr(exc, "code", "") or "")
            if "NotExist" in code or "UserNotExist" in code:
                return ProvisioningResult(
                    ok=True,
                    external_ref=user_name,
                    raw_response={"note": "user already absent", "steps": steps},
                )
            return ProvisioningResult(
                ok=False,
                error_message=(
                    "Tencent DeleteUser failed: "
                    f"{redact_provider_error(exc)}"
                ),
                raw_response={"steps": steps},
            )

        return ProvisioningResult(
            ok=True,
            external_ref=user_name,
            raw_response={"steps": steps},
        )

    # ── 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.

        Tencent CAM is single-tenant — there is no provider-side
        "close this account" API that an organization admin can call
        on a member account (Tencent's organization model does not
        expose CloseMember the way AWS does). For the
        ``standalone_iam`` scope_type that wraps the tenant
        connection, "real closure" therefore means **evacuate every
        Helix-tracked CAM user on the scope** by running the same
        multi-step revoke chain offboarding uses. This is the cloud-
        side action that actually removes Helix's footprint.

        Anything else raises ``NotImplementedError``.
        """
        if scope.scope_type != "standalone_iam":
            raise NotImplementedError(
                f"Tencent driver cannot close scope_type={scope.scope_type!r}",
            )
        if not _TC_AVAILABLE:
            return CloseScopeResult(
                ok=False, error_code="tc_sdk_missing",
                error_message="tencentcloud-sdk-python not installed",
            )
        return await self._evacuate_cam_users(iam_account_ids or [])

    async def _evacuate_cam_users(
        self, iam_account_ids: list[str],
    ) -> CloseScopeResult:
        """Run the per-user CAM revoke chain for every Helix-tracked user."""
        if not iam_account_ids:
            return CloseScopeResult(
                ok=True, provider_state="users_evacuated",
                revoked_users=0,
                detail={"note": "No Helix-tracked CAM users on this scope."},
            )
        evacuated: list[dict[str, Any]] = []
        failures: list[dict[str, Any]] = []
        for user_name in iam_account_ids:
            ctx = CloudAccountRevokeContext(
                provider=self.provider, account_id=user_name,
            )
            try:
                result = await self.revoke_account(ctx)
            except RuntimeError as exc:  # config / SDK errors surface here
                failures.append({
                    "user": user_name,
                    "error": redact_provider_error(exc),
                })
                continue
            if result.ok:
                evacuated.append({
                    "user": user_name,
                    "raw": result.raw_response or {},
                })
            else:
                failures.append({
                    "user": user_name,
                    "error": result.error_message or "no error message returned by driver",
                })
        if failures:
            return CloseScopeResult(
                ok=False, error_code="evacuation_partial",
                error_message=(
                    f"Tencent evacuation partial: {len(failures)} of "
                    f"{len(iam_account_ids)} CAM user(s) failed to delete. "
                    "Inspect the audit detail for per-user errors."
                ),
                revoked_users=len(evacuated),
                detail={"evacuated": evacuated, "failures": failures},
            )
        return CloseScopeResult(
            ok=True, provider_state="users_evacuated",
            revoked_users=len(evacuated),
            detail={"evacuated": evacuated},
        )

    # ── Daily billing (DescribeBillDetail aggregation) ───────

    async def list_billing_daily(
        self, start_date: date, end_date: date,
        *, scope: CloudScopeRef | None = None,
        include_owner_breakdown: bool = False,
        bq_billing: dict[str, Any] | None = None,
    ) -> list[CloudDailyCost]:
        """Daily cost breakdown via ``DescribeBillDetail`` aggregation.

        Tencent bills by **calendar cycle** (``BillMonth``). We query each
        cycle the window touches via the ``Month`` parameter (the figure
        that matches the console 按计费周期 view), sum each row's real cost
        (``ComponentSet[].RealCost``, CNY, discount-applied — see
        ``_bill_detail_real_cost_cny``), and group by ``(usage-day,
        BusinessCodeName)``. The day comes from ``FeeBeginTime``
        (``_bill_detail_usage_day``), NOT ``PayTime`` (扣费时间): monthly
        settlements pay on the 1st of the following month, so PayTime
        grouping mis-attributed a whole cycle's settlement to the next
        month (the bug this method's 2026-06 rewrite fixed).

        Currency stays CNY (same decision as Aliyun — see the comment
        there). The ``CloudDailyCost.total_cost_usd`` field name is a
        legacy misnomer; downstream reads the ``currency`` column as
        source of truth.

        Pagination: ``Limit`` maxes at 100 per page. The loop is capped
        at 1000 pages (~100,000 rows) per cycle. If the cap is hit we
        RAISE rather than return a truncated ledger — a partial cycle
        total would silently understate that month's CNY spend in every
        downstream USD aggregate. The lifecycle service catches the raise
        as a per-scope error and the next cron tick re-ingests the cycle.
        """
        del bq_billing  # AWS-driver-specific kwarg; accept and ignore
        # The Tencent Cloud SDK is a hard runtime dependency declared in
        # pyproject. ``_TC_AVAILABLE`` being False means the deployment is
        # broken — raise so the scheduled job logs the configuration
        # error rather than silently writing zero rows.
        if not _TC_AVAILABLE:
            raise RuntimeError(
                "tencentcloud-sdk-python not importable — Tencent billing "
                "daily sync cannot run on this worker.",
            )
        return await asyncio.to_thread(
            self._list_billing_daily_sync, start_date, end_date, scope,
        )

    def _list_billing_daily_sync(
        self,
        start_date: date,
        end_date: date,
        scope: CloudScopeRef | None,
    ) -> list[CloudDailyCost]:
        # Re-raise instead of returning ``[]``. A missing/expired secret
        # or region misconfiguration would otherwise persist as a
        # successful run with zero rows — every downstream USD aggregate
        # silently understates Tencent spend until an admin notices.
        # The lifecycle service catches the per-scope error and records
        # it against the right scope row so other clouds keep syncing.
        client = self._billing_client()

        # ``by_day[date] = (total, {product_name: cost})``
        by_day: dict[date, tuple[float, dict[str, float]]] = {}
        limit = 100
        # Hard cap so a pathological account can't stall the cron.
        # 1000 pages × 100 rows = 100,000 line items per cycle month.
        max_pages = 1000
        # Query one BILLING CYCLE per month via the ``Month`` parameter
        # (NOT BeginTime/EndTime, which filter by 扣费时间 and split a
        # monthly settlement across the month boundary). ``Month`` returns
        # exactly that cycle's line items — the console 按计费周期 figure —
        # and each row is attributed to its usage day via
        # ``_bill_detail_usage_day`` (FeeBeginTime), so a settlement paid on
        # the 1st of the next month still lands in its own cycle. The
        # window only selects WHICH cycle months to (re-)ingest; the daily
        # cron's trailing window re-pulls the current (and, on month
        # boundaries, the previous) whole cycle — idempotent via the
        # upsert, and it picks up late-settled monthly charges.
        for cycle in _bill_cycle_months(start_date, end_date):
            cy, cm = int(cycle[:4]), int(cycle[5:7])
            offset = 0
            hit_cap = True
            for _ in range(max_pages):
                req = billing_models.DescribeBillDetailRequest()  # type: ignore[misc]
                req.Offset = offset
                req.Limit = limit
                req.Month = cycle
                req.NeedRecordNum = 1
                try:
                    resp = self._call(
                        client.DescribeBillDetail, req, service="billing",
                    )
                except TencentCloudSDKException as exc:
                    # Mid-pagination raise (parity with UCloud + the
                    # cost-facts loop below): a transient throttle / 5xx
                    # mid-cycle would otherwise flush a partial cycle as if
                    # complete — silent under-reporting. The lifecycle
                    # service catches the per-scope error and the next cron
                    # tick re-ingests the whole cycle.
                    raise RuntimeError(
                        f"Tencent DescribeBillDetail Month={cycle} "
                        f"offset={offset} failed: {redact_provider_error(exc)}",
                    ) from exc
                detail = getattr(resp, "DetailSet", None) or []
                for row in detail:
                    # Attribute by billing-cycle usage day, NOT PayTime —
                    # monthly settlements pay on the 1st of the next month.
                    day = _bill_detail_usage_day(row, cy, cm)
                    # Cost lives in ``row.ComponentSet[].RealCost`` — NOT in
                    # a top-level ``RealTotalCost``/``TotalCost`` (which
                    # ``DescribeBillDetail`` rows simply don't have). See
                    # ``_bill_detail_real_cost_cny``.
                    cost = _bill_detail_real_cost_cny(row)
                    # Service-name fallback must match the value
                    # ``_list_cost_facts_sync`` emits (``"Other"``) so the
                    # same logical line item lands in the same bucket on
                    # both the ``cmdb_cloud_billing_daily`` aggregate and
                    # ``cmdb_cost_facts`` line items.
                    product = str(
                        getattr(row, "BusinessCodeName", "")
                        or getattr(row, "ProductCodeName", "")
                        or "Other",
                    )
                    total, services = by_day.get(day, (0.0, {}))
                    total += cost
                    services[product] = services.get(product, 0.0) + cost
                    by_day[day] = (total, services)
                if len(detail) < limit:
                    hit_cap = False
                    break
                offset += limit
            if hit_cap:
                # Symmetric with ``_list_cost_facts_sync`` — a truncated
                # cycle silently undercounts and corrupts the 7-day
                # spike-detection baseline. Raise so the lifecycle service
                # records a per-scope failure and the next tick retries.
                raise RuntimeError(
                    f"Tencent DescribeBillDetail hit {max_pages}-page cap "
                    f"for Month={cycle}; refusing to return a truncated "
                    "billing ledger. Bump ``max_pages`` in "
                    "tencent_cloud._list_billing_daily_sync if this is a "
                    "legitimate >100,000-line-item cycle, then re-trigger "
                    "the billing sync.",
                )

        out: list[CloudDailyCost] = []
        for day, (total, services) in sorted(by_day.items()):
            out.append(CloudDailyCost(
                date=day,
                scope=scope,
                total_cost_usd=total,  # legacy field name; CNY, see currency
                currency="CNY",
                cost_by_service=services,
            ))
        return out

    # ── Line-item cost facts (DescribeBillDetail per-row) ────────────
    #
    # Tencent ``DescribeBillDetail`` returns one row per billable
    # line item (instance × billing-action × pay-time). Unlike
    # ``list_billing_daily`` (which sums to per-day totals), this
    # method preserves the line-item shape — each row becomes one
    # CostFact with full dimensions: ResourceId, RegionName, Tags,
    # ProductCodeName, BusinessCodeName, ActionType.

    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. Tencent CAM has no
        # public per-CAM-user cost dimension — accept and ignore.
        # ``bq_billing`` likewise AWS-driver-specific (CUR-in-BigQuery).
        del principal_tag_keys, bq_billing
        if not _TC_AVAILABLE:
            raise RuntimeError(
                "tencentcloud-sdk-python not importable — Tencent cost-facts "
                "ingestion cannot run on this worker.",
            )
        # FX rate guard — mirrors AliyunCloudDriver (same fix, same
        # rationale). Guard is conditioned on self.requires_cny_fx so
        # a future subclass that sets requires_cny_fx=False (intl
        # billing in non-CNY) can inherit this method safely.
        #
        # requires_cny_fx=True (this class): service MUST resolve the
        # rate; None / ≤0 is a service-layer bug → raise hard.
        # requires_cny_fx=False (hypothetical intl subclass): normalise
        # to 0.0 so CNY rows are dropped by the inner sync method
        # rather than crashing the entire run.
        if cny_per_usd is None or cny_per_usd <= 0:
            if self.requires_cny_fx:
                raise RuntimeError(
                    f"Tencent integration {self.integration_id}: "
                    "``cny_per_usd`` not provided. The cost-ingestion "
                    "service must resolve the rate via "
                    "app.core.fx_rate.resolve_cny_per_usd before calling "
                    "list_cost_facts.",
                )
            cny_per_usd = 0.0
        return await asyncio.to_thread(
            self._list_cost_facts_sync,
            start_date, end_date, scope, float(cny_per_usd),
        )

    def _list_cost_facts_sync(
        self,
        start_date: date,
        end_date: date,
        scope: CloudScopeRef | None,
        fx_rate: float,
    ) -> list[CostFact]:
        # Re-raise instead of returning ``[]``. A missing/expired secret
        # or region misconfiguration would otherwise persist as a
        # successful run with zero rows — every downstream USD aggregate
        # silently understates Tencent spend until an admin notices.
        # The lifecycle service catches the per-scope error and records
        # it against the right scope row so other clouds keep syncing.
        # Round-27 pre-launch (2026-05-25): inner sync MUST guard
        # ``fx_rate > 0`` itself even though the outer ``list_cost_facts``
        # already raises for ``requires_cny_fx=True`` callers. The outer
        # ``cny_per_usd = 0.0`` normalisation path for hypothetical intl
        # subclasses (``requires_cny_fx=False``) would otherwise reach
        # this body and ``native / fx_rate`` would ``ZeroDivisionError``
        # on the first non-zero row — a P0 trap for the next maintainer.
        if not fx_rate or fx_rate <= 0:
            raise RuntimeError(
                f"Tencent integration {self.integration_id}: "
                "_list_cost_facts_sync invoked with fx_rate "
                f"{fx_rate!r} (≤0); caller must resolve a positive "
                "CNY→USD rate before invoking. CNY cost rows cannot "
                "be expressed in USD without an FX rate.",
            )
        client = self._billing_client()

        # Aggregate to the fact-key grain BEFORE emitting. ``ingest_cost_facts``
        # UPSERTs by ``compute_fact_key`` = (integration, date, scope,
        # resource_id, usage_type) with REPLACE semantics (so a re-ingest is
        # idempotent). But Tencent ``DescribeBillDetail`` returns several
        # sub-line-item rows that SHARE that grain — e.g. a single COS bucket
        # on one day yields ~4 rows (different ItemCode / sub-period), same
        # (day, resource_id, ActionType). Emitting them individually made the
        # last UPSERT win and silently dropped the rest (prod 2026-05: COS
        # real ¥3,344 → stored ¥30.51, a 99% undercount). Summing here at the
        # exact key grain makes every emitted ``fact_key`` unique and carries
        # the full amount, so cost_facts totals reconcile to billing_daily /
        # the console. Keyed ``(date, resource_id, usage_type)`` — integration
        # and scope are constant within one call.
        _agg: dict[tuple[date, str | None, str | None], dict[str, Any]] = {}
        integration_id = self.integration_id
        limit = 100
        # Cap exists to prevent a runaway loop; hitting it means we
        # silently truncated, which is a real ingestion-incompleteness
        # hazard. Raise instead of breaking. 1000 pages × 100 = 100,000
        # line items per billing cycle.
        max_pages = 1000
        # Query one BILLING CYCLE per month via the ``Month`` parameter
        # (NOT BeginTime/EndTime, which filter by 扣费时间) and attribute
        # each line item to its usage day (FeeBeginTime), mirroring
        # billing_daily. Matches the console 按计费周期 view and keeps a
        # next-month-settled monthly charge (CDN/COS) in its own cycle.
        for cycle in _bill_cycle_months(start_date, end_date):
            cy, cm = int(cycle[:4]), int(cycle[5:7])
            offset = 0
            hit_cap = True
            for _ in range(max_pages):
                req = billing_models.DescribeBillDetailRequest()  # type: ignore[misc]
                req.Offset = offset
                req.Limit = limit
                req.Month = cycle
                req.NeedRecordNum = 1
                try:
                    resp = self._call(
                        client.DescribeBillDetail, req, service="billing",
                    )
                except TencentCloudSDKException as exc:
                    # Surface every SDK error to the caller. A transient
                    # throttle-exhaust mid-pagination produces a partial
                    # cycle that would land in the fact table as complete —
                    # silently under-counting. Fail and retry next tick.
                    raise RuntimeError(
                        f"Tencent DescribeBillDetail Month={cycle} "
                        f"offset={offset} failed mid-pagination: "
                        f"{redact_provider_error(exc)}",
                    ) from exc
                detail = getattr(resp, "DetailSet", None) or []
                for row in detail:
                    pay_time = str(getattr(row, "PayTime", "") or "")
                    # Attribute by billing-cycle usage day (FeeBeginTime),
                    # NOT PayTime — monthly settlements pay on the 1st of
                    # the following month. See ``_bill_detail_usage_day``.
                    day = _bill_detail_usage_day(row, cy, cm)
                    # Clamp emitted facts to the requested window. Unlike
                    # billing_daily (pure UPSERT, no eviction), the
                    # cost-facts ingestion EVICTS stale rows only within
                    # ``[start_date, end_date]`` (see
                    # ``cost_service.sync_cost_facts_for_integration``), so
                    # emitting an out-of-window day would leave a
                    # never-evicted stale fact if that day's natural key
                    # later disappears. Whole-cycle settlements (FeeBegin =
                    # cycle day 1, available next month) are reconciled by
                    # ``cmdb.cost_facts_monthclose``, which re-pulls the
                    # whole previous month (its window covers day 1).
                    if day < start_date or day > end_date:
                        continue
                    # Cost lives in ``row.ComponentSet[].RealCost`` — NOT in
                    # a top-level ``RealTotalCost``/``TotalCost``. See
                    # ``_bill_detail_real_cost_cny``.
                    native = _bill_detail_real_cost_cny(row)
                    if native == 0:
                        continue
                    # Tags — Tencent line items carry a ``Tags`` array of
                    # ``{TagKey, TagValue}``. Some accounts have ``ResourceId``
                    # tagged at the project level; keep both.
                    raw_tags = getattr(row, "Tags", None) or []
                    tag_map: dict[str, str] = {}
                    for tag in raw_tags:
                        k = str(
                            getattr(tag, "TagKey", "")
                            or getattr(tag, "Key", "") or "",
                        )
                        v = str(
                            getattr(tag, "TagValue", "")
                            or getattr(tag, "Value", "") or "",
                        )
                        if k:
                            tag_map[k] = v

                    resource_id = str(
                        getattr(row, "ResourceId", "") or "",
                    ) or None
                    usage_type = str(
                        getattr(row, "ActionType", "") or "",
                    ) or None
                    # Accumulate into the fact-key grain — sum amounts,
                    # merge tags, keep the first row's dimensions (they are
                    # consistent within a (resource, usage_type) grain).
                    agg_key = (day, resource_id, usage_type)
                    entry = _agg.get(agg_key)
                    if entry is None:
                        entry = {
                            "date": day,
                            "amount_native": 0.0,
                            "service": str(
                                getattr(row, "BusinessCodeName", "")
                                or getattr(row, "ProductCodeName", "")
                                or "Other",
                            ),
                            "resource_group": str(
                                getattr(row, "ProjectName", "") or "",
                            ) or None,
                            "resource_id": resource_id,
                            "resource_name": str(
                                getattr(row, "ResourceName", "") or "",
                            ) or resource_id,
                            "resource_type": str(
                                getattr(row, "ProductCodeName", "") or "",
                            ) or None,
                            "region": str(
                                getattr(row, "RegionName", "") or "",
                            ) or None,
                            "usage_type": usage_type,
                            "tags": {},
                            "raw": {
                                # Subset that's always present + safe to JSON
                                # serialize. The full row often holds binary
                                # Decimal types that confuse the JSON column.
                                "PayTime": pay_time,
                                "ProductCodeName": str(
                                    getattr(row, "ProductCodeName", "") or "",
                                ),
                                "BusinessCodeName": str(
                                    getattr(row, "BusinessCodeName", "") or "",
                                ),
                                "ActionType": str(
                                    getattr(row, "ActionType", "") or "",
                                ),
                            },
                        }
                        _agg[agg_key] = entry
                    entry["amount_native"] += native
                    if tag_map:
                        entry["tags"].update(tag_map)
                if len(detail) < limit:
                    hit_cap = False
                    break
                offset += limit
            if hit_cap:
                raise RuntimeError(
                    f"Tencent DescribeBillDetail hit {max_pages}-page cap "
                    f"for Month={cycle} — ingestion would be incomplete. "
                    "Raise max_pages.",
                )

        out: list[CostFact] = []
        for entry in _agg.values():
            native_sum = entry["amount_native"]
            # A grain whose sub-rows net to zero (charge + offsetting
            # refund/adjustment) is dropped — same intent as the per-row
            # ``native == 0`` skip above, applied to the summed grain.
            if native_sum == 0:
                continue
            out.append(CostFact(
                date=entry["date"],
                integration_id=integration_id,
                service=entry["service"],
                amount_usd=native_sum / fx_rate,
                amount_native=native_sum,
                currency="CNY",
                scope=scope,
                resource_group=entry["resource_group"],
                resource_id=entry["resource_id"],
                resource_name=entry["resource_name"],
                resource_type=entry["resource_type"],
                region=entry["region"],
                usage_type=entry["usage_type"],
                tags=entry["tags"],
                raw=entry["raw"],
            ))
        return out


# Silence unused-import warnings when optional deps are unavailable.
_ = (
    cam_client, cam_models, tc_credential, ClientProfile, HttpProfile,
    billing_client, billing_models, timedelta, CloudPrincipalNotFoundError,
)