"""AWS cloud driver.

Single integration covers both CMDB sync (EC2 / Route 53 / CloudFront /
ELB / S3) *and* per-user IAM account revocation at offboarding. IT
Admins typically register one AWS integration per AWS account:

  * Main AWS account (Google SSO federated) → capabilities: CMDB sync.
    Per-user accounts here are ``sso_cascade`` (they die with Google).
  * AWS SG (Singapore) → capabilities: CMDB sync + IAM revoke.
    Per-user ``cmdb_cloud_accounts`` rows use ``auto_revoke`` policy;
    at offboarding the driver calls ``revoke_account`` to delete the
    IAM user (password, access keys, MFA, policies, group memberships,
    then ``DeleteUser``).
"""
from __future__ import annotations

import asyncio
import re
from datetime import UTC, date, datetime, timedelta
from typing import Any, NamedTuple

try:
    import boto3
    from botocore.config import Config as _BotoConfig
    from botocore.exceptions import BotoCoreError, ClientError, NoCredentialsError
    _BOTO3_AVAILABLE = True
except ImportError:  # pragma: no cover — declared in pyproject
    boto3 = None  # type: ignore[assignment]
    _BotoConfig = None  # type: ignore[assignment, misc]
    BotoCoreError = Exception  # type: ignore[assignment, misc]
    ClientError = Exception  # type: ignore[assignment, misc]
    NoCredentialsError = Exception  # type: ignore[assignment, misc]
    _BOTO3_AVAILABLE = False


class _BqTablePick(NamedTuple):
    """One table the BQ billing helpers should read.

    Produced by ``AwsCloudDriver._bq_resolve_tables`` to centralise
    layout-aware table resolution. See that method's docstring for
    field semantics. Defined at module scope so tests can import it
    without poking at private class state.
    """

    label: str
    fq_table: str
    account_filter: str | None
    columns: set[str] | None


def _aws_client_config() -> Any:
    """Centralised botocore Config for every boto3.client(...) call.

    Default boto3 timeouts are unbounded enough that a hung Identity
    Center / Cost Explorer call (observed in support tickets) holds the
    worker thread for tens of minutes. Pin explicit connect / read
    timeouts and use adaptive retries so transient throttling backs off
    correctly. Returns ``None`` when boto3 isn't installed (drivers
    short-circuit before constructing clients in that case).
    """
    if _BotoConfig is None:
        return None
    return _BotoConfig(
        connect_timeout=10,
        read_timeout=30,
        retries={"max_attempts": 3, "mode": "adaptive"},
    )

from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.providers.aws import AWSProvider
from app.domains.cmdb.providers.exceptions import ProviderAuthError
from app.integrations.driver_base import (
    COST_BUCKET_ADJUSTMENT,
    COST_BUCKET_COMMITMENT_DISCOUNT,
    COST_BUCKET_COMMITMENT_FEE,
    COST_BUCKET_CONTRACT_DISCOUNT,
    COST_BUCKET_CREDIT,
    COST_BUCKET_OTHER,
    COST_BUCKET_TAX,
    COST_BUCKET_USAGE,
    BillingByDimensionRow,
    CloseScopeResult,
    CloudAccessContext,
    CloudAccountRevokeContext,
    CloudDailyCost,
    CloudGrantResult,
    CloudPrincipalNotFoundError,
    CloudRoleInfo,
    CloudScopeInfo,
    CloudScopeRef,
    CostFact,
    DiscoveredGrant,
    OrphanAssignment,
    PrincipalAssignment,
    ProvisioningResult,
    ResourceCreatorInfo,
    ResourceCreatorWindowExceeded,
    ScopeClosureState,
)
from app.integrations.drivers._cloud_defaults import load_default_roles
from app.integrations.drivers._cloud_shared import _CloudDriverBase
from app.integrations.registry import register_driver
from app.integrations.schemas import IntegrationFieldInfo

logger = get_safe_logger(__name__)


# Fallback defaults — used if ``platform_settings`` entry is missing or
# empty (e.g. on a fresh install where migration 054 hasn't run yet).
# Keep in sync with migration 054.
_AWS_IAM_FALLBACK: list[dict[str, Any]] = [
    {"role_key": "arn:aws:iam::aws:policy/AdministratorAccess", "display_name": "AdministratorAccess"},
    {"role_key": "arn:aws:iam::aws:policy/PowerUserAccess", "display_name": "PowerUserAccess"},
    {"role_key": "arn:aws:iam::aws:policy/ReadOnlyAccess", "display_name": "ReadOnlyAccess"},
    {"role_key": "arn:aws:iam::aws:policy/Billing", "display_name": "Billing"},
    {"role_key": "arn:aws:iam::aws:policy/IAMReadOnlyAccess", "display_name": "IAMReadOnlyAccess"},
]


_MAX_CE_PAGES = 200


def _ce_paged(ce: Any, **kwargs: Any):
    """Manual pager for Cost Explorer ``GetCostAndUsage``.

    ``GetCostAndUsage`` is not in boto3's paginator registry
    (https://github.com/boto/boto3/issues/2553) — calling
    ``ce.get_paginator("get_cost_and_usage")`` raises
    ``OperationNotPageableError`` at runtime. Drive the
    ``NextPageToken`` cursor ourselves instead. Yields one raw
    response per page; callers flatten ``ResultsByTime``.

    A hard cap and repeated-token guard fail loudly on a buggy or
    adversarially-crafted NextPageToken chain. Returning the accumulated
    prefix would make authoritative cost syncs persist partial billing data.
    """
    next_token: str | None = None
    seen_tokens: set[str] = set()
    page = 0
    while True:
        if page >= _MAX_CE_PAGES:
            raise RuntimeError(
                "AWS Cost Explorer pagination exceeded "
                f"MAX_CE_PAGES={_MAX_CE_PAGES}; refusing partial results"
            )
        call_kwargs = dict(kwargs)
        if next_token:
            call_kwargs["NextPageToken"] = next_token
        resp = ce.get_cost_and_usage(**call_kwargs)
        yield resp
        page += 1
        next_token = resp.get("NextPageToken")
        if not next_token:
            return
        if next_token in seen_tokens:
            raise RuntimeError(
                "AWS Cost Explorer returned a repeated NextPageToken; "
                "refusing partial results"
            )
        seen_tokens.add(next_token)


# Round-1 review follow-up (Review-A P2-1): warn-once state keyed by
# ``(integration_id, currency)`` so a CNY-rate-missing warning fired
# for integration A doesn't silence the same condition on integration
# B. Pre-fix, the dedup state lived as function-object attributes on
# ``_convert_to_usd`` (module singleton) — cross-integration noise was
# suppressed silently. Module-level set is acceptable for warn-once
# semantics (no cross-process state required); per-process dedup is
# what operators expect from a log signal.
_FX_WARNED_KEYS: set[tuple[str, str]] = set()


def _convert_to_usd(
    *,
    amount: float,
    currency: str,
    cny_per_usd: float | None,
    integration_id: str = "",
) -> float:
    """Map ``amount_native`` to ``amount_usd`` per Helix billing rules.

    Pre-launch hardening 2026-05-16 (LS1 P1): pre-fix the AWS driver
    set ``amount_usd=0`` for every non-USD row, silently undercounting
    AWS China (CNY-billed) spend in every USD aggregate. Now:

    * ``USD``     → pass-through.
    * ``CNY``     → divide by ``cny_per_usd`` (global FX rate resolved
                    in :mod:`app.core.fx_rate`). Falls back to 0.0 +
                    one-shot-per-(integration, currency) log if no
                    rate is configured.
    * other       → 0.0 + one-shot-per-(integration, currency) log
                    (no platform FX rate for BRL / JPY / etc.;
                    operator must add one).

    ``integration_id`` is used only for the warning dedup key so the
    same warning can fire once per integration. Defaults to empty
    string for callers that don't have the id (tests, ad-hoc).
    """
    cur = (currency or "").upper()
    if cur == "USD":
        return float(amount)
    if cur == "CNY":
        if cny_per_usd and cny_per_usd > 0:
            return float(amount) / float(cny_per_usd)
        # Rate missing → drop to zero with a log so the operator sees
        # the gap rather than mystery silent $0.
        warn_key = (integration_id, "CNY")
        if warn_key not in _FX_WARNED_KEYS:
            logger.warning(
                "AWS cost-facts (integration=%s): CNY row encountered "
                "but cny_per_usd rate is missing — amount_usd defaulted "
                "to 0.0. Check platform_settings.ai_spending.cny_to_usd_rate.",
                integration_id or "<unknown>",
            )
            _FX_WARNED_KEYS.add(warn_key)
        return 0.0
    if cur:
        warn_key = (integration_id, cur)
        if warn_key not in _FX_WARNED_KEYS:
            logger.warning(
                "AWS cost-facts (integration=%s): unsupported currency "
                "%s — amount_usd defaulted to 0.0 (Helix has no FX rate "
                "for it). Native amount preserved.",
                integration_id or "<unknown>", cur,
            )
            _FX_WARNED_KEYS.add(warn_key)
    return 0.0


def _aws_resource_type_from_id(resource_id: str | None) -> str | None:
    """Best-effort classification of a Cost Explorer ``RESOURCE_ID``.

    Cost Explorer returns the per-resource id in whatever format the
    underlying service emits — full ARN for many services, bare id
    (``i-0abc...``) for some. This walks the obvious patterns to
    surface a coarse "EC2 / S3 / Lambda / RDS / …" label so the
    dashboard can pivot on resource type without re-issuing a per-
    service describe call. Returns None when the id doesn't match a
    known shape — the dashboard renders those as "Other AWS".
    """
    if not resource_id:
        return None
    # ARN: arn:aws:<service>:<region>:<account>:<rest>
    if resource_id.startswith("arn:aws:"):
        parts = resource_id.split(":", 5)
        if len(parts) >= 6:
            service = parts[2]
            rest = parts[5]
            head = rest.split("/", 1)[0] if "/" in rest else rest
            return f"{service}/{head}" if head else service
        return parts[2] if len(parts) >= 3 else None
    # Bare ids
    if resource_id.startswith("i-"):
        return "ec2/instance"
    if resource_id.startswith("vol-"):
        return "ec2/volume"
    if resource_id.startswith("snap-"):
        return "ec2/snapshot"
    if resource_id.startswith("eni-"):
        return "ec2/network-interface"
    if resource_id.startswith("eip-") or resource_id.startswith("eipalloc-"):
        return "ec2/eip"
    return None


@register_driver
class AwsCloudDriver(_CloudDriverBase):
    provider = "aws"
    display_name = "AWS"
    description = (
        "Amazon Web Services(EC2 / Route 53 / CloudFront / ELB / S3)。"
        "同一连接器还支持在离职时删除该 AWS 账号下的 IAM 用户"
        "(配合「员工云账号」页面的 auto_revoke 策略)。"
    )
    provider_cls = AWSProvider
    capabilities = [
        "servers", "domains", "dns_records", "cdn",
        "object_storage", "load_balancers", "iam", "user_revoke",
    ]
    # AWS Managed Policy ARN for AdministratorAccess — the de facto
    # "owner" role on both Identity Center permission sets (attached
    # as a customer-managed reference) and directly-attached IAM user
    # policies. Include the `AWSAdministratorAccess` permission set
    # name prefix used in many Identity Center setups; the sync looks
    # at both ARN-shape role_keys and permission-set ARNs to match.
    owner_role_keys = frozenset({
        "arn:aws:iam::aws:policy/AdministratorAccess",
        "AWSAdministratorAccess",
    })

    # Role governance (see app/integrations/cloud_role_governance.py).
    # AdministratorAccess is owner-equivalent (above). IAMFullAccess =
    # IAM-write escalation → restricted. PowerUserAccess = everything but
    # IAM → elevated. SSO permission sets named *Admin*/*PowerUser* are
    # caught by the classifier keyword backstop.
    restricted_role_keys = frozenset({
        "arn:aws:iam::aws:policy/IAMFullAccess",
    })
    elevated_role_keys = frozenset({
        "arn:aws:iam::aws:policy/PowerUserAccess",
    })
    config_schema = [
        IntegrationFieldInfo(
            key="access_key_id", label="Access Key ID",
            is_secret=True, required=True,
        ),
        IntegrationFieldInfo(
            key="secret_access_key", label="Secret Access Key",
            is_secret=True, required=True,
        ),
        IntegrationFieldInfo(
            key="default_region", label="Default Region",
            is_secret=False, required=False,
            placeholder="us-east-1",
            default="us-east-1",
        ),
        # Account mode — "sso" = AWS Organizations + IAM Identity Center
        # (multi-account); "iam" = single-account IAM users (AWS SG,
        # Bedrock-style). Determines which lifecycle codepath fires on
        # grant/revoke.
        IntegrationFieldInfo(
            key="account_mode", label="账号模式",
            is_secret=False, required=False,
            placeholder="sso",
            default="sso",
        ),
        # Identity Center specifics (only meaningful when account_mode='sso').
        # Populated by the admin once per AWS Org: the Identity Center
        # instance ARN and the identity store id. Discovered via
        # ``sso-admin list-instances`` — written into the config rather
        # than discovered per-call to save on cold-path latency.
        IntegrationFieldInfo(
            key="sso_instance_arn", label="Identity Center Instance ARN",
            is_secret=False, required=False,
            placeholder="arn:aws:sso:::instance/ssoins-xxx",
            visible_when={"account_mode": "sso"},
        ),
        IntegrationFieldInfo(
            key="sso_identity_store_id", label="Identity Store ID",
            is_secret=False, required=False,
            placeholder="d-xxxxxxxxxx",
            visible_when={"account_mode": "sso"},
        ),
        IntegrationFieldInfo(
            key="sso_region", label="Identity Center Region",
            is_secret=False, required=False,
            placeholder="us-east-1",
            default="us-east-1",
            visible_when={"account_mode": "sso"},
        ),
        # ── Billing source ─────────────────────────────────────────
        # When set to "bigquery", all three billing surfaces
        # (list_billing_daily / list_cost_facts / list_billing_by_dimension)
        # route to the BigQuery-hosted CUR instead of AWS Cost Explorer.
        # AWS SG and any single-account IAM-mode integration leave this
        # at the default "cost_explorer" — those orgs have CE access and
        # no CUR export.
        IntegrationFieldInfo(
            key="billing_source", label="账单数据源",
            is_secret=False, required=False,
            placeholder="cost_explorer",
            default="cost_explorer",
        ),
        # When billing_source="bigquery": fully-qualified BQ dataset
        # holding the CUR export (e.g. ``gcp-digital-bi.aws_billing_no_payer``
        # for CUR 2.0 per-month-table layout, or
        # ``gcp-digital-bi.aws_nonprod_billing`` for CUR 1.0 per-account
        # external Parquet tables). The driver picks the table(s) via
        # ``billing_bq_table_pattern`` + ``billing_bq_table_layout``.
        IntegrationFieldInfo(
            key="billing_bq_dataset", label="BigQuery CUR Dataset",
            is_secret=False, required=False,
            placeholder="gcp-digital-bi.aws_billing_no_payer",
            visible_when={"billing_source": "bigquery"},
        ),
        # UUID of the Helix GCP cloud integration whose SA holds
        # ``bigquery.dataViewer`` on the CUR dataset. The cost-service
        # layer resolves the SA JSON before each call and passes it
        # into the driver via ``bq_billing`` so AWS driver code stays
        # session-free (matches the existing ``cny_per_usd`` /
        # ``principal_tag_keys`` resolution pattern).
        IntegrationFieldInfo(
            key="billing_bq_credentials_integration_id",
            label="BigQuery 凭证(GCP integration UUID)",
            is_secret=False, required=False,
            placeholder="<uuid of GCP cloud integration>",
            visible_when={"billing_source": "bigquery"},
        ),
        # Table-name template. Two layouts are supported (see
        # ``billing_bq_table_layout`` below):
        #   per_account → uses ``{account_id}`` placeholder, e.g.
        #                 ``{account_id}_aws_billing_data`` (CUR 1.0,
        #                 PingCAP legacy ``aws_nonprod_billing``).
        #   per_month   → uses ``{yyyymmdd}`` placeholder substituted with
        #                 each month's first-day (``20260501`` for May
        #                 2026), e.g. ``aws_billing_{yyyymmdd}`` (CUR 2.0
        #                 / Data Exports, PingCAP ``aws_billing_no_payer``).
        IntegrationFieldInfo(
            key="billing_bq_table_pattern",
            label="BigQuery 表名模板",
            is_secret=False, required=False,
            placeholder="aws_billing_{yyyymmdd}",
            visible_when={"billing_source": "bigquery"},
        ),
        # Table layout switch — selects partition predicate, table
        # resolution, and scope-filter strategy.
        #
        #   per_account (default — CUR 1.0):
        #     - Tables: one EXTERNAL Parquet table per linked AWS account,
        #       named via the ``{account_id}`` placeholder.
        #     - Partition: Hive ``year`` + ``month`` INT columns.
        #     - Scope filter: route to the per-account table; tenant-wide
        #       discovers tables via ``INFORMATION_SCHEMA.TABLES``.
        #
        #   per_month (CUR 2.0 / Data Exports):
        #     - Tables: one native TABLE per month, all linked accounts
        #       merged, named via the ``{yyyymmdd}`` placeholder (month's
        #       first-day in YYYYMMDD form).
        #     - Partition: native DAY partition on
        #       ``line_item_usage_end_date``.
        #     - Scope filter: ``WHERE line_item_usage_account_id = ?``
        #       (the table holds all accounts); tenant-wide enumerates
        #       month-tables in the requested date range.
        IntegrationFieldInfo(
            key="billing_bq_table_layout", label="BigQuery 表布局",
            is_secret=False, required=False,
            placeholder="per_month",
            default="per_account",
            visible_when={"billing_source": "bigquery"},
        ),
        # Tag schema switch — selects how ``resource_tags_user_*`` is
        # projected in SELECT and harvested into ``CostFact.tags``.
        #
        #   flat (default — CUR 1.0): one column per tag named
        #     ``resource_tags_user_<sanitised_key>``. Schema heterogeneity
        #     across per-account tables means a tag key may be absent in
        #     some tables — the driver introspects ``Table.schema`` and
        #     drops the projection for tables that don't carry it.
        #
        #   nested (CUR 2.0): tags live in a single nested column
        #     ``resource_tags.key_value`` of type
        #     ``ARRAY<STRUCT<key STRING, value STRING>>``. Element keys
        #     carry the ``user_`` prefix (e.g. ``user_owner``). Schema is
        #     uniform across all per-month tables — the per-table
        #     schema-drop logic does not apply.
        IntegrationFieldInfo(
            key="billing_bq_tag_schema", label="BigQuery Tag Schema",
            is_secret=False, required=False,
            placeholder="nested",
            default="flat",
            visible_when={"billing_source": "bigquery"},
        ),
        # Fact-grain switch — selects how cost rows are aggregated
        # before landing in ``cmdb_cost_facts``.
        #
        #   per_product (default):
        #     Aggregate at (day × account × product_code × usage_type ×
        #     currency). Resource_id is synthesized as
        #     ``aws-account:<acc>:<product_code>``. This matches the
        #     CUR 1.0 absent-resource_id behaviour Helix has shipped
        #     since the BQ-billing path was introduced; one fact per
        #     (day × product) per scope. Volume bounded to ~hundreds
        #     of facts per scope per day.
        #
        #   per_resource:
        #     Aggregate at (day × account × product_code × resource_id
        #     × usage_type × currency). Real ``line_item_resource_id``
        #     used; produces thousands of facts per scope per day
        #     (proportional to active-resource count). Required when
        #     the cost-allocation rule ladder needs per-resource
        #     attribution (e.g. matching a chargeback rule on an EC2
        #     instance ARN). Caller must size the cron's per-scope
        #     timeout (currently 5 minutes) for the resulting TiDB
        #     upsert volume.
        #
        # CUR 2.0 (the ``aws_billing_no_payer`` dataset) makes
        # ``line_item_resource_id`` present on every row, while CUR
        # 1.0 (legacy ``aws_nonprod_billing``) had it absent for many
        # per-account tables — the CUR 1.0 per_account+flat path
        # therefore behaved as ``per_product`` by accident. The
        # explicit ``per_product`` default keeps behaviour stable as
        # the dataset cuts over.
        IntegrationFieldInfo(
            key="billing_bq_fact_grain", label="BigQuery Fact Grain",
            is_secret=False, required=False,
            placeholder="per_resource",
            default="per_product",
            visible_when={"billing_source": "bigquery"},
        ),
        # ── AWS Organizations cross-account fan-out (servers + metrics) ──
        # When on, inventory + utilization enumerate EVERY active member
        # account of the AWS Organization by STS-assuming ``org_member_role``
        # into each (the management/payer account itself usually runs 0
        # EC2 — all compute lives in member accounts). Off by default: a
        # single-account integration keeps its current single-account
        # behaviour. Requires the management credential to hold
        # ``organizations:ListAccounts`` + ``sts:AssumeRole`` on the
        # member role. Only meaningful for the org management account.
        IntegrationFieldInfo(
            key="org_fanout", label="组织跨账号扇出(EC2/利用率)",
            is_secret=False, required=False,
            placeholder="false",
            default="false",
            visible_when={"account_mode": "sso"},
        ),
        # IAM role name to STS-assume in each member account. Default is
        # the role AWS Organizations auto-creates in every org-created
        # member account (``OrganizationAccountAccessRole``). Set to a
        # dedicated read-only role (e.g. ``HelixCloudReadOnly`` deployed
        # org-wide via a CloudFormation StackSet) for least-privilege.
        IntegrationFieldInfo(
            key="org_member_role", label="成员账号角色名",
            is_secret=False, required=False,
            placeholder="OrganizationAccountAccessRole",
            default="OrganizationAccountAccessRole",
            visible_when={"org_fanout": "true"},
        ),
        # Region denylist — opt-in regions enabled on some account but
        # holding no resources and unreachable from the deploy host (e.g.
        # me-south-1). Scanning them only yields a per-tick connect-timeout
        # partial_failure that pages IT for pure noise. Comma/space-
        # separated region names; excluded from EC2 region discovery.
        IntegrationFieldInfo(
            key="region_denylist", label="跳过的区域(逗号分隔)",
            is_secret=False, required=False,
            placeholder="me-south-1",
        ),
        IntegrationFieldInfo(
            key="resource_inventory_enabled", label="启用 Resource Explorer 全资源目录",
            is_secret=False, required=False, placeholder="false", default="false",
        ),
        IntegrationFieldInfo(
            key="resource_explorer_region", label="Resource Explorer 聚合索引区域",
            is_secret=False, required=False, placeholder="ap-southeast-1",
            visible_when={"resource_inventory_enabled": "true"},
        ),
    ]

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        # Per-instance cache of dataset strings whose existence has
        # already been confirmed via ``_bq_probe_dataset_exists``.
        # ``_load_cloud_driver`` (cost_service / lifecycle_service)
        # builds a FRESH driver instance per scope per cron tick, so
        # this cache only de-dupes within a single driver instance —
        # i.e. across the 3 BQ methods ``list_billing_daily`` /
        # ``list_cost_facts`` / ``list_billing_by_dimension`` called
        # for the SAME scope (3 probes → 1). Across 60 scopes the
        # cache is essentially a no-op (60 driver instances → 60
        # fresh probes). Modest saving, but the alternative (lazy
        # ``getattr``-fallback init) reads as a defence against
        # something more dramatic than the cache actually does.
        # Initialised here (vs. lazy ``getattr``) so the contract is
        # explicit at the class boundary.
        self._bq_probed_datasets: set[str] = set()
        # Cross-scope batched cost-facts cache (per_month / CUR 2.0 +
        # per_product grain only). The cost_facts cron reuses ONE driver
        # instance across every linked-account scope (cost_service
        # ``_sync_cost_facts_for_integration``: "Load and reuse one
        # driver instance across all scope calls"). For per_month, all
        # linked accounts live in one DAY-partitioned month-table and
        # the ``line_item_usage_account_id`` WHERE filter does NOT reduce
        # bytes scanned — the table is partitioned by date, not clustered
        # by account (verified 2026-06: a one-day account-filtered scan
        # and the unfiltered all-accounts scan bill identical bytes).
        # Issuing one query per account therefore re-scans the same
        # partitions N times (prod cost: ~38× — the recurring daily-cron
        # bleed + multi-hundred-$ backfill spikes, all billed to the GCP
        # project that runs the SA's BQ jobs). When cost_service hands us
        # the full target set via ``_cost_facts_all_account_scopes``, the
        # FIRST scope's BQ call scans the window ONCE for every account
        # and buckets the facts here keyed by account id; the rest are
        # served from this cache. Keyed by (start, end) — constant across
        # one cron run — and GC'd with the instance when the run's driver
        # is closed. ``None`` until cost_service opts in (other call
        # sites / providers leave the per-account streaming path intact).
        self._cost_facts_all_account_scopes: list[CloudScopeRef] | None = (
            None
        )
        self._bq_cost_facts_batch_cache: dict[
            tuple[date, date], dict[str, list[CostFact]]
        ] = {}
        # Same cross-scope batching for the daily billing sync
        # (``sync_billing_daily_for_integration``): per_month CUR runs up
        # to 3 per-account queries (totals / service / owner) that each
        # re-scan the same date-partitions, since the account filter
        # doesn't prune bytes. When the caller opts in via
        # ``_billing_daily_all_account_scopes`` the first scope scans all
        # accounts once and the rest are served from this cache (keyed by
        # (start, end, include_owner_breakdown) — the owner pass is
        # conditional, so its presence changes the cached shape).
        self._billing_daily_all_account_scopes: (
            list[CloudScopeRef] | None
        ) = None
        self._bq_billing_daily_batch_cache: dict[
            tuple[date, date, bool], dict[str, list[CloudDailyCost]]
        ] = {}

    # ── Per-user IAM account revocation ────────────────────────

    async def revoke_account(
        self, ctx: CloudAccountRevokeContext,
    ) -> ProvisioningResult:
        if not _BOTO3_AVAILABLE:
            return ProvisioningResult(
                ok=False, error_message="boto3 not installed",
            )
        user_name = ctx.account_id
        if not user_name:
            return ProvisioningResult(
                ok=False,
                error_message="Empty account_id (expected IAM UserName)",
            )
        try:
            deleted_steps = await asyncio.to_thread(self._revoke_sync, user_name)
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "ClientError")
            if code == "NoSuchEntity":
                return ProvisioningResult(
                    ok=True,
                    external_ref=user_name,
                    raw_response={"note": "user not found — already deleted"},
                )
            return ProvisioningResult(
                ok=False,
                error_message=f"AWS ClientError {code}: {exc}",
            )
        except (BotoCoreError, NoCredentialsError) as exc:
            return ProvisioningResult(ok=False, error_message=str(exc))
        return ProvisioningResult(
            ok=True,
            external_ref=user_name,
            raw_response={"steps": deleted_steps},
        )

    # ── IAM user re-tag (API-Key transfer flow) ───────────────

    async def tag_iam_user(
        self,
        user_name: str,
        tags: dict[str, str],
    ) -> ProvisioningResult:
        """Set / overwrite tags on an existing IAM user.

        Used by [`ApiKeyService.apply_transfer`](server/app/domains/apikey/service.py)
        on ownership transfer: the Bedrock auto-grant flow stamps the
        applicant's work_email into both the ``Owner`` and
        ``helix-owner-email`` IAM principal tags so AWS Cost Explorer
        groups Bedrock spend by ``iamPrincipal/Owner``. When the
        assignment changes hands the tag must follow — otherwise CE
        keeps attributing new spend to the prior owner's email.

        ``iam.tag_user`` is an UPSERT (adds new keys, overwrites
        existing ones); we don't need to inspect-and-diff first.
        Idempotent on the same payload. ``NoSuchEntity`` from the
        underlying IAM call surfaces as ``ok=False`` with the original
        AWS error code so the caller's audit row captures the
        condition — typical when the IAM user was already deleted out
        of band before the transfer flow ran (rare; offboarding +
        transfer normally don't coexist on the same row).
        """
        if not _BOTO3_AVAILABLE:
            return ProvisioningResult(
                ok=False, error_message="boto3 not installed",
            )
        if not user_name:
            return ProvisioningResult(
                ok=False, error_message="Empty user_name",
            )
        if not tags:
            # No-op: an empty payload would be a no-op at AWS anyway,
            # but exiting early surfaces the caller's mistake instead
            # of writing a useless audit row.
            return ProvisioningResult(
                ok=True,
                external_ref=user_name,
                raw_response={"note": "empty tag set — noop"},
            )
        try:
            await asyncio.to_thread(
                self._tag_user_sync, user_name, tags,
            )
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "ClientError")
            return ProvisioningResult(
                ok=False,
                error_message=f"AWS ClientError {code}: {exc}",
            )
        except (BotoCoreError, NoCredentialsError) as exc:
            return ProvisioningResult(ok=False, error_message=str(exc))
        return ProvisioningResult(
            ok=True,
            external_ref=user_name,
            raw_response={"tagged": list(tags.keys())},
        )

    def _tag_user_sync(self, user_name: str, tags: dict[str, str]) -> None:
        iam = self._iam_client()
        iam.tag_user(
            UserName=user_name,
            Tags=[{"Key": k, "Value": v} for k, v in tags.items()],
        )

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

        Dispatches by ``scope_type``:

        * ``aws_account`` → ``organizations.close_account`` on the
          configured Organizations management credentials. AWS moves the
          account to ``PENDING_CLOSURE`` immediately and permanently
          deletes it after a 90-day grace period. Idempotent: an already-
          closed (``SUSPENDED``) account returns ``already_closed=True``.

        * ``standalone_iam`` → enumerate the IAM users this Helix scope
          tracks (``iam_account_ids``) and call the same multi-step
          revoke chain that offboarding uses. Real ``DeleteUser`` per
          user — login profile, access keys, MFA, policies, group
          memberships removed first. Returns the count actually deleted.
          The AWS account itself is not closed (that requires root-user
          billing actions, not IAM credentials).

        Anything else raises ``NotImplementedError`` so the lifecycle
        service surfaces a 501 rather than a silent no-op.
        """
        if not _BOTO3_AVAILABLE:
            return CloseScopeResult(
                ok=False,
                error_message="boto3 not installed",
                error_code="boto3_missing",
            )
        if scope.scope_type == "aws_account":
            return await asyncio.to_thread(
                self._close_org_account_sync, scope.external_id,
            )
        if scope.scope_type == "standalone_iam":
            return await asyncio.to_thread(
                self._evacuate_iam_users_sync,
                scope.external_id,
                iam_account_ids or [],
            )
        raise NotImplementedError(
            f"AWS driver cannot close scope_type={scope.scope_type!r}",
        )


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

    async def get_scope_closure_state(
        self, scope: CloudScopeRef,
    ) -> ScopeClosureState:
        """Re-read Organizations' view of an account Helix already closed.

        ``CloseAccount`` parks the member in ``PENDING_CLOSURE`` and AWS
        permanently deletes it ~90 days later; the account row keeps
        answering ``DescribeAccount`` the whole time. ``ACTIVE`` here is
        the alarming reading: it means the close never took or the
        account was restored, while Helix has already disabled its
        grants.

        ``standalone_iam`` closure is a set of ``DeleteUser`` calls with
        no provider-side after-state, so it raises NotImplementedError
        and the reconciler skips it for good.
        """
        if scope.scope_type != "aws_account":
            raise NotImplementedError(
                "AWS driver has no post-close state for "
                f"scope_type={scope.scope_type!r}",
            )
        if not _BOTO3_AVAILABLE:
            return ScopeClosureState(
                disposition="unknown", error_message="boto3 not installed",
            )
        return await asyncio.to_thread(
            self._closure_state_sync, (scope.external_id or "").strip(),
        )

    def _closure_state_sync(self, account_id: str) -> ScopeClosureState:
        org = self._boto_client("organizations")
        try:
            describe = org.describe_account(AccountId=account_id)
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code in {
                "AccountNotFoundException",
                "AWSOrganizationsNotInUseException",
            }:
                return ScopeClosureState(
                    disposition="closed", provider_state="not_in_org",
                    detail={"reason": "Account no longer in this Organization"},
                )
            return ScopeClosureState(
                disposition="unknown",
                error_message=f"AWS describe_account: {exc}"[:300],
            )
        except (BotoCoreError, NoCredentialsError) as exc:
            return ScopeClosureState(
                disposition="unknown", error_message=str(exc)[:300],
            )
        status = str(describe.get("Account", {}).get("Status") or "").upper()
        detail = {"aws_status": status}
        if status == "SUSPENDED":
            return ScopeClosureState(
                disposition="closed", provider_state="suspended", detail=detail,
            )
        if status == "PENDING_CLOSURE":
            return ScopeClosureState(
                disposition="pending", provider_state="pending_closure",
                detail=detail,
            )
        if status == "ACTIVE":
            return ScopeClosureState(
                disposition="alive", provider_state="active", detail=detail,
                error_message=(
                    "AWS Organizations reports this member account as "
                    "ACTIVE, but Helix has it marked closed and has "
                    "already disabled its grants."
                ),
            )
        return ScopeClosureState(
            disposition="unknown", detail=detail,
            error_message=f"AWS returned an unrecognised account status {status!r}",
        )

    def _close_org_account_sync(self, account_id: str) -> CloseScopeResult:
        """Call ``organizations.close_account`` and classify the response.

        AWS Organizations CloseAccount goes from the management account
        only — the integration's credentials must be either the org
        management account or have ``organizations:CloseAccount`` granted
        via service-control / delegation. Errors caused by missing
        permissions or unmet preconditions surface verbatim so the admin
        knows what to fix on the AWS side.

        Pre-flight: read ``describe_account`` first so we can tell
        ``ALREADY_CLOSED`` (state ``SUSPENDED``) apart from ``does not
        exist``. Both are idempotent successes but the audit detail
        differs.
        """
        org = self._boto_client("organizations")
        try:
            describe = org.describe_account(AccountId=account_id)
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code in {"AccountNotFoundException", "AWSOrganizationsNotInUseException"}:
                return CloseScopeResult(
                    ok=True,
                    already_closed=True,
                    provider_state="not_in_org",
                    detail={"reason": "Account not in this Organization"},
                )
            if code == "AccessDeniedException":
                return CloseScopeResult(
                    ok=False,
                    error_code="access_denied",
                    error_message=(
                        "AWS denied organizations:DescribeAccount. "
                        "The integration credentials must be the Organizations "
                        "management account (or a delegated administrator) "
                        "with organizations:CloseAccount permission."
                    ),
                )
            return CloseScopeResult(
                ok=False, error_code=code or "describe_failed",
                error_message=f"AWS Organizations describe_account: {exc}",
            )
        except (BotoCoreError, NoCredentialsError) as exc:
            return CloseScopeResult(
                ok=False, error_code="boto_error", error_message=str(exc),
            )

        existing_status = str(
            describe.get("Account", {}).get("Status") or "",
        ).upper()
        if existing_status in {"SUSPENDED", "PENDING_CLOSURE"}:
            return CloseScopeResult(
                ok=True,
                already_closed=True,
                provider_state=existing_status.lower(),
                detail={"prior_status": existing_status},
            )

        try:
            org.close_account(AccountId=account_id)
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            msg = (exc.response or {}).get("Error", {}).get("Message", str(exc))
            if code == "AccountAlreadyClosedException":
                return CloseScopeResult(
                    ok=True,
                    already_closed=True,
                    provider_state="suspended",
                    detail={"prior_status": "SUSPENDED"},
                )
            if code == "AccountNotFoundException":
                return CloseScopeResult(
                    ok=True,
                    already_closed=True,
                    provider_state="not_in_org",
                    detail={"reason": "Account not in this Organization"},
                )
            if code == "AccessDeniedException":
                return CloseScopeResult(
                    ok=False, error_code="access_denied",
                    error_message=(
                        "AWS denied organizations:CloseAccount. "
                        "Verify the integration credentials are the "
                        "Organizations management account (or a delegated "
                        "administrator with organizations:CloseAccount)."
                    ),
                )
            if code == "ConstraintViolationException":
                # Most common production failure — account doesn't meet
                # AWS's preconditions (no payment method, root user
                # missing MFA, holdable resources, etc.). Surface the
                # exact provider message so the admin can act.
                return CloseScopeResult(
                    ok=False, error_code="constraint_violation",
                    error_message=(
                        "AWS rejected CloseAccount: " + msg
                        + " — fix the precondition on the AWS console "
                        + "(Account → Close account checklist) and retry."
                    ),
                )
            return CloseScopeResult(
                ok=False, error_code=code or "close_failed",
                error_message=f"AWS Organizations close_account: {exc}",
            )
        except (BotoCoreError, NoCredentialsError) as exc:
            return CloseScopeResult(
                ok=False, error_code="boto_error", error_message=str(exc),
            )

        return CloseScopeResult(
            ok=True,
            provider_state="pending_closure",
            detail={
                "grace_period_days": 90,
                "note": (
                    "Account is in PENDING_CLOSURE; AWS will permanently "
                    "delete it 90 days after this call."
                ),
            },
        )

    def _evacuate_iam_users_sync(
        self, scope_external_id: str, iam_account_ids: list[str],
    ) -> CloseScopeResult:
        """Run the per-user revoke chain for every Helix-tracked IAM user.

        ``iam_account_ids`` carries the IAM ``UserName`` values the
        service layer collected from ``cmdb_iam_users``. We deliberately
        do NOT call ``iam.list_users`` and delete every IAM user in the
        account — that would trash service users Helix never touched.
        A scope with no Helix-tracked principals is a clean ``ok=True``
        with ``revoked_users=0``.
        """
        if not iam_account_ids:
            return CloseScopeResult(
                ok=True,
                provider_state="users_evacuated",
                revoked_users=0,
                detail={
                    "scope_external_id": scope_external_id,
                    "evacuated": [],
                    "failures": [],
                    "note": "No Helix-tracked IAM users on this scope.",
                },
            )
        evacuated: list[dict[str, Any]] = []
        failures: list[dict[str, Any]] = []
        for user_name in iam_account_ids:
            try:
                steps = self._revoke_sync(user_name)
                evacuated.append({"user": user_name, "steps": steps})
            except ClientError as exc:
                code = (exc.response or {}).get("Error", {}).get("Code", "")
                if code == "NoSuchEntity":
                    evacuated.append({
                        "user": user_name,
                        "steps": ["already_absent"],
                    })
                else:
                    failures.append({"user": user_name, "code": code, "error": str(exc)})
            except (BotoCoreError, NoCredentialsError) as exc:
                failures.append({"user": user_name, "error": str(exc)})

        if failures:
            return CloseScopeResult(
                ok=False, error_code="evacuation_partial",
                error_message=(
                    f"AWS evacuation partial: {len(failures)} of "
                    f"{len(iam_account_ids)} IAM user(s) failed to delete. "
                    "Inspect the audit detail for per-user errors."
                ),
                revoked_users=len(evacuated),
                detail={
                    "scope_external_id": scope_external_id,
                    "evacuated": evacuated,
                    "failures": failures,
                },
            )
        return CloseScopeResult(
            ok=True,
            provider_state="users_evacuated",
            revoked_users=len(evacuated),
            detail={
                "scope_external_id": scope_external_id,
                "evacuated": evacuated,
                "failures": [],
            },
        )

    # ── IAM delete chain internals ─────────────────────────────

    def _iam_client(self) -> Any:
        # Route through _boto_client so the credential guard fires —
        # without this, missing AK/SK silently falls through to the
        # EC2 instance profile and IAM mutations run against the wrong
        # AWS account. IAM is a global service so region is irrelevant,
        # but _boto_client's guard is what we need here.
        region = (self.config.get("default_region") or "us-east-1").strip()
        return self._boto_client("iam", region=region)

    def _revoke_sync(self, user_name: str) -> list[str]:
        iam = self._iam_client()
        steps: list[str] = []

        def _swallow_nse(label: str, fn, /, *args: Any, **kwargs: Any) -> None:
            try:
                fn(*args, **kwargs)
                steps.append(label)
            except ClientError as exc:
                code = (exc.response or {}).get("Error", {}).get("Code", "")
                if code == "NoSuchEntity":
                    return
                raise

        # Round-22 audit: every list_* below MUST paginate. AWS IAM caps
        # most list responses at 100 items (some at 1000) and returns
        # ``IsTruncated=True`` + ``Marker`` for the rest. Previously
        # callers ignored ``Marker`` and DeleteUser at step 7 then
        # blew up on residual attachments with the misleading "user
        # has policies attached" message — the leaver in fact still
        # had policies #101..N undetached, plus access keys #101..N
        # still live. Use boto3 paginators uniformly.

        def _paginate_iam(method: str, key: str, **kwargs: Any) -> list[Any]:
            """Return the full list across all pages for an IAM list_* call."""
            paginator = iam.get_paginator(method)
            out: list[Any] = []
            for page in paginator.paginate(**kwargs):
                out.extend(page.get(key) or [])
            return out

        # 1. Console password
        _swallow_nse("delete_login_profile", iam.delete_login_profile, UserName=user_name)

        # 2. Access keys (paginated)
        keys = _paginate_iam(
            "list_access_keys", "AccessKeyMetadata", UserName=user_name,
        )
        for k in keys:
            _swallow_nse(
                f"delete_access_key:{k['AccessKeyId']}",
                iam.delete_access_key,
                UserName=user_name, AccessKeyId=k["AccessKeyId"],
            )

        # 3. Attached + inline policies (paginated)
        attached = _paginate_iam(
            "list_attached_user_policies", "AttachedPolicies",
            UserName=user_name,
        )
        for p in attached:
            _swallow_nse(
                f"detach_user_policy:{p['PolicyName']}",
                iam.detach_user_policy,
                UserName=user_name, PolicyArn=p["PolicyArn"],
            )
        inline = _paginate_iam(
            "list_user_policies", "PolicyNames", UserName=user_name,
        )
        for name in inline:
            _swallow_nse(
                f"delete_user_policy:{name}",
                iam.delete_user_policy,
                UserName=user_name, PolicyName=name,
            )

        # 4. Group memberships (paginated)
        groups = _paginate_iam(
            "list_groups_for_user", "Groups", UserName=user_name,
        )
        for g in groups:
            _swallow_nse(
                f"remove_user_from_group:{g['GroupName']}",
                iam.remove_user_from_group,
                UserName=user_name, GroupName=g["GroupName"],
            )

        # 5. MFA devices (paginated)
        mfa = _paginate_iam(
            "list_mfa_devices", "MFADevices", UserName=user_name,
        )
        for m in mfa:
            _swallow_nse(
                f"deactivate_mfa_device:{m['SerialNumber']}",
                iam.deactivate_mfa_device,
                UserName=user_name, SerialNumber=m["SerialNumber"],
            )
            if m.get("SerialNumber", "").startswith("arn:aws:iam::"):
                _swallow_nse(
                    f"delete_virtual_mfa_device:{m['SerialNumber']}",
                    iam.delete_virtual_mfa_device,
                    SerialNumber=m["SerialNumber"],
                )

        # 6. Signing certificates + service-specific credentials + SSH keys
        # Use paginators where available; list_service_specific_credentials
        # has no paginator so we call it directly (AWS caps it at 100).
        certs = _paginate_iam(
            "list_signing_certificates", "Certificates", UserName=user_name,
        )
        for c in certs:
            _swallow_nse(
                f"delete_signing_certificate:{c['CertificateId']}",
                iam.delete_signing_certificate,
                UserName=user_name, CertificateId=c["CertificateId"],
            )
        ssh_keys = _paginate_iam(
            "list_ssh_public_keys", "SSHPublicKeys", UserName=user_name,
        )
        for s in ssh_keys:
            _swallow_nse(
                f"delete_ssh_public_key:{s['SSHPublicKeyId']}",
                iam.delete_ssh_public_key,
                UserName=user_name, SSHPublicKeyId=s["SSHPublicKeyId"],
            )
        svc_creds = iam.list_service_specific_credentials(UserName=user_name).get(
            "ServiceSpecificCredentials",
        ) or []
        for sc in svc_creds:
            _swallow_nse(
                f"delete_service_specific_credential:{sc['ServiceSpecificCredentialId']}",
                iam.delete_service_specific_credential,
                UserName=user_name,
                ServiceSpecificCredentialId=sc["ServiceSpecificCredentialId"],
            )

        # 7. Final delete
        _swallow_nse("delete_user", iam.delete_user, UserName=user_name)
        return steps

    # ── Account mode helper ────────────────────────────────────

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

    def _boto_client(self, service: str, *, region: str | None = None) -> Any:
        # Fail-closed if the configured AK/SK pair is missing. Without
        # this, boto3 silently falls through to its credential chain
        # (env vars → ~/.aws → EC2 IMDS instance role). On the prod EC2
        # host that means Helix would operate against the host's
        # instance role instead of the integration's account — silently
        # tagging billing rows / running IAM mutations against the wrong
        # AWS account. Mirrors azure_cloud.py:263 posture; same rule
        # applies to every cloud driver. (Round-17 audit.)
        ak = self.config.get("access_key_id")
        sk = self.config.get("secret_access_key")
        if not ak or not sk:
            raise RuntimeError(
                "AWS integration is missing access_key_id / "
                "secret_access_key — refusing to fall through to "
                "ambient credentials (would operate against the wrong "
                "AWS account). Re-save the integration with valid keys."
            )
        return boto3.client(  # type: ignore[union-attr]
            service,
            aws_access_key_id=ak,
            aws_secret_access_key=sk,
            region_name=(
                region or self.config.get("sso_region")
                or self.config.get("default_region") or "us-east-1"
            ),
            config=_aws_client_config(),
        )

    # ── Scope sync ─────────────────────────────────────────────

    async def list_scopes(self) -> list[CloudScopeInfo]:
        if not _BOTO3_AVAILABLE:
            raise RuntimeError("boto3 not installed")
        if self.account_mode == "iam":
            return await asyncio.to_thread(self._iam_self_scope_sync)
        return await asyncio.to_thread(self._list_org_accounts_sync)

    def _own_account_id(self) -> str:
        """Return the 12-digit AWS account id the configured AK lives in.

        Resolved via ``sts:GetCallerIdentity``. The result is the
        canonical key for *every* downstream join: Cost Explorer's
        ``LINKED_ACCOUNT`` dimension, the ``cmdb_cloud_scopes.external_id``
        unique constraint, billing line items, and offboarding IAM
        target lookup. There is **no fallback** — if STS is
        unreachable we cannot identify the tenant, so writing a row
        anyway would create a ghost scope (production incident
        2026-04-23: integration UUID was used as external_id, the
        next successful sync wrote the real account id alongside,
        and the ghost row collected stale grants until manual
        cleanup). Raise ``ProviderAuthError`` so the sync job fails
        loudly and the admin fixes credentials before any data lands.
        """
        try:
            sts = self._boto_client("sts")
            account = str(sts.get_caller_identity().get("Account") or "")
        except (ClientError, BotoCoreError, NoCredentialsError) as exc:
            raise ProviderAuthError(
                f"AWS sts:GetCallerIdentity failed — cannot resolve "
                f"account id for integration {self.integration_id}: {exc}",
            ) from exc
        if not account.isdigit() or len(account) != 12:
            raise ProviderAuthError(
                f"AWS sts:GetCallerIdentity returned a non-canonical "
                f"account id ({account!r}) for integration "
                f"{self.integration_id}; refusing to write a scope row.",
            )
        return account

    def _iam_self_scope_sync(self) -> list[CloudScopeInfo]:
        """Single-tenant IAM-mode: one scope = this AWS account."""
        alias = str(self.config.get("_alias") or self.integration_id)
        account_id = self._own_account_id()
        return [CloudScopeInfo(
            scope_type="standalone_iam",
            external_id=account_id,
            display_name=alias,
            status="active",
            raw={"mode": "iam", "aws_account_id": account_id},
        )]

    def _list_org_accounts_sync(self) -> list[CloudScopeInfo]:
        org = self._boto_client("organizations")
        out: list[CloudScopeInfo] = []
        try:
            paginator = org.get_paginator("list_accounts")
            for page in paginator.paginate():
                for acct in page.get("Accounts") or []:
                    status = str(acct.get("Status", "ACTIVE")).lower()
                    out.append(CloudScopeInfo(
                        scope_type="aws_account",
                        external_id=str(acct["Id"]),
                        display_name=acct.get("Name"),
                        status="active" if status == "active" else status,
                        raw={
                            "email": acct.get("Email"),
                            "arn": acct.get("Arn"),
                            "status": acct.get("Status"),
                            "joined_method": acct.get("JoinedMethod"),
                        },
                    ))
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code in ("AWSOrganizationsNotInUseException", "AccessDeniedException"):
                # Not part of an Organization — treat as single-account
                # tenant. Resolve the real 12-digit AWS account id via
                # STS so Cost Explorer's LINKED_ACCOUNT filter matches
                # and the scope.external_id is canonical. STS failure
                # raises ProviderAuthError (no UUID fallback —
                # see _own_account_id docstring).
                alias = str(self.config.get("_alias") or self.integration_id)
                account_id = self._own_account_id()
                return [CloudScopeInfo(
                    scope_type="aws_account",
                    external_id=account_id,
                    display_name=alias,
                    status="active",
                    raw={"note": "org-not-in-use", "aws_account_id": account_id},
                )]
            raise
        return out

    # ── Scope creation (Organizations CreateAccount) ──────────────

    async def create_scope(
        self,
        *,
        name: str,
        description: str | None,
        owner_email: str,
        budget_usd_monthly: float,
        metadata: dict[str, Any] | None = None,
        idempotency_key: str | None = None,
    ) -> CloudScopeInfo:
        """AWS Organizations ``CreateAccount`` — production path.

        Requires the integration's credentials to be the management
        (payer) account credentials with ``organizations:CreateAccount``
        + the Organizations service-linked role
        (``AWSServiceRoleForOrganizations``) already provisioned.

        The Organizations API requires a unique email per member
        account; we deliberately use ``owner_email`` so the new account
        belongs (paperwork-wise) to the requester. AWS sends the
        account-recovery flow to that address; the requester can then
        complete IAM root-user setup if needed (we strongly recommend
        leaving root unused and managing access via Identity Center
        permission sets only — ``post_create_grants`` in the lifecycle
        service handles that handshake).

        ``CreateAccount`` is **async**: the API returns a
        ``CreateAccountStatus`` immediately and the actual account
        provision can take 30s-5min. We poll up to 6 minutes (24×15s);
        on timeout we raise so the scope-request lands in ``failed``
        with the request id — an admin can then check
        ``DescribeCreateAccountStatus`` from the AWS console and either
        re-trigger the lifecycle hook or close out the request.

        **Idempotency**: AWS Organizations CreateAccount accepts NO
        ``ClientRequestToken`` (the parameter is rejected client-side
        by botocore). Redrive safety comes from
        ``_find_existing_create_account_request`` — before issuing
        ``create_account`` we sweep ``list_create_account_status`` for
        an IN_PROGRESS / SUCCEEDED entry matching this ``AccountName``
        and, if found, resume polling that RequestId. The 30-day
        retention on CreateAccountStatus naturally bounds the lookup
        window. ``idempotency_key`` is accepted for signature
        compatibility but is not propagated to AWS — there's no slot
        for it on this API.

        IAM-mode tenants (``account_mode=iam``) skip Organizations
        entirely; calling ``create_scope`` against them is meaningless
        (the tenant *is* the single account). Raise so the route
        surfaces ``cannot create scope on a single-account tenant``.
        """
        if self.account_mode == "iam":
            raise NotImplementedError(
                "AWS integration is in IAM (single-account) mode; new "
                "sub-accounts must be requested against an Organizations-"
                "enabled tenant.",
            )
        if not _BOTO3_AVAILABLE:
            raise RuntimeError(
                "boto3 not importable — AWS create_scope cannot run on "
                "this worker.",
            )
        return await asyncio.to_thread(
            self._create_account_sync,
            name, description, owner_email, budget_usd_monthly, metadata,
            idempotency_key,
        )

    @staticmethod
    def _find_existing_create_account_request(
        org: Any, *, account_name: str,
    ) -> str | None:
        """Return the RequestId of any existing IN_PROGRESS / SUCCEEDED
        ``CreateAccountStatus`` whose AccountName matches.

        Powers redrive-safe idempotency for
        :meth:`_create_account_sync`: a worker crash after
        ``create_account`` lands but before Helix-side persistence
        completes leaves a real in-flight AWS request that the next
        attempt should resume polling rather than reissue. AWS only
        retains CreateAccountStatus rows for ~30 days; older entries
        roll off naturally, so the lookup window is self-limiting.

        Returns ``None`` if no match — caller proceeds with a fresh
        ``create_account``. On any ``ClientError`` from
        ``list_create_account_status`` we return ``None`` rather than
        bubbling: the worst-case failure mode (AWS rejects the
        ensuing duplicate ``create_account`` with the AccountName
        already-in-use) is more debuggable than a list-call timeout
        crashing the whole approve path.
        """
        try:
            paginator = org.get_paginator("list_create_account_status")
            for page in paginator.paginate(
                States=["IN_PROGRESS", "SUCCEEDED"],
            ):
                for entry in page.get("CreateAccountStatuses") or []:
                    if str(entry.get("AccountName") or "").strip() == \
                            account_name.strip():
                        rid = str(entry.get("Id") or "").strip()
                        if rid:
                            return rid
        except ClientError as exc:
            logger.warning(
                "list_create_account_status sweep failed for "
                "AccountName=%r: %s — falling through to create_account",
                account_name, exc,
            )
        return None

    def _create_account_sync(
        self,
        name: str,
        description: str | None,
        owner_email: str,
        budget_usd_monthly: float,
        metadata: dict[str, Any] | None,
        idempotency_key: str | None = None,
    ) -> CloudScopeInfo:
        org = self._boto_client("organizations")
        # AWS allows letters / digits / spaces / a few specials; we let
        # the upstream validator do the heavy lifting (front-end +
        # Pydantic), and pass through what the user typed. Anything
        # truly illegal will surface as an InvalidInputException with
        # a clear AWS-side message — better than a Helix-side regex
        # that drifts from AWS rules.
        request: dict[str, Any] = {
            "Email": owner_email,
            "AccountName": name,
            "IamUserAccessToBilling": "ALLOW",
        }
        # Idempotency model (corrected 2026-05-25): AWS Organizations
        # ``CreateAccount`` does NOT accept ``ClientRequestToken`` (only
        # Email/AccountName/RoleName/IamUserAccessToBilling/Tags are
        # valid). The previous implementation passed one and botocore
        # rejected the request client-side — every retry hit
        # ``Parameter validation failed`` before the call ever reached
        # AWS, leaving scope_requests stuck in ``provisioning``.
        #
        # Real idempotency: before calling ``create_account``, sweep
        # ``list_create_account_status`` for IN_PROGRESS / SUCCEEDED
        # entries with the SAME ``AccountName``. AWS-side AccountName
        # uniqueness isn't enforced (you CAN have duplicates), but
        # Helix-side the (integration_id, name) is already the
        # request-row's natural identity, so a redrive after a worker
        # crash will see the in-flight status row and resume polling
        # it instead of minting a second account. The
        # ``idempotency_key`` kwarg is kept for signature compatibility
        # — it could be wired into request metadata in the future, but
        # AWS gives us no first-class slot for it on this API.
        existing_request_id = self._find_existing_create_account_request(
            org, account_name=name,
        )
        if existing_request_id:
            request_id = existing_request_id
            logger.info(
                "Organizations.CreateAccount idempotent resume: "
                "AccountName=%r already has in-flight/succeeded request %s "
                "— skipping create_account, polling existing",
                name, request_id,
            )
        else:
            try:
                resp = org.create_account(**request)
            except ClientError as exc:
                raise RuntimeError(
                    f"Organizations.CreateAccount rejected: {exc}",
                ) from exc
            status = resp.get("CreateAccountStatus") or {}
            request_id = str(status.get("Id") or "")
            if not request_id:
                raise RuntimeError(
                    "Organizations.CreateAccount returned no Id — cannot "
                    "track provisioning",
                )

        # Poll DescribeCreateAccountStatus up to 6 min (Organizations
        # SLA on first-account creation in a fresh org is up to 15 min,
        # but a steady-state add typically lands in <2 min; 6 min is a
        # pragmatic ceiling that keeps the worker from holding the
        # async task forever while still covering the realistic 95th
        # percentile).
        import time as _time
        deadline = _time.monotonic() + 360
        last_state = "IN_PROGRESS"
        last_failure_reason = ""
        account_id: str | None = None
        while _time.monotonic() < deadline:
            try:
                desc = org.describe_create_account_status(
                    CreateAccountRequestId=request_id,
                ).get("CreateAccountStatus") or {}
            except ClientError as exc:
                raise RuntimeError(
                    f"DescribeCreateAccountStatus({request_id}) failed: {exc}",
                ) from exc
            last_state = str(desc.get("State") or last_state).upper()
            if last_state == "SUCCEEDED":
                account_id = str(desc.get("AccountId") or "")
                break
            if last_state == "FAILED":
                last_failure_reason = str(desc.get("FailureReason") or "")
                raise RuntimeError(
                    f"Organizations.CreateAccount failed "
                    f"(reason={last_failure_reason or 'unspecified'}, "
                    f"request_id={request_id})",
                )
            _time.sleep(15)

        if not account_id:
            raise RuntimeError(
                f"Organizations.CreateAccount timed out after 6 min "
                f"(state={last_state}, request_id={request_id}). "
                f"Check DescribeCreateAccountStatus and link the account "
                f"manually if it eventually succeeds.",
            )

        return CloudScopeInfo(
            scope_type="aws_account",
            external_id=account_id,
            display_name=name,
            status="active",
            raw={
                "create_account_request_id": request_id,
                "owner_email": owner_email,
                "description": description,
                "metadata": metadata or {},
                "budget_usd_monthly": budget_usd_monthly,
            },
        )

    # ── Role catalog ───────────────────────────────────────────

    async def list_scope_roles(
        self, scope: CloudScopeRef,
    ) -> list[CloudRoleInfo]:
        if not _BOTO3_AVAILABLE:
            raise RuntimeError("boto3 not installed")
        if self.account_mode == "iam":
            # IAM mode: role_key = managed-policy ARN. Admin-curated
            # default list comes from platform_settings so it's editable
            # without a deploy; customer-managed policies on the account
            # are still listed live below.
            defaults = await load_default_roles(
                "cloud.aws_iam.default_policies", _AWS_IAM_FALLBACK,
            )
            return await asyncio.to_thread(
                self._list_iam_policies_sync, defaults,
            )
        # SSO mode: permission sets live at the Identity Center instance
        # level (not per-account). They're bindable to any member
        # account; we return them once per tenant (scope-wide catalog).
        return await asyncio.to_thread(self._list_permission_sets_sync, scope)

    def _list_iam_policies_sync(
        self,
        defaults: list[dict[str, Any]],
        limit: int = 200,
    ) -> list[CloudRoleInfo]:
        iam = self._boto_client("iam")
        out: list[CloudRoleInfo] = []
        # Admin-curated AWS-managed policies (platform_setting). All
        # promoted — they're already a hand-picked subset.
        for entry in defaults:
            arn = str(entry.get("role_key") or "").strip()
            if not arn:
                continue
            out.append(CloudRoleInfo(
                role_key=arn,
                display_name=str(entry.get("display_name") or arn),
                description=entry.get("description"),
                is_custom=False,
                source="aws_managed",
                is_promoted=True,
            ))
        # Customer-managed (a.k.a. "local") policies — live, all promoted.
        try:
            paginator = iam.get_paginator("list_policies")
            count = 0
            for page in paginator.paginate(Scope="Local"):
                for p in page.get("Policies") or []:
                    out.append(CloudRoleInfo(
                        role_key=p["Arn"],
                        display_name=p.get("PolicyName") or p["Arn"],
                        description=p.get("Description"),
                        is_custom=True, source="customer_managed",
                        raw={"default_version": p.get("DefaultVersionId")},
                        is_promoted=True,
                    ))
                    count += 1
                    if count >= limit:
                        break
                if count >= limit:
                    break
        except ClientError as exc:
            # Caller may have iam:ListPolicies on AWS-managed but not
            # on customer-managed (rare but seen in least-privilege
            # roles). Log so the gap shows up in support tickets;
            # AWS-managed policies already collected above are still
            # returned.
            logger.warning(
                "AWS list_policies(Scope=Local) failed (returning "
                "AWS-managed only): %s", exc,
            )
        return out

    def _list_permission_sets_sync(
        self, scope: CloudScopeRef,
    ) -> list[CloudRoleInfo]:
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        if not instance_arn:
            return []
        sso = self._boto_client("sso-admin", region=self.config.get("sso_region"))
        out: list[CloudRoleInfo] = []
        try:
            paginator = sso.get_paginator("list_permission_sets")
            for page in paginator.paginate(InstanceArn=instance_arn):
                for arn in page.get("PermissionSets") or []:
                    desc = sso.describe_permission_set(
                        InstanceArn=instance_arn,
                        PermissionSetArn=arn,
                    ).get("PermissionSet") or {}
                    out.append(CloudRoleInfo(
                        role_key=arn,
                        display_name=desc.get("Name") or arn,
                        description=desc.get("Description"),
                        is_custom=False,
                        source="permission_set",
                        scope=scope,
                        raw={"session_duration": desc.get("SessionDuration")},
                        # Permission Sets are already an admin-curated set
                        # (the org defines them in Identity Center); always
                        # promoted.
                        is_promoted=True,
                    ))
        except ClientError as exc:
            logger.warning("list_permission_sets failed: %s", exc)
        return out

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

    async def grant_access(
        self, ctx: CloudAccessContext,
    ) -> CloudGrantResult:
        if not _BOTO3_AVAILABLE:
            return CloudGrantResult(ok=False, error_message="boto3 not installed")
        if self.account_mode == "iam":
            return await asyncio.to_thread(self._grant_iam_sync, ctx)
        return await asyncio.to_thread(self._grant_sso_sync, ctx)

    async def revoke_access(
        self, ctx: CloudAccessContext,
    ) -> CloudGrantResult:
        if not _BOTO3_AVAILABLE:
            return CloudGrantResult(ok=False, error_message="boto3 not installed")
        if self.account_mode == "iam":
            return await asyncio.to_thread(self._revoke_iam_sync, ctx)
        return await asyncio.to_thread(self._revoke_sso_sync, ctx)

    # ── IAM-mode grant/revoke (single-tenant, e.g. AWS SG) ───

    def _iam_user_name_for(self, email: str) -> str:
        # Keep the derivation stable so repeated grants on the same
        # email hit the same user — IAM-mode semantics is "one IAM user
        # per person, multiple policies attached".
        local = email.split("@", 1)[0].lower()
        # IAM UserName allowed chars: [A-Za-z0-9+=,.@_-], ≤64 chars.
        sanitized = "".join(
            c for c in local
            if c.isascii() and (c.isalnum() or c in "+=,.@_-")
        )
        if sanitized:
            return sanitized[:64]
        # Round-20 audit (A1F5): if the entire local-part was non-ASCII
        # (e.g. ``张三@x.com``), the previous shape collapsed every such
        # email to the literal ``helix-user``. Two non-ASCII employees
        # then collided on one IAM user — they shared policies, and
        # offboarding one deleted the IAM user out from under the other.
        # Fall back to a deterministic hash of the full email so the
        # mapping is still 1:1 and stable across runs.
        import hashlib
        digest = hashlib.sha1(
            email.encode("utf-8"),
            usedforsecurity=False,
        ).hexdigest()[:12]
        return f"helix-{digest}"

    def _grant_iam_sync(self, ctx: CloudAccessContext) -> CloudGrantResult:
        iam = self._boto_client("iam")
        user_name = self._iam_user_name_for(ctx.principal_email)
        policy_arn = ctx.role_key  # role_key is the policy ARN in IAM mode

        # Ensure user exists. The check-then-create pattern races on
        # concurrent grants for the same email (admin retry, two grant
        # paths fired in parallel) — if both pass ``get_user``→NoSuchEntity
        # and both call ``create_user``, the loser sees
        # ``EntityAlreadyExistsException`` from AWS. Treating that as
        # failure (the previous behaviour) leaves the user existing but
        # the requested policy un-attached — manual cleanup. Treat
        # ``EntityAlreadyExists`` from create as success-equivalent
        # (someone else just won the race) and continue to attach.
        try:
            iam.get_user(UserName=user_name)
            created_user = False
        except ClientError as exc:
            if (exc.response or {}).get("Error", {}).get("Code") == "NoSuchEntity":
                try:
                    iam.create_user(
                        UserName=user_name,
                        Tags=[
                            {"Key": "helix:email", "Value": ctx.principal_email},
                            {"Key": "helix:managed", "Value": "true"},
                        ],
                    )
                    created_user = True
                except ClientError as create_exc:
                    create_code = (
                        (create_exc.response or {})
                        .get("Error", {}).get("Code", "")
                    )
                    if create_code == "EntityAlreadyExistsException" or \
                            create_code == "EntityAlreadyExists":
                        # Lost the race — another concurrent grant just
                        # created the same IAM user. Continue with
                        # ``created_user=False`` and attach the policy.
                        created_user = False
                    else:
                        return CloudGrantResult(
                            ok=False, error_message=str(create_exc),
                        )
            else:
                return CloudGrantResult(ok=False, error_message=str(exc))

        # Attach policy (idempotent — AWS returns no error on re-attach)
        try:
            existing = {
                p["PolicyArn"]
                for p in iam.list_attached_user_policies(
                    UserName=user_name,
                ).get("AttachedPolicies") or []
            }
            if policy_arn in existing:
                return CloudGrantResult(
                    ok=True, external_ref=f"{user_name}::{policy_arn}",
                    already_exists=True,
                    detail={"user_name": user_name, "policy_arn": policy_arn},
                )
            iam.attach_user_policy(UserName=user_name, PolicyArn=policy_arn)
        except ClientError as exc:
            return CloudGrantResult(ok=False, error_message=str(exc))

        detail: dict[str, Any] = {
            "user_name": user_name,
            "policy_arn": policy_arn,
            "created_user": created_user,
        }

        # Only mint AK/SK when asked by the caller (``extra.create_access_key``)
        # — Bedrock auto-grant still owns Bedrock's AK/SK flow.
        if ctx.extra.get("create_access_key"):
            try:
                ak = iam.create_access_key(UserName=user_name).get("AccessKey") or {}
                detail["access_key_id"] = ak.get("AccessKeyId")
                detail["secret_access_key"] = ak.get("SecretAccessKey")
            except ClientError as exc:
                # Grant succeeded; leave the AK error on detail for the
                # lifecycle service to surface without failing the grant.
                detail["access_key_error"] = str(exc)

        return CloudGrantResult(
            ok=True,
            external_ref=f"{user_name}::{policy_arn}",
            detail=detail,
        )

    def _revoke_iam_sync(self, ctx: CloudAccessContext) -> CloudGrantResult:
        iam = self._boto_client("iam")
        if ctx.existing_external_ref and "::" in ctx.existing_external_ref:
            user_name, policy_arn = ctx.existing_external_ref.split("::", 1)
        else:
            user_name = self._iam_user_name_for(ctx.principal_email)
            policy_arn = ctx.role_key
        try:
            iam.detach_user_policy(UserName=user_name, PolicyArn=policy_arn)
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code == "NoSuchEntity":
                return CloudGrantResult(ok=True, already_absent=True)
            return CloudGrantResult(ok=False, error_message=str(exc))
        # Note: we deliberately DO NOT delete the IAM user here — the
        # user may still have other Helix grants. Full user deletion is
        # what ``revoke_account`` does, called only on offboarding.
        return CloudGrantResult(ok=True, external_ref=ctx.existing_external_ref)

    # ── SSO-mode grant/revoke (Identity Center) ─────────────

    def _resolve_identity_store_user_id(self, email: str) -> str:
        identity_store_id = str(self.config.get("sso_identity_store_id") or "").strip()
        if not identity_store_id:
            raise RuntimeError("sso_identity_store_id not configured")
        ids = self._boto_client("identitystore", region=self.config.get("sso_region"))
        # Identity Store ListUsers only supports a `UserName` filter — any
        # other AttributePath (e.g. `Emails.Value`) raises
        # ValidationException ("Only filter with value 'userName' is allowed
        # for this operation"). Try the cheap exact UserName filter first,
        # then fall back to a paginated client-side scan matching on the
        # listed emails (covers SCIM users whose UserName != work email).
        res = ids.list_users(
            IdentityStoreId=identity_store_id,
            Filters=[{"AttributePath": "UserName", "AttributeValue": email}],
        )
        users = res.get("Users") or []
        if users:
            return str(users[0]["UserId"])

        target = email.strip().lower()
        next_token: str | None = None
        while True:
            kwargs: dict[str, Any] = {
                "IdentityStoreId": identity_store_id,
                "MaxResults": 100,
            }
            if next_token:
                kwargs["NextToken"] = next_token
            page = ids.list_users(**kwargs)
            for user in page.get("Users") or []:
                if str(user.get("UserName") or "").strip().lower() == target:
                    return str(user["UserId"])
                for em in user.get("Emails") or []:
                    if str(em.get("Value") or "").strip().lower() == target:
                        return str(user["UserId"])
            next_token = page.get("NextToken")
            if not next_token:
                break

        raise CloudPrincipalNotFoundError(
            f"Identity Center has no user with email/username {email!r}. "
            "SCIM-sync them from Google Workspace first.",
        )

    def _grant_sso_sync(self, ctx: CloudAccessContext) -> CloudGrantResult:
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        if not instance_arn:
            return CloudGrantResult(
                ok=False, error_message="sso_instance_arn not configured",
            )
        try:
            user_id = self._resolve_identity_store_user_id(ctx.principal_email)
        except CloudPrincipalNotFoundError as exc:
            return CloudGrantResult(ok=False, error_message=str(exc))
        except (ClientError, BotoCoreError) as exc:
            return CloudGrantResult(ok=False, error_message=str(exc))
        sso = self._boto_client("sso-admin", region=self.config.get("sso_region"))
        try:
            resp = sso.create_account_assignment(
                InstanceArn=instance_arn,
                TargetId=ctx.scope.external_id,
                TargetType="AWS_ACCOUNT",
                PermissionSetArn=ctx.role_key,
                PrincipalType="USER",
                PrincipalId=user_id,
            )
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code == "ConflictException":
                # Already assigned — idempotent success.
                return CloudGrantResult(
                    ok=True, already_exists=True,
                    external_ref=f"{ctx.scope.external_id}::{ctx.role_key}::{user_id}",
                    detail={"principal_id": user_id},
                )
            return CloudGrantResult(ok=False, error_message=str(exc))
        request_status = (resp.get("AccountAssignmentCreationStatus") or {})
        status = str(request_status.get("Status") or "").upper()
        external_ref = f"{ctx.scope.external_id}::{ctx.role_key}::{user_id}"
        detail = {
            "principal_id": user_id,
            "request_id": request_status.get("RequestId"),
            "status": request_status.get("Status"),
            "failure_reason": request_status.get("FailureReason"),
        }
        # create_account_assignment is asynchronous server-side: it returns
        # 200 with a status of IN_PROGRESS | SUCCEEDED | FAILED. A synchronous
        # FAILED (e.g. permission-set not provisioned to the account, SCIM
        # principal mismatch) previously landed as ok=True — the grant row
        # went active while AWS granted nothing (#1650). Only SUCCEEDED /
        # IN_PROGRESS are non-failures; FAILED (or any unexpected value) is
        # a hard failure the caller must surface + retry.
        if status not in ("SUCCEEDED", "IN_PROGRESS"):
            return CloudGrantResult(
                ok=False,
                external_ref=external_ref,
                error_message=(
                    "AWS SSO create_account_assignment returned status "
                    f"{request_status.get('Status')!r}: "
                    f"{request_status.get('FailureReason') or 'no reason given'}"
                ),
                detail=detail,
            )
        return CloudGrantResult(
            ok=True,
            external_ref=external_ref,
            detail=detail,
        )

    def _revoke_sso_sync(self, ctx: CloudAccessContext) -> CloudGrantResult:
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        if not instance_arn:
            return CloudGrantResult(
                ok=False, error_message="sso_instance_arn not configured",
            )
        # Recover principal id from existing_external_ref if we have it;
        # otherwise re-resolve by email.
        user_id: str | None = None
        if ctx.existing_external_ref and ctx.existing_external_ref.count("::") == 2:
            _, _, user_id = ctx.existing_external_ref.split("::", 2)
        if not user_id:
            try:
                user_id = self._resolve_identity_store_user_id(ctx.principal_email)
            except CloudPrincipalNotFoundError:
                # Principal gone → the assignment is already effectively
                # revoked (AWS removes assignments when the principal is
                # deleted from Identity Store).
                return CloudGrantResult(ok=True, already_absent=True)
            except (ClientError, BotoCoreError) as exc:
                return CloudGrantResult(ok=False, error_message=str(exc))
        sso = self._boto_client("sso-admin", region=self.config.get("sso_region"))
        try:
            resp = sso.delete_account_assignment(
                InstanceArn=instance_arn,
                TargetId=ctx.scope.external_id,
                TargetType="AWS_ACCOUNT",
                PermissionSetArn=ctx.role_key,
                PrincipalType="USER",
                PrincipalId=user_id,
            )
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code == "ResourceNotFoundException":
                return CloudGrantResult(ok=True, already_absent=True)
            return CloudGrantResult(ok=False, error_message=str(exc))
        # delete_account_assignment is async server-side (same as the grant
        # path, #1650): it returns 200 with a status of IN_PROGRESS |
        # SUCCEEDED | FAILED. A synchronous FAILED must NOT be reported as a
        # successful revoke — that would close an offboarding blocker while
        # the leaver keeps the permission set. Only SUCCEEDED / IN_PROGRESS
        # are non-failures.
        del_status = (resp.get("AccountAssignmentDeletionStatus") or {})
        status = str(del_status.get("Status") or "").upper()
        if status and status not in ("SUCCEEDED", "IN_PROGRESS"):
            return CloudGrantResult(
                ok=False,
                external_ref=ctx.existing_external_ref,
                error_message=(
                    "AWS SSO delete_account_assignment returned status "
                    f"{del_status.get('Status')!r}: "
                    f"{del_status.get('FailureReason') or 'no reason given'}"
                ),
                detail={
                    "request_id": del_status.get("RequestId"),
                    "status": del_status.get("Status"),
                    "failure_reason": del_status.get("FailureReason"),
                },
            )
        return CloudGrantResult(ok=True, external_ref=ctx.existing_external_ref)

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

    async def list_scope_grants(
        self, scope: CloudScopeRef,
    ) -> list[DiscoveredGrant]:
        """Enumerate currently-effective grants on an AWS scope.

        SSO mode (``account_mode='sso'``): walk every permission set
        assigned to this account via ``sso-admin list_permission_sets``
        + ``list_account_assignments``. Resolve each
        ``PrincipalId`` (Identity Store user id) to the user's primary
        email via ``identitystore.describe_user``.

        IAM mode (``account_mode='iam'``): iterate ``iam.list_users``
        + each user's attached managed policies. The "email" we return
        is the UserName if it looks like an email, else the IAM user's
        ``email`` tag if set, else the bare username (the service will
        skip unresolvable ones).
        """
        if self.account_mode == "iam":
            return await asyncio.to_thread(
                self._list_scope_grants_iam_sync, scope,
            )
        # ``account_mode`` defaults to ``"sso"``, but a tenant can be
        # registered without populating ``sso_instance_arn`` /
        # ``sso_identity_store_id`` (e.g. AWS SG in prod, which uses
        # plain IAM users). The original SSO codepath silently returns
        # ``[]`` in that case — masking dozens of real IAM grants. Fall
        # back to the IAM enumeration so admins don't have to know
        # they're "supposed to" set ``account_mode='iam'`` to surface
        # their data.
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        identity_store_id = str(
            self.config.get("sso_identity_store_id") or "",
        ).strip()
        if not instance_arn or not identity_store_id:
            logger.info(
                "AWS list_scope_grants: account_mode=sso but Identity "
                "Center not configured — falling back to IAM enumeration "
                "for scope %s",
                scope.external_id,
            )
            return await asyncio.to_thread(
                self._list_scope_grants_iam_sync, scope,
            )
        return await asyncio.to_thread(
            self._list_scope_grants_sso_sync, scope,
        )

    def _list_scope_grants_sso_sync(
        self, scope: CloudScopeRef,
    ) -> list[DiscoveredGrant]:
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        identity_store_id = str(
            self.config.get("sso_identity_store_id") or "",
        ).strip()
        if not instance_arn or not identity_store_id:
            logger.info(
                "AWS list_scope_grants SSO: missing sso_instance_arn or "
                "sso_identity_store_id — skip scope %s",
                scope.external_id,
            )
            return []
        sso = self._boto_client("sso-admin", region=self.config.get("sso_region"))
        ids = self._boto_client("identitystore", region=self.config.get("sso_region"))
        account_id = scope.external_id

        # Which permission sets are actually provisioned for this
        # account? Cheaper than walking every permission set in the
        # instance and asking for its account assignments.
        ps_arns: list[str] = []
        try:
            paginator = sso.get_paginator(
                "list_permission_sets_provisioned_to_account",
            )
            for page in paginator.paginate(
                InstanceArn=instance_arn, AccountId=account_id,
            ):
                ps_arns.extend(page.get("PermissionSets") or [])
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code in ("AuthFailure", "InvalidClientTokenId"):
                raise ProviderAuthError(str(exc)) from exc
            # Pre-launch P0 fix (mass-revoke safety): only a benign
            # "permission set never provisioned" condition is allowed
            # to degrade to an empty list. Throttling / 5xx / regional
            # outage / unexpected codes MUST raise so the grants_sync
            # caller skips reconciliation rather than overwriting
            # state with [] and revoking every Helix-tracked grant.
            if code in ("ResourceNotFoundException",):
                logger.info(
                    "AWS list_scope_grants SSO: account %s has no "
                    "provisioned permission sets — empty grant list",
                    account_id,
                )
                return []
            logger.warning(
                "AWS list_scope_grants SSO: list_permission_sets %s on "
                "account %s — re-raising (mass-revoke safety): %s",
                code, account_id, exc,
            )
            raise

        email_cache: dict[str, str | None] = {}
        out: list[DiscoveredGrant] = []
        for ps_arn in ps_arns:
            # Pre-launch P0 fix (partial-snapshot safety): a per-PS
            # ClientError MUST surface to the caller. Previously the
            # ``except → continue`` produced a silently partial
            # snapshot which the grants-sync drift detector treated as
            # "grant absent on cloud → revoke from Helix" — same
            # mass-revoke risk class as the outer except.
            paginator = sso.get_paginator("list_account_assignments")
            for page in paginator.paginate(
                InstanceArn=instance_arn,
                AccountId=account_id,
                PermissionSetArn=ps_arn,
            ):
                for asgn in page.get("AccountAssignments") or []:
                    ptype = str(asgn.get("PrincipalType") or "").upper()
                    pid = str(asgn.get("PrincipalId") or "")
                    if ptype != "USER" or not pid:
                        continue
                    email = email_cache.get(pid)
                    if email is None and pid not in email_cache:
                        email = self._identity_store_email(
                            ids, identity_store_id, pid,
                        )
                        email_cache[pid] = email
                    if not email:
                        continue
                    out.append(DiscoveredGrant(
                        principal_email=email,
                        principal_type="user",
                        role_key=ps_arn,
                        role_display_name=None,
                        external_ref=f"{account_id}::{ps_arn}::{pid}",
                        scope=scope,
                        granted_at=None,
                        raw={
                            "account_id": account_id,
                            "permission_set_arn": ps_arn,
                            "identity_store_user_id": pid,
                        },
                    ))
        return out

    def _identity_store_email(
        self, ids_client, identity_store_id: str, user_id: str,
    ) -> str | None:
        try:
            u = ids_client.describe_user(
                IdentityStoreId=identity_store_id, UserId=user_id,
            )
        except ClientError as exc:
            logger.info(
                "identitystore.describe_user %s on store %s: %s",
                user_id, identity_store_id, exc,
            )
            return None
        # Prefer the first email marked Primary; fall back to UserName
        # (which for many setups IS an email).
        for e in u.get("Emails") or []:
            if e.get("Primary") and e.get("Value"):
                return str(e["Value"]).lower()
        for e in u.get("Emails") or []:
            v = e.get("Value")
            if v:
                return str(v).lower()
        uname = str(u.get("UserName") or "")
        if "@" in uname:
            return uname.lower()
        return None

    # ── Orphan-assignment discovery (Identity Center "User not found") ──

    async def list_scope_orphan_assignments(
        self, scope: CloudScopeRef,
    ) -> list[OrphanAssignment]:
        """Walk Identity Center assignments; surface those whose
        ``PrincipalId`` no longer resolves in the Identity Store.

        Only meaningful for SSO mode + a real Identity Center instance.
        IAM mode and SSO-mode-without-Identity-Center return empty:
        the IAM-user-as-principal model collapses orphans by
        construction (deleting the user removes the binding).
        ``standalone_iam`` scopes (AWS SG, Tencent — single-tenant
        with one degenerate scope) likewise have no Identity Center
        assignments to enumerate.
        """
        if self.account_mode == "iam":
            return []
        # Identity Center assignments are bound to AWS Org member
        # accounts; ``standalone_iam`` scopes use the same provider
        # but a different binding model. Returning [] keeps the API
        # surface simple instead of letting a confusing AWS
        # ValidationException bubble up to the admin.
        if scope.scope_type != "aws_account":
            return []
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        identity_store_id = str(
            self.config.get("sso_identity_store_id") or "",
        ).strip()
        if not instance_arn or not identity_store_id:
            return []
        return await asyncio.to_thread(
            self._list_scope_orphan_assignments_sync, scope,
        )

    def _list_scope_orphan_assignments_sync(
        self, scope: CloudScopeRef,
    ) -> list[OrphanAssignment]:
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        identity_store_id = str(
            self.config.get("sso_identity_store_id") or "",
        ).strip()
        sso = self._boto_client(
            "sso-admin", region=self.config.get("sso_region"),
        )
        ids = self._boto_client(
            "identitystore", region=self.config.get("sso_region"),
        )
        account_id = scope.external_id

        ps_arns: list[str] = []
        try:
            paginator = sso.get_paginator(
                "list_permission_sets_provisioned_to_account",
            )
            for page in paginator.paginate(
                InstanceArn=instance_arn, AccountId=account_id,
            ):
                ps_arns.extend(page.get("PermissionSets") or [])
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code in ("AuthFailure", "InvalidClientTokenId"):
                raise ProviderAuthError(str(exc)) from exc
            logger.info(
                "AWS list_orphan_assignments: list_permission_sets %s "
                "on %s: %s", code, account_id, exc,
            )
            return []

        # Resolve each permission set's display name once so the admin's
        # confirm-cleanup table reads "DBaaS-Prod-Admin-Role" instead of
        # the long ARN. Fall back to ARN suffix if describe fails.
        ps_name_cache: dict[str, str] = {}

        def _ps_name(arn: str) -> str:
            if arn in ps_name_cache:
                return ps_name_cache[arn]
            try:
                desc = sso.describe_permission_set(
                    InstanceArn=instance_arn, PermissionSetArn=arn,
                ).get("PermissionSet") or {}
                name = str(desc.get("Name") or "")
            except ClientError:
                name = ""
            if not name:
                # ``arn:...:permissionSet/ssoins-XXX/ps-YYY`` → ``ps-YYY``
                name = arn.rsplit("/", 1)[-1]
            ps_name_cache[arn] = name
            return name

        # Cache resolution by principal id — many permission sets typically
        # bind the same principals, and ``describe_user`` is the slow call.
        # The cache value is True (resolves) / False (orphan).
        resolved: dict[str, bool] = {}
        out: list[OrphanAssignment] = []
        for ps_arn in ps_arns:
            try:
                paginator = sso.get_paginator("list_account_assignments")
                for page in paginator.paginate(
                    InstanceArn=instance_arn,
                    AccountId=account_id,
                    PermissionSetArn=ps_arn,
                ):
                    for asgn in page.get("AccountAssignments") or []:
                        ptype = str(asgn.get("PrincipalType") or "").upper()
                        pid = str(asgn.get("PrincipalId") or "")
                        if ptype != "USER" or not pid:
                            # Group / unknown principals are out of scope
                            # for this v1; the cleanup endpoint only
                            # handles user-typed orphans.
                            continue
                        is_alive = resolved.get(pid)
                        if is_alive is None:
                            try:
                                ids.describe_user(
                                    IdentityStoreId=identity_store_id,
                                    UserId=pid,
                                )
                                is_alive = True
                            except ClientError as exc:
                                code = (exc.response or {}).get(
                                    "Error", {},
                                ).get("Code", "")
                                # Only ``ResourceNotFoundException`` is
                                # an orphan signal; throttling / 5xx /
                                # auth errors must not nominate live
                                # users for deletion. Treat as alive
                                # so we err on the safe side and let
                                # the next sync re-evaluate.
                                if code == "ResourceNotFoundException":
                                    is_alive = False
                                else:
                                    logger.info(
                                        "identitystore.describe_user "
                                        "%s on store %s: %s — treating "
                                        "as alive",
                                        pid, identity_store_id, exc,
                                    )
                                    is_alive = True
                            resolved[pid] = is_alive
                        if is_alive:
                            continue
                        out.append(OrphanAssignment(
                            principal_id=pid,
                            role_key=ps_arn,
                            role_display_name=_ps_name(ps_arn),
                            scope=scope,
                            external_ref=(
                                f"{account_id}::{ps_arn}::{pid}"
                            ),
                            recovered_email=None,
                            # Naive-UTC matches the rest of the codebase's
                            # convention (see _utc_now_naive in
                            # cmdb/lifecycle_router.py); ``utcnow()`` is
                            # deprecated in 3.12 and inconsistent with
                            # the timezone-aware paths.
                            last_seen_at=datetime.now(UTC).replace(
                                tzinfo=None,
                            ),
                            raw={
                                "account_id": account_id,
                                "permission_set_arn": ps_arn,
                                "identity_store_user_id": pid,
                            },
                        ))
            except ClientError as exc:
                logger.info(
                    "AWS list_account_assignments orphan-scan on "
                    "%s / %s skipped: %s",
                    account_id, ps_arn, exc,
                )
                continue
        return out

    async def delete_orphan_assignment(
        self, scope: CloudScopeRef, orphan: OrphanAssignment,
    ) -> CloudGrantResult:
        """Delete one Identity Center assignment for an orphaned
        principal. Idempotent: ResourceNotFoundException = already gone.
        """
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        if not instance_arn:
            return CloudGrantResult(
                ok=False,
                error_message="sso_instance_arn not configured",
            )
        return await asyncio.to_thread(
            self._delete_orphan_assignment_sync, scope, orphan,
        )

    def _delete_orphan_assignment_sync(
        self, scope: CloudScopeRef, orphan: OrphanAssignment,
    ) -> CloudGrantResult:
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        sso = self._boto_client(
            "sso-admin", region=self.config.get("sso_region"),
        )
        try:
            sso.delete_account_assignment(
                InstanceArn=instance_arn,
                TargetId=scope.external_id,
                TargetType="AWS_ACCOUNT",
                PermissionSetArn=orphan.role_key,
                PrincipalType="USER",
                PrincipalId=orphan.principal_id,
            )
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code == "ResourceNotFoundException":
                return CloudGrantResult(ok=True, already_absent=True)
            return CloudGrantResult(ok=False, error_message=str(exc))
        return CloudGrantResult(
            ok=True, external_ref=orphan.external_ref,
        )

    # ── Live-principal assignment enumeration (leaver sweep) ──────

    async def list_principal_assignments(
        self, email: str,
    ) -> list[PrincipalAssignment]:
        """Enumerate every Identity Center account assignment a live
        principal holds, resolved from ``email``.

        Powers the offboarding leaver-sweep: delete the departing
        person's assignments while their Identity Store user still
        exists, so untracked (console-created) assignments never become
        blank-name orphans after SCIM deprovision deletes the user.

        SSO mode + a real Identity Center instance only. IAM mode and
        SSO-without-Identity-Center return ``[]``: the IAM-user model
        collapses assignments into the user object (deleting the user
        removes the binding), so there is nothing left to sweep
        separately.
        """
        if self.account_mode == "iam":
            return []
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        identity_store_id = str(
            self.config.get("sso_identity_store_id") or "",
        ).strip()
        if not instance_arn or not identity_store_id:
            return []
        return await asyncio.to_thread(
            self._list_principal_assignments_sync, email,
        )

    def _list_principal_assignments_sync(
        self, email: str,
    ) -> list[PrincipalAssignment]:
        instance_arn = str(self.config.get("sso_instance_arn") or "").strip()
        try:
            user_id = self._resolve_identity_store_user_id(email)
        except CloudPrincipalNotFoundError:
            # Principal already gone from the Identity Store (SCIM
            # deprovision ran first). Nothing to sweep by attribution —
            # any residual assignment is now a blank-name orphan handled
            # by the orphan-scan surface. Return empty rather than raise.
            logger.info(
                "AWS list_principal_assignments: no Identity Store user "
                "for %s — nothing to sweep (already deprovisioned)", email,
            )
            return []
        except (ClientError, BotoCoreError):
            # Transient / auth failure resolving the principal. Raise so
            # the caller surfaces it and retries — we must NOT silently
            # report "no assignments" (which would close the sweep as a
            # clean no-op while the person keeps their access).
            raise

        sso = self._boto_client(
            "sso-admin", region=self.config.get("sso_region"),
        )

        # ``list_account_assignments_for_principal`` (sso-admin) returns
        # every (AccountId, PermissionSetArn) the principal holds across
        # the whole Org in one paginated call — far cheaper than walking
        # every account × permission set. It is a relatively recent API;
        # guard for an older botocore that lacks it and fall back to
        # "no rows" (the caller degrades gracefully — the manual orphan
        # scan remains the backstop) rather than crashing the offboarding
        # advance tick.
        try:
            paginator = sso.get_paginator(
                "list_account_assignments_for_principal",
            )
        except Exception as exc:  # noqa: BLE001 — OperationNotPageableError / unknown op
            logger.warning(
                "AWS list_principal_assignments: botocore lacks "
                "list_account_assignments_for_principal (%s) — leaver "
                "sweep degraded to orphan-scan backstop for %s",
                exc, email,
            )
            return []

        ps_name_cache: dict[str, str] = {}

        def _ps_name(arn: str) -> str:
            if arn in ps_name_cache:
                return ps_name_cache[arn]
            try:
                desc = sso.describe_permission_set(
                    InstanceArn=instance_arn, PermissionSetArn=arn,
                ).get("PermissionSet") or {}
                name = str(desc.get("Name") or "")
            except ClientError:
                name = ""
            if not name:
                name = arn.rsplit("/", 1)[-1]
            ps_name_cache[arn] = name
            return name

        out: list[PrincipalAssignment] = []
        try:
            for page in paginator.paginate(
                InstanceArn=instance_arn,
                PrincipalId=user_id,
                PrincipalType="USER",
            ):
                for asgn in page.get("AccountAssignments") or []:
                    account_id = str(asgn.get("AccountId") or "")
                    ps_arn = str(asgn.get("PermissionSetArn") or "")
                    if not account_id or not ps_arn:
                        continue
                    out.append(PrincipalAssignment(
                        account_id=account_id,
                        principal_id=user_id,
                        role_key=ps_arn,
                        role_display_name=_ps_name(ps_arn),
                        external_ref=f"{account_id}::{ps_arn}::{user_id}",
                        raw={
                            "account_id": account_id,
                            "permission_set_arn": ps_arn,
                            "identity_store_user_id": user_id,
                        },
                    ))
        except ClientError as exc:
            code = (exc.response or {}).get("Error", {}).get("Code", "")
            if code in ("AuthFailure", "InvalidClientTokenId"):
                raise ProviderAuthError(str(exc)) from exc
            # Any other cloud-API failure mid-enumeration MUST raise: a
            # partial list handed to the sweep would leave some of the
            # leaver's assignments un-deleted while reporting success.
            logger.warning(
                "AWS list_principal_assignments for %s (user_id=%s): %s",
                email, user_id, exc,
            )
            raise
        return out

    def _list_scope_grants_iam_sync(
        self, scope: CloudScopeRef,
    ) -> list[DiscoveredGrant]:
        iam = self._boto_client("iam", region="us-east-1")
        out: list[DiscoveredGrant] = []
        try:
            paginator = iam.get_paginator("list_users")
            for page in paginator.paginate():
                for u in page.get("Users") or []:
                    user_name = str(u.get("UserName") or "")
                    if not user_name:
                        continue
                    # Prefer the ``email`` tag if the admin set one,
                    # otherwise treat the UserName as an email when
                    # it has ``@``. Otherwise skip (can't resolve).
                    email = self._iam_user_email(iam, user_name)
                    if not email:
                        continue
                    # Use the paginator family ``boto3`` exposes for
                    # ``list_attached_user_policies`` so a future raise
                    # of the per-user attached-policy quota (AWS default
                    # is 10, can be raised via support ticket to 20) or
                    # a server-side response truncation can't silently
                    # cut the tail. Same defensive shape as the
                    # ``list_users`` paginator a few lines above and the
                    # UCloud / Tencent / Aliyun matching fixes. The
                    # single-call form (``iam.list_attached_user_policies
                    # (...).get("AttachedPolicies")``) returns at most
                    # 100 by default and sets ``IsTruncated=True`` for
                    # the rest — those rest rows then vanish until the
                    # next sync, when grants_sync interprets the
                    # absence as drift and mass-revokes.
                    attached: list[dict[str, Any]] = []
                    try:
                        attached_paginator = iam.get_paginator(
                            "list_attached_user_policies",
                        )
                        for ap_page in attached_paginator.paginate(
                            UserName=user_name,
                        ):
                            attached.extend(
                                ap_page.get("AttachedPolicies") or [],
                            )
                    except ClientError as exc:
                        # Round-42 audit: ``except ClientError: attached
                        # = []`` silently drops user X's policies on a
                        # transient throttle — grants_sync then mass-
                        # revokes X's grants on the next sweep. ``NoSuch
                        # Entity`` (user just deleted) is the only
                        # legitimate empty-result code; everything else
                        # (throttle / 5xx / auth) MUST abort the snapshot.
                        code = (
                            (exc.response or {})
                            .get("Error", {}).get("Code", "")
                        )
                        if code in ("NoSuchEntity",):
                            attached = []
                        else:
                            logger.warning(
                                "AWS IAM list_attached_user_policies(%s) "
                                "failed (%s) — refusing to emit user "
                                "with zero policies (would mass-revoke "
                                "their grants).",
                                user_name, code or "no-code",
                            )
                            raise
                    for p in attached:
                        arn = str(p.get("PolicyArn") or "")
                        if not arn:
                            continue
                        out.append(DiscoveredGrant(
                            principal_email=email,
                            principal_type="user",
                            role_key=arn,
                            role_display_name=str(p.get("PolicyName") or ""),
                            external_ref=f"{scope.external_id}::{user_name}::{arn}",
                            scope=scope,
                            granted_at=(
                                u.get("CreateDate")
                                if isinstance(u.get("CreateDate"), datetime)
                                else None
                            ),
                            raw={
                                "user_name": user_name,
                                "policy_arn": arn,
                                "policy_name": p.get("PolicyName"),
                            },
                        ))
        except (BotoCoreError, ClientError) as exc:
            # Round-42 audit: same mass-revoke bug class as the UCloud
            # fix. A transient throttle / 5xx / network blip during
            # ``list_users`` (or mid-pagination) used to return a
            # partial accumulator + log INFO. The grants-sync sweep
            # treats the snapshot as authoritative, so the next drift
            # tick mass-revokes every Helix-tracked AWS IAM grant on
            # this scope. Raise so the sweep aborts ``status=failed``
            # rather than committing a corrupting empty/partial snapshot.
            code = (
                (exc.response or {}).get("Error", {}).get("Code", "")
                if isinstance(exc, ClientError) else ""
            )
            if code in ("AuthFailure", "InvalidClientTokenId"):
                raise ProviderAuthError(str(exc)) from exc
            logger.warning(
                "AWS IAM list_users failed (%s) — refusing to write "
                "partial grants snapshot (would mass-revoke against "
                "drift): %s",
                code or "no-code", exc,
            )
            raise
        return out

    def _iam_user_email(self, iam, user_name: str) -> str | None:
        # 1. UserName is itself an email (most pingcap.com IAM users).
        if "@" in user_name:
            return user_name.lower()
        # 2. Fall back to per-user tags. Production AWS SG uses ``Owner``
        #    (capital O) almost universally for bare-name users like
        #    ``li.da`` / ``breezewish``; ``email`` / ``Email`` are
        #    documented but rare. Match all common spellings case-
        #    insensitively, and skip values that don't look like emails
        #    (``Owner: BedrockAPIKey-d6l8`` is a label, not an address).
        #    Priority: explicit ``email``-style key > ``owner`` fallback.
        try:
            tags = iam.list_user_tags(UserName=user_name).get("Tags") or []
        except ClientError:
            return None
        normalised = {
            str(t.get("Key") or "").lower(): str(t.get("Value") or "").strip()
            for t in tags
        }
        for key in ("email", "user_email", "email_address", "owner"):
            v = normalised.get(key, "")
            if "@" in v:
                return v.lower()
        return None

    # ── Last login (IAM mode: PasswordLastUsed + AK LastUsed) ──

    async def get_last_login(
        self, ctx: CloudAccessContext,
    ) -> datetime | None:
        if not _BOTO3_AVAILABLE or self.account_mode != "iam":
            return None
        return await asyncio.to_thread(
            self._get_iam_last_activity_sync,
            self._iam_user_name_for(ctx.principal_email),
        )

    def _get_iam_last_activity_sync(self, user_name: str) -> datetime | None:
        iam = self._boto_client("iam")
        candidates: list[datetime] = []
        try:
            u = iam.get_user(UserName=user_name).get("User") or {}
            if u.get("PasswordLastUsed"):
                candidates.append(u["PasswordLastUsed"])
        except ClientError:
            return None
        try:
            keys = iam.list_access_keys(UserName=user_name).get("AccessKeyMetadata") or []
            for k in keys:
                info = iam.get_access_key_last_used(
                    AccessKeyId=k["AccessKeyId"],
                ).get("AccessKeyLastUsed") or {}
                if info.get("LastUsedDate"):
                    candidates.append(info["LastUsedDate"])
        except ClientError as exc:
            # Falling back to PasswordLastUsed only — log so the gap
            # surfaces in audit. Common cause: SP missing
            # iam:ListAccessKeys on this user.
            logger.warning(
                "AWS list_access_keys / get_access_key_last_used "
                "failed for user %s (last-login uses password only): %s",
                user_name, exc,
            )
        if not candidates:
            return None
        latest = max(candidates)
        if isinstance(latest, datetime) and latest.tzinfo is None:
            from datetime import UTC
            latest = latest.replace(tzinfo=UTC)
        return latest

    # ── BigQuery CUR helpers (used when billing_source="bigquery") ─
    #
    # PingCAP's prod AWS Org publishes its CUR (Cost and Usage Report)
    # to a per-linked-account Parquet export on GCS, surfaced as
    # external BigQuery tables under ``gcp-digital-bi.aws_nonprod_billing``.
    # The IAM user the AWS driver authenticates as (``user/PingCAP``)
    # was never granted Cost Explorer's ``ce:GetCostAndUsage*`` family —
    # querying CE from prod returns ``AccessDeniedException`` for every
    # call. Rather than chasing IAM grants across 60+ linked accounts,
    # we read the same CUR pchub reads. AWS SG (single-account IAM
    # mode, separate creds with CE access) leaves ``billing_source=
    # cost_explorer`` and the existing CE codepath fires for it.
    #
    # All three billing methods share the same routing pattern: if
    # ``self._billing_source == "bigquery"`` AND ``bq_billing`` was
    # threaded in, dispatch to the BQ helper; else fall through to CE.
    # ``bq_billing`` is pre-resolved in the service layer (cost_service
    # / lifecycle_service / lifecycle_router) so the driver never
    # touches a DB session — same pattern as ``cny_per_usd`` and
    # ``principal_tag_keys``.

    # ─── BigQuery query safety caps ─────────────────────────────
    #
    # ``maximum_bytes_billed`` — hard ceiling on the data scanned
    # per query. Mirrors the GCP driver's posture (PR-235 onwards):
    # without this guard, a malformed predicate (e.g. losing the
    # year/month partition filter through a refactor) could scan
    # every month of every linked-account table — ~250 GB for the
    # prod 60-account / 13-month CUR dataset.
    #
    # The original 50 GiB ceiling was sized against the legacy CUR 1.0
    # ``aws_nonprod_billing`` dataset (per_account tables). It is TOO
    # LOW for the CUR 2.0 ``aws_billing_no_payer`` dataset: there, all
    # 60 linked accounts share one DAY-partitioned month-table and the
    # per-scope ``line_item_usage_account_id`` filter is NOT a
    # clustering/partition key, so EVERY scope's one-month query scans
    # the whole month's partitions for every account. A 30-day
    # ``cmdb.cost_facts_monthclose`` re-pull of the AWS main account
    # measured 73.57 GB (~68.5 GiB) on a prod BQ dry-run (2026-06-18) —
    # over the 50 GiB cap — so the month-close aborted with
    # ``bytesBilledLimitExceeded`` for every AWS scope and never
    # completed (discovered 2026-06-17 during the #1197/#1200
    # reconciliation backfill). Raised to 128 GiB: comfortably above a
    # single-calendar-month per-scope scan (~70 GiB) with ~1.8× growth
    # headroom, while still aborting a genuine runaway (a pp-removal
    # bug scans the month-table's full far-future annual-reservation
    # partitions → 150 GB+; a lost-window-filter tenant-wide scan →
    # 250 GB+). Per-query stays bounded to ~1 month because
    # ``_bq_cost_facts_windows`` splits multi-month windows by calendar
    # month (a 13-month backfill = 13 queries, not one 900 GB scan).
    # BQ aborts with a clear ``Query exceeded limit for bytes billed``
    # error rather than silently completing on a runaway query.
    _BQ_MAX_BYTES_BILLED = 128 * 1024 ** 3
    # ``job_timeout_ms`` — wall-clock cap. Prod-observed tenant-wide
    # CUR queries land ≤30 s; 5 min provides a 10× safety margin and
    # ensures a wedged BQ slot can't hold the worker thread forever
    # (worker would otherwise stay alive past the cron's outer lock
    # TTL of 900 s and double-fire on retry).
    _BQ_JOB_TIMEOUT_MS = 5 * 60 * 1000

    # Max window span (inclusive days) for the batched all-accounts
    # cost-facts path (see ``_bq_cost_facts_batch_results``). The daily
    # cron window is 3 days and a monthly close is ~31; 70 covers both
    # plus a ~2-month operator backfill while bounding how many
    # accounts' facts the per-run cache holds at once. Wider windows
    # fall back to the per-account streaming path (memory-light, one
    # scope at a time) — rare, operator-initiated, and chunkable to
    # ≤2 months to regain the cheap path. Paired with a ``per_product``
    # grain gate so row cardinality stays bounded regardless of span.
    _BQ_COST_FACTS_BATCH_MAX_SPAN_DAYS = 70

    # ── BQ CUR SQL — Savings Plan discount semantics ───────────────
    #
    # DO NOT filter out ``line_item_line_item_type = 'SavingsPlanNegation'``
    # rows or ``line_item_product_code = 'ComputeSavingsPlans'`` rows
    # from CUR SUMs. Those rows ARE the SP discount and the SP
    # recurring fee respectively; excluding them inflates reported
    # cost by the SP discount rate (typically 28–40%; reproduced
    # prod incident was 34.15%).
    #
    # A single SP-covered EC2 hour produces THREE rows in CUR:
    #   1. ``Usage``                     — unblended_cost = +$X (on-demand list)
    #   2. ``SavingsPlanNegation``       — unblended_cost = -$Y (SP discount)
    #                                      product_code  = 'ComputeSavingsPlans'
    #   3. ``SavingsPlanCoveredUsage``   — unblended_cost =  0 (tracking only;
    #                                      effective cost lives in
    #                                      savings_plan_savings_plan_effective_cost)
    #
    # On the PAYER account's CUR additionally:
    #   4. ``SavingsPlanRecurringFee``   — unblended_cost = +$Z (the SP commitment)
    #
    # ``SUM(line_item_unblended_cost)`` over ALL rows = ($X − $Y) on
    # linked account + $Z on payer = the actual invoiced amount.
    # That matches AWS Cost Explorer's "Net unblended cost" / monthly
    # invoice view.
    #
    # History: the BQ paths shipped with the negation row excluded by
    # mistake at the time the BQ source was introduced; the inflation
    # reproduced against AWS account 552185537300 (April 2026: Helix
    # $8,360.87 vs invoice $5,505.88, ~34% over). Fixed in PR #339.
    # **Do not reintroduce the filter without a corresponding
    # amortized-cost CASE expression.**

    # Map from Helix's canonical dimension aliases (the public
    # ``group_by`` contract) to CUR column names. Keep parity with
    # ``_CE_DIMENSION_ALIASES`` — any CE alias we expose must have a
    # BQ-side equivalent here OR explicitly raise in BQ mode (the
    # caller's UI should never offer a dimension that silently
    # half-works depending on integration). ``tag:KEY`` is handled
    # separately via ``_bq_tag_column``.
    _BQ_DIMENSION_COLUMN_MAP: dict[str, str] = {
        "service": "product_product_name",
        "linked_account": "line_item_usage_account_id",
        "account": "line_item_usage_account_id",
        "region": "product_region",
        "usage_type": "line_item_usage_type",
        "operation": "line_item_operation",
        "availability_zone": "line_item_availability_zone",
        "instance_type_family": "product_instance_type_family",
        # CE's PURCHASE_TYPE dimension (OnDemand / Reserved / Spot /
        # SavingsPlanCoveredUsage / …) maps to CUR's
        # ``line_item_line_item_type``: the same surface, different
        # source-of-truth column.
        "purchase_type": "line_item_line_item_type",
    }

    # Subset of ``_BQ_DIMENSION_COLUMN_MAP`` whose default values point
    # at CUR-1.0-style flat ``product_<key>`` columns — these move into
    # the nested ``product.key_value`` ARRAY<STRUCT> in CUR 2.0. The
    # value here is the BARE key inside the array (no ``product_``
    # prefix), passed straight to ``_bq_product_field_expression``.
    _BQ_DIMENSION_PRODUCT_KEY_FOR_ALIAS: dict[str, str] = {
        "service": "product_name",
        "region": "region",
        "instance_type_family": "instance_type_family",
    }

    @classmethod
    def _bq_dimension_expression(
        cls, tag_schema: str, alias: str,
    ) -> str | None:
        """Resolve a non-tag dimension alias to a SQL expression.

        Returns ``None`` when the alias is not a supported dimension —
        callers fall through to a clear ValueError. For CUR 2.0
        (``nested`` tag schema) the ``service`` / ``region`` /
        ``instance_type_family`` aliases resolve to a correlated
        subquery against ``product.key_value`` since CUR 2.0 collapses
        every ``product_*`` field into that nested record. Other
        aliases (``linked_account`` / ``usage_type`` / ``operation``
        / ``availability_zone`` / ``purchase_type``) map to
        ``line_item_*`` columns that are top-level in both CUR 1.0
        and CUR 2.0.
        """
        key = alias.lower()
        if key not in cls._BQ_DIMENSION_COLUMN_MAP:
            return None
        if tag_schema == "nested" and key in cls._BQ_DIMENSION_PRODUCT_KEY_FOR_ALIAS:
            return cls._bq_product_field_expression(
                tag_schema,
                cls._BQ_DIMENSION_PRODUCT_KEY_FOR_ALIAS[key],
            )
        return cls._BQ_DIMENSION_COLUMN_MAP[key]

    # ── 业务折前成本 (list cost) — CUR line-type whitelist ──────────
    #
    # 研发口径（与云管理团队 2026-07-27 用真实 6 月 CUR 核验定稿）：
    #   * ``Usage`` / ``SavingsPlanCoveredUsage`` → line_item_unblended_cost
    #     （spot / 阶梯 / 实际费率如实保留 = 部门功劳；SP 覆盖行的
    #     unblended 实测 == 按需等价，与 Negation 行精确对冲）；
    #   * ``DiscountedUsage``（RI 覆盖，当前无 RI — 预留分支）→ 用
    #     pricing_public_on_demand_cost 还原（该行 unblended 可能为 0）；
    #   * 其余一律不计（白名单制——新行类型默认出局）：
    #     SavingsPlanNegation / SavingsPlanRecurringFee /
    #     SavingsPlanUpfrontFee / RIFee / Fee / Credit / Refund / Tax /
    #     EdpDiscount / PrivateRateDiscount / BundledDiscount /
    #     FlatRateSubscription …（SP・RI 采购成本、抵扣、credit、税费、
    #     退款、账单调整——不代表业务资源原始消耗）。
    # 按需牌价（⓪，「研发自主节省」= 牌价 − list）= 白名单行的
    # pricing_public_on_demand_cost。
    _LIST_WHITELIST_TYPES: tuple[str, ...] = ("Usage", "SavingsPlanCoveredUsage")
    _LIST_RI_RESTORE_TYPE = "DiscountedUsage"

    @classmethod
    def _bq_list_measure_exprs(
        cls, *, line_type_expr: str, gross_expr: str,
    ) -> tuple[str, str]:
        """SQL for (native_list, native_public) under the whitelist.

        ``line_type_expr`` / ``gross_expr`` are pre-resolved column
        expressions (``CAST(NULL AS …)`` on legacy CUR 1.0 tables that
        lack the column — the CASE then yields NULL end-to-end, which
        the row builders persist as "measure not wired", never a fake 0).
        """
        wl = ", ".join(f"'{t}'" for t in cls._LIST_WHITELIST_TYPES)
        native_list = f"""SUM(CASE
                    WHEN {line_type_expr} IN ({wl})
                      THEN line_item_unblended_cost
                    WHEN {line_type_expr} = '{cls._LIST_RI_RESTORE_TYPE}'
                      THEN {gross_expr}
                    ELSE NULL END)"""
        native_public = f"""SUM(CASE
                    WHEN {line_type_expr} IN ({wl}, '{cls._LIST_RI_RESTORE_TYPE}')
                      THEN {gross_expr}
                    ELSE NULL END)"""
        return native_list, native_public

    @staticmethod
    def _classify_cur_bucket(
        line_types: list[str] | None,
        *, discount_bucket: str | None,
        product_code: str,
        native: float,
    ) -> str:
        """Map a CUR fact group's row types onto the canonical cost bucket.

        Priority: whitelist usage first — a group mixing Usage with a
        stray Refund keeps the usage classification (the engineering
        list measure is already isolated per-row by the SQL CASE, so
        bucketing only affects the finance waterfall's slice labels).
        Unknown / unclassifiable shapes land in OTHER, never USAGE.
        """
        lt = {str(t) for t in (line_types or []) if t}
        if lt & {
            *AwsCloudDriver._LIST_WHITELIST_TYPES,
            AwsCloudDriver._LIST_RI_RESTORE_TYPE,
        }:
            return COST_BUCKET_USAGE
        if discount_bucket:  # EdpDiscount / PrivateRateDiscount rows
            return COST_BUCKET_CONTRACT_DISCOUNT
        if "Tax" in lt:
            return COST_BUCKET_TAX
        if "Refund" in lt:
            return COST_BUCKET_ADJUSTMENT
        if "Credit" in lt:
            return COST_BUCKET_CREDIT
        if "SavingsPlanNegation" in lt:
            return COST_BUCKET_COMMITMENT_DISCOUNT
        if lt & {
            "SavingsPlanRecurringFee", "SavingsPlanUpfrontFee",
            "RIFee", "Fee",
        }:
            return COST_BUCKET_COMMITMENT_FEE
        if product_code.endswith("SavingsPlans"):
            # CUR 1.0 path: SP rows arrive by product_code with no
            # line-type column projected — sign decides fee vs discount.
            return (
                COST_BUCKET_COMMITMENT_DISCOUNT
                if native < 0 else COST_BUCKET_COMMITMENT_FEE
            )
        return COST_BUCKET_OTHER

    @property
    def _billing_source(self) -> str:
        # Normalise — admins may type "BigQuery", "bigQuery", or paste
        # the value with surrounding whitespace; the cron must still
        # route correctly.
        return str(
            self.config.get("billing_source") or "cost_explorer",
        ).strip().lower()

    @staticmethod
    def _bq_tag_column(tag_key: str) -> str:
        """Translate an AWS resource-tag key into the CUR column name.

        AWS exports user-defined resource tags as
        ``resource_tags_user_<lower(key)>``. Non-alphanumeric chars
        in the key get replaced with ``_`` in the column name. Reject
        the empty case loudly so a typo'd ``tag:`` doesn't quietly
        evaluate to ``resource_tags_user_`` (which doesn't exist and
        crashes the query with a confusing column-not-found).
        """
        if not tag_key or not tag_key.strip():
            raise ValueError("tag dimension requires a non-empty key")
        # Normalise the key into the suffix AWS CUR uses. Anything
        # weird becomes ``_`` so the column name is always a valid
        # BQ identifier. Lowercase to match CUR's convention.
        cleaned = "".join(
            (c.lower() if c.isalnum() else "_")
            for c in tag_key.strip()
        )
        return f"resource_tags_user_{cleaned}"

    def _bq_table_for_account(
        self, account_id: str, *, dataset: str, pattern: str,
    ) -> str:
        # ``{account_id}`` is the only token recognised in the pattern;
        # anything else is passed through verbatim. This keeps the
        # template trivially safe (no eval, no general format-string
        # injection from the integration's config).
        if "{account_id}" not in pattern:
            raise RuntimeError(
                f"billing_bq_table_pattern {pattern!r} must contain "
                "{account_id} — got literal pattern with no placeholder",
            )
        name = pattern.replace("{account_id}", account_id)
        return f"{dataset}.{name}"

    # ── CUR 2.0 / per-month-table layout helpers ─────────────────
    #
    # New AWS billing dataset ``gcp-digital-bi.aws_billing_no_payer``
    # uses Data Exports format: one native-partitioned table per month
    # (``aws_billing_{yyyymmdd}`` with yyyymmdd = month first-day), all
    # linked accounts merged inside. The helpers below are layout-aware
    # so the existing CUR 1.0 ``per_account`` path (PingCAP legacy
    # ``aws_nonprod_billing``) keeps working byte-for-byte while the
    # new ``per_month`` path resolves to month-tables + injects a
    # ``WHERE line_item_usage_account_id = ?`` scope filter.

    def _bq_table_for_month(
        self, month_first: date, *, dataset: str, pattern: str,
    ) -> str:
        """Resolve one ``per_month`` table name for the given month.

        ``month_first`` MUST be a date with day=1 — the new dataset's
        table names use the month-first-day yyyymmdd token. Substituted
        verbatim into ``{yyyymmdd}`` so no general format-string
        evaluation runs against admin-supplied config.
        """
        if "{yyyymmdd}" not in pattern:
            raise RuntimeError(
                f"billing_bq_table_pattern {pattern!r} must contain "
                "{yyyymmdd} for table_layout=per_month",
            )
        if month_first.day != 1:
            raise RuntimeError(
                f"_bq_table_for_month: month_first {month_first!r} "
                "must be the first day of a month",
            )
        name = pattern.replace(
            "{yyyymmdd}", month_first.strftime("%Y%m%d"),
        )
        return f"{dataset}.{name}"

    @staticmethod
    def _bq_months_in_window(start: date, end: date) -> list[date]:
        """Enumerate month-first-day dates covering ``[start, end]``.

        Used by the per_month layout to project the requested date
        range onto the set of month-tables to UNION across. Mirrors
        ``_bq_month_predicate``'s month enumeration so the two layouts
        produce identical (year, month) coverage when callers query
        across a window — the per_account leg pruning Hive partitions,
        the per_month leg picking month-tables.
        """
        months: list[date] = []
        cur = date(start.year, start.month, 1)
        end_marker = date(end.year, end.month, 1)
        while cur <= end_marker:
            months.append(cur)
            cur = (
                date(cur.year + 1, 1, 1) if cur.month == 12
                else date(cur.year, cur.month + 1, 1)
            )
        return months

    @classmethod
    def _bq_partition_predicate(
        cls, layout: str, start: date, end: date,
    ) -> tuple[str, list[Any]]:
        """Layout-aware partition pruning predicate.

        Returns ``(sql_fragment, parameters)`` AND-ready alongside the
        per-table ``DATE(line_item_usage_start_date) BETWEEN @start
        AND @end`` filter — that secondary predicate is the day-grain
        gate; this one is the partition-prune gate.

        ``per_account`` (CUR 1.0 EXTERNAL): Hive ``year`` + ``month``
        INT columns. Predicate is ``(year=Y0 AND month=M0) OR (year=Y1
        AND month=M1) OR ...`` for each month the window touches.
        Without this predicate BigQuery would scan every Hive
        partition (years of data, 100s of GB).

        ``per_month`` (CUR 2.0 native DAY-partitioned): the per-table
        BETWEEN predicate is already enough to prune the table's native
        DAY partitions to the window; we still emit a clamp against
        the partition column to keep BQ's optimiser honest and
        symmetrical with the per_account path.
        """
        from google.cloud import bigquery  # type: ignore

        if layout == "per_month":
            # Native DAY partition on line_item_usage_end_date. The
            # per-day filter the callers already emit
            # (``DATE(line_item_usage_start_date) BETWEEN @start AND
            # @end``) refers to ``_start_date`` while the partition
            # column is ``_end_date`` — for any non-boundary row those
            # are the same calendar day, but at month boundaries a
            # midnight-UTC row's start_date can land on the previous
            # day relative to end_date. Add a +2-day pad on the end
            # side to keep all in-window rows visible. The same +2-day
            # pad applies in ``_bq_resolve_tables`` (per_month branch)
            # when enumerating month-tables, so the next-month table
            # is brought into the FROM clause and this predicate's
            # extended bound has rows to clamp against. Both numbers
            # MUST stay in sync — drifting one would silently lose
            # last-day-of-month cost.
            pad_end = end + timedelta(days=2)
            return (
                "(line_item_usage_end_date >= @bq_pp_start "
                "AND line_item_usage_end_date < @bq_pp_end)"
            ), [
                bigquery.ScalarQueryParameter(
                    "bq_pp_start", "TIMESTAMP",
                    datetime(start.year, start.month, start.day, tzinfo=UTC),
                ),
                bigquery.ScalarQueryParameter(
                    "bq_pp_end", "TIMESTAMP",
                    datetime(pad_end.year, pad_end.month, pad_end.day, tzinfo=UTC),
                ),
            ]
        # per_account / CUR 1.0 Hive partitions.
        clauses: list[str] = []
        params: list[Any] = []
        for i, m_first in enumerate(cls._bq_months_in_window(start, end)):
            clauses.append(f"(year = @bq_yr_{i} AND month = @bq_mo_{i})")
            params.append(bigquery.ScalarQueryParameter(
                f"bq_yr_{i}", "INT64", m_first.year,
            ))
            params.append(bigquery.ScalarQueryParameter(
                f"bq_mo_{i}", "INT64", m_first.month,
            ))
        return "(" + " OR ".join(clauses) + ")", params

    # Backwards-compatible alias kept so existing tests pinning the
    # legacy helper name still pass without churn while the rename
    # propagates through the three callers.
    @classmethod
    def _bq_month_predicate(
        cls, start: date, end: date,
    ) -> tuple[str, list[Any]]:
        return cls._bq_partition_predicate("per_account", start, end)

    # ── Tag / service / region projection (CUR 1.0 vs 2.0) ──────
    #
    # CUR 1.0 (``tag_schema='flat'``) lands user-defined tags as one
    # top-level column per tag named ``resource_tags_user_<key>``, with
    # ``product_product_name`` / ``product_region`` also top-level.
    #
    # CUR 2.0 (``tag_schema='nested'``) collapses user tags into a
    # single ``resource_tags`` RECORD column with shape
    # ``STRUCT<key_value ARRAY<STRUCT<key STRING, value STRING>>>`` —
    # the array element ``key`` carries the ``user_`` prefix
    # (``user_owner`` not ``owner``). Same pattern for ``product``,
    # ``cost_category``, and ``discount`` — they're each a RECORD with
    # an array of key/value pairs.
    #
    # The expression helpers below produce the right SQL for each
    # schema so the per-list helpers stay layout-neutral above the SQL
    # template line. Callers receive a SQL EXPRESSION string they drop
    # into SELECT / GROUP BY / WHERE — never an identifier alone, so
    # ``ANY_VALUE(expr)`` style aggregation still works.

    @classmethod
    def _bq_tag_expression(
        cls, tag_schema: str, tag_key: str,
    ) -> str:
        """Return the SQL expression that projects a user-defined tag.

        For CUR 1.0 (``flat``): the bare column name
        ``resource_tags_user_<sanitised>``. For CUR 2.0 (``nested``):
        a correlated subquery against ``resource_tags.key_value`` that
        returns the first matching element's value (or NULL).

        Reuses ``_bq_tag_column`` for the CUR 1.0 column name so the
        sanitisation rule (lowercase, non-alphanumeric→underscore) is
        defined once.
        """
        flat_col = cls._bq_tag_column(tag_key)  # validates non-empty + sanitises
        if tag_schema == "nested":
            # CUR 2.0 array element key is ``user_<sanitised>``. The
            # ``resource_tags_user_`` prefix on the flat column maps
            # to ``user_`` on the array element key.
            nested_key = "user_" + flat_col[len("resource_tags_user_"):]
            # Charset guard — the tag key has already been sanitised
            # by ``_bq_tag_column`` so this is a defence-in-depth
            # assertion; never expect to fire.
            if not cls._BQ_TAG_KEY_RE.match(nested_key):
                raise RuntimeError(
                    f"_bq_tag_expression: sanitised nested key "
                    f"{nested_key!r} contains characters outside "
                    "[A-Za-z0-9_]",
                )
            return (
                "(SELECT v.value FROM "
                "UNNEST(resource_tags.key_value) v "
                f"WHERE v.key = '{nested_key}' LIMIT 1)"
            )
        return flat_col

    # CUR 2.0 nested-array element keys after sanitisation must match
    # this charset — the helper interpolates them as a SQL string
    # literal so we hard-validate to prevent quote / metacharacter
    # injection (defence-in-depth on top of ``_bq_tag_column``'s
    # sanitisation).
    _BQ_TAG_KEY_RE = re.compile(r"^[A-Za-z0-9_]+$")

    @classmethod
    def _bq_product_field_expression(
        cls, tag_schema: str, key: str,
    ) -> str:
        """Project a CUR ``product_<key>`` field, schema-aware.

        CUR 1.0 flattens these (``product_product_name``,
        ``product_region``, …); CUR 2.0 nests them into the
        ``product.key_value`` ARRAY<STRUCT> with bare key names
        (``product_name``, ``region``).
        """
        if not cls._BQ_TAG_KEY_RE.match(key):
            raise RuntimeError(
                f"_bq_product_field_expression: key {key!r} contains "
                "characters outside [A-Za-z0-9_]",
            )
        if tag_schema == "nested":
            return (
                "(SELECT v.value FROM "
                "UNNEST(product.key_value) v "
                f"WHERE v.key = '{key}' LIMIT 1)"
            )
        return f"product_{key}"

    @classmethod
    def _bq_resource_tags_dict_expression(
        cls, tag_schema: str, requested_tags: list[tuple[str, str]] | None,
    ) -> str:
        """Return a SQL expression producing the full tags dict.

        Used by ``_list_cost_facts_bq_sync`` to harvest the
        ``CostFact.tags`` payload. For CUR 1.0 we ANY_VALUE-aggregate
        each requested flat column individually (the caller still
        emits a separate ``tag_<alias>`` per requested tag — this
        helper isn't used in that branch). For CUR 2.0 we aggregate
        the entire ``resource_tags`` ARRAY via ``ANY_VALUE`` so the
        caller can iterate ``row.tag_kv`` to build the dict — single
        SELECT projection instead of N.

        ``requested_tags`` is the same ``[(alias, key)]`` list the CUR
        1.0 path consumes; for ``nested`` schema it's unused (we
        harvest the whole array) but accepted for symmetry.
        """
        if tag_schema == "nested":
            return "ANY_VALUE(resource_tags.key_value)"
        # CUR 1.0 callers don't use this helper — they emit one
        # ``ANY_VALUE(resource_tags_user_<key>) AS tag_<alias>`` per
        # requested tag, gated by per-table schema-presence. Keep
        # this as an explicit raise so a future refactor can't
        # accidentally call us in the flat branch.
        raise RuntimeError(
            "_bq_resource_tags_dict_expression is nested-only; the "
            "flat (CUR 1.0) caller emits per-tag projections",
        )

    # ── Cost-fact tag extraction helpers ────────────────────────
    #
    # Pulled out of ``_list_cost_facts_bq_sync`` so the parsing
    # contract (user_ prefix strip, aws_ skip, 128-char key cap,
    # 256-char value truncation, 50-tag-per-fact cap, malformed-
    # STRUCT counting) is unit-testable with a fake BQ row instead
    # of requiring a full BQ client mock.

    # AWS CUR / billing API tag specs:
    #   - tag key max 128 chars
    #   - tag value max 256 chars
    #   - max 50 tags per resource
    # These limits are mirrored on the read side as defence-in-depth
    # so a future cap bump or a malformed bill can't unbounded-grow
    # our JSON row payload.
    _TAG_KEY_MAX = 128
    _TAG_VALUE_MAX = 256
    _TAGS_PER_FACT_MAX = 50

    @classmethod
    def _cur1_extract_tags_from_row(
        cls, row: Any, tag_aliases: list[tuple[str, str]],
    ) -> dict[str, str]:
        """CUR 1.0 (flat) — read each ``tag_<alias>`` column projected
        by ``_list_cost_facts_bq_sync``'s per-pick SELECT.

        ``alias`` is dynamic so ``getattr`` is the only option, but
        pass no default — every per-pick SELECT projected
        ``tag_{alias}`` for every entry in ``tag_aliases``. A missing
        attribute signals SQL-vs-Python drift and should raise loudly.
        """
        tags: dict[str, str] = {}
        for alias, _tag_key in tag_aliases:
            val = getattr(row, f"tag_{alias}")
            if val:
                tags[alias] = str(val)
        return tags

    @classmethod
    def _cur2_extract_tags_from_row(
        cls, row: Any,
    ) -> tuple[dict[str, str], int]:
        """CUR 2.0 (nested) — parse ``row.tag_kv`` ARRAY<STRUCT<key,
        value>> into a clean user-defined tags dict.

        Returns ``(tags, malformed_struct_count)`` so the caller can
        emit one aggregated WARN per fact row instead of logging
        per-item spam on a degenerate CUR snapshot.

        Filters applied (mirrors AWS / Helix bill-storage contract):
          * ``aws_*`` keys skipped (AWS-managed metadata, not "who
            created this")
          * ``user_`` prefix stripped (CUR 2.0 stamps it on
            customer-defined entries)
          * Over-128-char keys dropped (signal of corrupt CUR)
          * Empty / NULL values dropped
          * Values truncated at 256 chars (URLs in tags survive as
            useful prefixes)
          * Hard cap at 50 tags per fact (matches AWS spec)
        """
        kv = getattr(row, "tag_kv", None)
        tags: dict[str, str] = {}
        malformed_struct_count = 0
        for item in kv or ():
            # google-cloud-bigquery returns each STRUCT as either a
            # dict-ish Row (``item['key']``) or a plain dict
            # depending on client version — handle both via subscript
            # access; ``getattr`` would miss the dict case.
            try:
                raw_k = item["key"]
                raw_v = item["value"]
            except (KeyError, TypeError):
                malformed_struct_count += 1
                continue
            if not raw_k:
                continue
            k = str(raw_k)
            if not k.startswith("user_"):
                continue
            k = k[len("user_"):]
            if not k:
                continue
            if len(k) > cls._TAG_KEY_MAX:
                continue
            if raw_v is None or raw_v == "":
                continue
            tags[k] = str(raw_v)[:cls._TAG_VALUE_MAX]
            if len(tags) >= cls._TAGS_PER_FACT_MAX:
                break
        return tags, malformed_struct_count

    def _bq_query_job_config(self, params: list[Any]) -> Any:
        """Build a ``QueryJobConfig`` with safety caps baked in.

        Every query the driver issues goes through this helper so the
        ``maximum_bytes_billed`` + ``job_timeout_ms`` posture stays
        uniform — easy to audit, impossible to forget on a future
        ``client.query(...)`` addition.
        """
        from google.cloud import bigquery  # type: ignore
        return bigquery.QueryJobConfig(
            query_parameters=params,
            maximum_bytes_billed=self._BQ_MAX_BYTES_BILLED,
            job_timeout_ms=self._BQ_JOB_TIMEOUT_MS,
        )

    def _bq_client(self, credentials_json: dict[str, Any]) -> Any:
        """Build a BigQuery client from the SA JSON passed in by the
        service layer.

        Scope MUST be ``.../auth/bigquery`` (NOT
        ``.../auth/bigquery.readonly`` — that scope grants metadata
        listing only and 403s on ``Jobs.InsertJob`` which every BQ
        query goes through, even reads — confirmed via prod probe
        2026-05-15, also burnt the GCP driver's first cut). The IAM
        layer is what restricts the SA to read-only at the actual
        permission boundary.
        """
        from google.cloud import bigquery  # type: ignore
        from google.oauth2 import service_account  # type: ignore

        if not isinstance(credentials_json, dict):
            raise RuntimeError(
                "AWS BQ billing: credentials_json must be a parsed "
                f"SA dict, got {type(credentials_json).__name__}",
            )
        project_id = credentials_json.get("project_id")
        if not project_id:
            raise RuntimeError(
                "AWS BQ billing: SA credentials missing project_id",
            )
        creds = service_account.Credentials.from_service_account_info(
            credentials_json,
            scopes=["https://www.googleapis.com/auth/bigquery"],
        )
        # Pin BQ jobs to the SA's home project, NOT the dataset's
        # project. Cross-project reads work because the dataset
        # has dataViewer granted on the SA explicitly. Pre-fix the
        # GCP driver hit ``BigQuery API not enabled`` when the SA
        # lived in a different project than the export — same gotcha
        # applies here.
        return bigquery.Client(credentials=creds, project=project_id)

    # BigQuery identifier charset for project / dataset / table tokens.
    # GCP project IDs allow lowercase letters, digits, and hyphens (6-30
    # chars); dataset and table IDs allow letters, digits, and
    # underscores. We take the union of both since ``dataset`` is
    # ``<project>.<dataset_id>`` and we don't want to over-reject. This
    # is the security-hardening guard against an admin (or compromised
    # config) injecting backticks / quotes / SQL keywords through the
    # ``billing_bq_dataset`` or ``billing_bq_table_pattern`` fields,
    # which are interpolated into backtick-quoted identifiers in the
    # CUR SELECT / INFORMATION_SCHEMA queries.
    _BQ_DATASET_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
    _BQ_PATTERN_RE = re.compile(r"^[A-Za-z0-9_{}-]+$")

    @classmethod
    def _bq_resolve_options(
        cls,
        bq_billing: dict[str, Any] | None,
    ) -> tuple[dict[str, Any], str, str, str, str, str]:
        """Validate the ``bq_billing`` kwarg from the service layer.

        Returns ``(credentials_json, dataset, table_pattern,
        table_layout, tag_schema, fact_grain)``. Raises RuntimeError
        loudly when any field is missing — silent fall-through to CE
        on a half-configured BQ source would yield 60 AccessDenied
        errors that look identical to the original prod failure.

        ``table_layout`` selects table-resolution / partition / scope-
        filter strategy: ``per_account`` (CUR 1.0, one external table
        per account, Hive year/month partitions, scope→route to table)
        vs ``per_month`` (CUR 2.0, one native table per month with all
        accounts merged, native DAY partition on
        ``line_item_usage_end_date``, scope→``WHERE
        line_item_usage_account_id``).

        ``tag_schema`` selects tag-projection strategy: ``flat`` (CUR
        1.0, one ``resource_tags_user_<key>`` column per tag with
        per-table schema heterogeneity) vs ``nested`` (CUR 2.0,
        ``resource_tags.key_value`` ARRAY<STRUCT<key,value>>; uniform
        schema across all tables).

        ``fact_grain`` selects the cost-facts aggregation grain:
        ``per_product`` (default — one fact per day × product, synthetic
        ``aws-account:<acc>:<product>`` resource_id, ~hundreds of rows
        per scope per day) vs ``per_resource`` (one fact per day ×
        product × line_item_resource_id, ~thousands of rows per scope
        per day; required for per-resource cost-allocation rules but
        amplifies TiDB upsert volume).
        """
        if not bq_billing:
            raise RuntimeError(
                "AWS billing_source=bigquery but bq_billing kwarg was "
                "not threaded through by the service layer — "
                "configuration defect",
            )
        creds = bq_billing.get("credentials_json")
        if not creds:
            raise RuntimeError(
                "AWS BQ billing: bq_billing.credentials_json missing — "
                "ensure billing_bq_credentials_integration_id points "
                "to a GCP integration with a valid service_account_json",
            )
        dataset = str(bq_billing.get("dataset") or "").strip()
        if not dataset:
            raise RuntimeError(
                "AWS BQ billing: bq_billing.dataset missing — set "
                "billing_bq_dataset on the AWS integration",
            )
        if dataset.count(".") != 1:
            raise RuntimeError(
                f"AWS BQ billing: dataset {dataset!r} must be "
                "fully qualified as ``project.dataset``",
            )
        # Charset-validate so backticks / quotes / whitespace / SQL
        # metacharacters can't break out of the backtick-quoted
        # identifier the dataset name lands inside (e.g. the
        # ``FROM `{dataset}.INFORMATION_SCHEMA.COLUMNS``` form in
        # ``_bq_resolve_table_columns`` and ``_bq_discover_account_tables``,
        # and the per-table backtick-FROM in the CUR data SELECTs).
        # Round-1 security review hardening.
        if not cls._BQ_DATASET_RE.match(dataset):
            raise RuntimeError(
                f"AWS BQ billing: dataset {dataset!r} contains "
                "characters outside the allowed BigQuery identifier "
                "set [A-Za-z0-9_.-]",
            )
        layout = str(bq_billing.get("table_layout") or "per_account").strip().lower()
        if layout not in ("per_account", "per_month"):
            raise RuntimeError(
                f"AWS BQ billing: table_layout {layout!r} is invalid — "
                "expected 'per_account' or 'per_month'",
            )
        tag_schema = str(bq_billing.get("tag_schema") or "flat").strip().lower()
        if tag_schema not in ("flat", "nested"):
            raise RuntimeError(
                f"AWS BQ billing: tag_schema {tag_schema!r} is invalid — "
                "expected 'flat' or 'nested'",
            )
        # Strip BEFORE the ``or`` fallback so a whitespace-only
        # admin entry (`"   "`) falls back to the default, not to
        # the stripped-empty-string that would later fail the
        # placeholder check with a confusing message.
        pattern_raw = str(bq_billing.get("table_pattern") or "").strip()
        pattern_default = (
            "{account_id}_aws_billing_data" if layout == "per_account"
            else "aws_billing_{yyyymmdd}"
        )
        pattern = pattern_raw or pattern_default
        # Same hardening as ``dataset``: pattern is interpolated into
        # a backtick-quoted table identifier downstream after the
        # ``{account_id}`` / ``{yyyymmdd}`` placeholder is substituted
        # (with strict format-validated values — 12-digit account id
        # or 8-digit yyyymmdd). Restrict to BigQuery's table-id
        # charset plus the literal ``{`` ``}`` tokens for placeholders.
        if not cls._BQ_PATTERN_RE.match(pattern):
            raise RuntimeError(
                f"AWS BQ billing: table_pattern {pattern!r} contains "
                "characters outside the allowed set "
                "[A-Za-z0-9_{}-]",
            )
        # Layout-specific placeholder must be present. Catching a
        # mismatch here (rather than in the table-resolution helper)
        # gives a clear diagnostic at the boundary, before any BQ
        # round-trip burns latency / quota.
        required_placeholder = (
            "{account_id}" if layout == "per_account" else "{yyyymmdd}"
        )
        if required_placeholder not in pattern:
            raise RuntimeError(
                f"AWS BQ billing: table_pattern {pattern!r} must "
                f"contain the {required_placeholder!r} placeholder for "
                f"table_layout={layout!r}",
            )
        fact_grain = str(
            bq_billing.get("fact_grain") or "per_product",
        ).strip().lower()
        if fact_grain not in ("per_product", "per_resource"):
            raise RuntimeError(
                f"AWS BQ billing: fact_grain {fact_grain!r} is invalid — "
                "expected 'per_product' or 'per_resource'",
            )
        return creds, dataset, pattern, layout, tag_schema, fact_grain

    @staticmethod
    def _bq_table_columns(table_obj: Any) -> set[str]:
        # PingCAP's ~60 per-account CUR external Parquet tables have
        # heterogeneous schemas — some accounts' CUR was created without
        # "Include resource IDs" toggled (no ``line_item_resource_id``
        # column), some lack one or more ``resource_tags_user_*`` columns
        # because no resource in that account ever carried that tag.
        # Referencing a missing column from a SELECT raises BQ's
        # ``invalidQuery: Unrecognized name`` and fails the whole cron
        # leg. Callers introspect ``Table.schema`` (free — already
        # retrieved when probing for existence) and drop missing
        # columns from SELECT / GROUP BY rather than assuming a uniform
        # CUR layout. Prod incident 2026-05-15 (PR #311 follow-up):
        # cost_facts_sync failed on 3+ linked accounts whose CUR tables
        # were missing ``line_item_resource_id`` because the original
        # PR's INFORMATION_SCHEMA probe only checked the master-account
        # table.
        schema = getattr(table_obj, "schema", None) or []
        cols: set[str] = set()
        for f in schema:
            name = getattr(f, "name", None)
            if name:
                cols.add(str(name))
        return cols

    def _bq_resolve_table_columns(
        self,
        client: Any,
        *,
        dataset: str,
        table_names: list[str],
        columns_of_interest: set[str],
    ) -> dict[str, set[str]]:
        # Bulk schema probe across many tables in a single
        # ``INFORMATION_SCHEMA.COLUMNS`` round trip — used by
        # ``_list_billing_by_dimension_bq_sync`` to learn which CUR
        # tables carry each requested ``resource_tags_user_*``
        # column (both for dim-substitution and for filter-drop).
        # ``_list_billing_daily_bq_sync`` and ``_list_cost_facts_bq_sync``
        # use the per-table ``_bq_table_columns`` helper instead —
        # both are per-scope (single table) and already pay for a
        # ``client.get_table`` existence-probe round trip, so reading
        # ``.schema`` off the returned object is free. Per-table
        # ``get_table`` across 60+ tables would issue 60 round trips;
        # this is one query and the result fits in a small dict.
        # Returns a mapping from FULLY-QUALIFIED table name to the
        # subset of ``columns_of_interest`` actually present.
        #
        # Contract: every entry in ``table_names`` MUST be fully
        # qualified (``<project>.<dataset>.<table>``) AND live in the
        # ``dataset`` argument (which the SQL probes). The result dict
        # always contains one key per requested table (empty set when
        # the table lacks all requested columns).
        from google.cloud import bigquery  # type: ignore

        # Pre-allocate the result so callers can rely on every
        # requested table appearing as a key — keeps the "no columns
        # of interest → return all-empty" early-return semantics
        # consistent with the "schema probe returned no matches"
        # full-execution path.
        result: dict[str, set[str]] = {t: set() for t in table_names}
        if not table_names or not columns_of_interest:
            return result
        # Reject short-name collisions: if two FQ table names map to
        # the same short name (different projects/datasets share a
        # table name, or duplicate input), the schema probe can't
        # disambiguate them. ``_bq_resolve_table_columns`` is only
        # used today with one dataset per call but the signature is
        # general; fail loud rather than silently returning an empty
        # column set for one of the colliding entries (which would
        # land it in the "drop table" bucket downstream). Round-1
        # review finding.
        short_to_fq: dict[str, str] = {}
        for fq in table_names:
            if "." not in fq:
                raise RuntimeError(
                    f"_bq_resolve_table_columns: table_name {fq!r} is "
                    "not fully qualified (expected "
                    "``<project>.<dataset>.<table>``)",
                )
            short = fq.rsplit(".", 1)[-1]
            if short in short_to_fq and short_to_fq[short] != fq:
                raise RuntimeError(
                    f"_bq_resolve_table_columns: short-name collision "
                    f"between {short_to_fq[short]!r} and {fq!r} — "
                    "INFORMATION_SCHEMA probe cannot disambiguate. "
                    "Caller must dedupe or restructure inputs.",
                )
            short_to_fq[short] = fq
        sql = (
            f"SELECT table_name, column_name "
            f"FROM `{dataset}.INFORMATION_SCHEMA.COLUMNS` "
            f"WHERE table_name IN UNNEST(@names) "
            f"AND column_name IN UNNEST(@cols)"
        )
        params = [
            bigquery.ArrayQueryParameter(
                "names", "STRING", sorted(short_to_fq.keys()),
            ),
            bigquery.ArrayQueryParameter(
                "cols", "STRING", sorted(columns_of_interest),
            ),
        ]
        job = client.query(
            sql, job_config=self._bq_query_job_config(params),
        )
        for row in job.result():
            fq = short_to_fq.get(str(row.table_name))
            if fq is not None:
                result[fq].add(str(row.column_name))
        return result

    def _bq_resolve_tables(
        self,
        client: Any,
        *,
        layout: str,
        dataset: str,
        pattern: str,
        scope: CloudScopeRef | None,
        start: date,
        end: date,
    ) -> list[_BqTablePick]:
        """Centralised, layout-aware table resolution.

        Returns a list of ``_BqTablePick`` describing every table the
        caller should UNION over. Each pick carries:

        - ``label``: diagnostic string (account id for per_account,
          YYYY-MM for per_month). Used in log messages and as a
          per-pick identifier; not interpolated into SQL.
        - ``fq_table``: fully-qualified BQ table name to put inside the
          backtick FROM clause.
        - ``account_filter``: when set (always for per_month + scoped
          calls), the caller MUST inject
          ``AND line_item_usage_account_id = @scope_acct`` so the
          query reads only the scoped account's rows from the all-
          accounts month-table. ``None`` for per_account (the table
          name itself already filters by account) and per_month
          tenant-wide (all accounts wanted).
        - ``columns``: present-columns set for per_account (CUR 1.0
          schema heterogeneity — used to drop SELECT/GROUP BY clauses
          for absent columns). ``None`` for per_month (CUR 2.0 has
          uniform schema across all per-month tables, no per-table
          drop needed).

        Per_month skips month-tables that don't exist (e.g. window
        extends past the first CUR-export month) — same NotFound
        graceful-skip semantics as the per_account path's missing
        per-account tables.
        """
        from google.api_core.exceptions import NotFound  # type: ignore

        # Dataset-existence probe — fail loudly on dataset-name typos
        # (PR #404 follow-up).
        #
        # Why both layouts need this: without an upfront probe, a typo
        # in ``billing_bq_dataset`` (or a SA whose access on the dataset
        # was revoked / never granted) produces silent-skip behaviour
        # on BOTH layouts:
        #
        #   per_account scoped: every scope's
        #     ``client.get_table('<typo>.<acc>_aws_billing_data')``
        #     raises NotFound → caught + ``return []`` → cron status ok
        #     with zero rows across all 60 scopes.
        #
        #   per_month (scoped or tenant-wide): every month-table's
        #     ``client.get_table('<typo>.aws_billing_YYYYMMDD')``
        #     raises NotFound → caught + ``continue`` → ``picks=[]`` +
        #     ``return []`` → cron status ok with zero rows for the
        #     whole backfill window.
        #
        # (per_account tenant-wide is the one path that surfaces the
        # typo on its own — ``_bq_discover_account_tables`` issues a
        # SELECT against ``<typo>.INFORMATION_SCHEMA.TABLES`` which BQ
        # rejects with a clear "Dataset ... was not found" error
        # bubbling up to the cron's errors[] list. But that's the
        # minority of calls; the daily cron fans out per scope.)
        #
        # The probe is gated by an instance-level set so subsequent
        # scope iterations within the SAME cron tick (~60 calls)
        # don't issue 60 redundant ``get_dataset`` metadata round
        # trips for the same dataset string. Driver instances are
        # short-lived (one per integration-sync invocation in
        # cost_service / lifecycle_service) so this caches across
        # scopes within a tick without leaking across ticks where a
        # genuinely-new admin config could land.
        self._bq_probe_dataset_exists(client, dataset)

        if layout == "per_account":
            if scope is not None and scope.external_id:
                ext = str(scope.external_id)
                if not (ext.isdigit() and len(ext) == 12):
                    raise RuntimeError(
                        f"AWS BQ billing: scope external_id {ext!r} is "
                        "not a 12-digit AWS account id",
                    )
                table = self._bq_table_for_account(
                    ext, dataset=dataset, pattern=pattern,
                )
                try:
                    table_obj = client.get_table(table)
                except NotFound:
                    # Reaching here MEANS the dataset itself exists (the
                    # probe above already passed) but this account's
                    # CUR table isn't in it yet — gracefully skip with
                    # ``[]`` so the per-scope status stays ``ok``. This
                    # is the legitimate "newly added linked account, CUR
                    # not yet onboarded" case (also covers the Org
                    # master account whose CUR is intentionally not
                    # exported under this pattern).
                    logger.info(
                        "AWS BQ billing (per_account): no CUR table "
                        "for account %s — skipping (table %s)",
                        ext, table,
                    )
                    return []
                return [
                    _BqTablePick(
                        label=ext,
                        fq_table=table,
                        account_filter=None,
                        columns=self._bq_table_columns(table_obj),
                    ),
                ]
            # Tenant-wide per_account — schema heterogeneity means the
            # caller MAY want per-table columns. We don't probe here;
            # callers that need column-presence info call
            # ``_bq_resolve_table_columns`` explicitly. Leaving
            # ``columns=None`` lets callers detect "tenant-wide, no
            # per-table schema lookup yet" cleanly.
            return [
                _BqTablePick(
                    label=acct, fq_table=fq,
                    account_filter=None, columns=None,
                )
                for acct, fq in self._bq_discover_account_tables(
                    client, dataset=dataset, pattern=pattern,
                )
            ]

        # per_month layout (CUR 2.0).
        account_filter: str | None = None
        if scope is not None and scope.external_id:
            ext = str(scope.external_id)
            if not (ext.isdigit() and len(ext) == 12):
                raise RuntimeError(
                    f"AWS BQ billing: scope external_id {ext!r} is "
                    "not a 12-digit AWS account id",
                )
            account_filter = ext

        # Month-table enumeration. Pad the window's end side by the
        # same +2 days that ``_bq_partition_predicate`` per_month uses
        # so a row that started late in month-N (``DAY=last_day_of_N``)
        # but ended early in month-(N+1) — landing in (N+1)'s per-month
        # table by the ``line_item_usage_end_date`` DAY partition — is
        # visible when the caller queries window ``[start, end]``
        # entirely within month-N. AWS posts SP/RI recurring fees and
        # similar midnight-rollover rows under this shape and the
        # day-grain cost for the last day of every month would
        # otherwise silently undercount (~0.1-1% of rows in PingCAP
        # prod CUR 2.0 sample). The partition predicate then clamps
        # the next-month table's scan to just the boundary-straddling
        # rows, so the extra scan cost is bounded to ~the row count
        # of the first ~2 days of the next month (typically tiny vs
        # full-month scan).
        end_padded = end + timedelta(days=2)
        picks: list[_BqTablePick] = []
        for month_first in self._bq_months_in_window(start, end_padded):
            table = self._bq_table_for_month(
                month_first, dataset=dataset, pattern=pattern,
            )
            try:
                client.get_table(table)
            except NotFound:
                # Dataset-level existence is already confirmed by the
                # probe at the top of this method; reaching here means
                # the dataset exists but this MONTH's table isn't in
                # it (window predates CUR-export start, mid-stream
                # export gap). Gracefully skip the month so a 12-month
                # backfill containing one pre-export month doesn't
                # fail the whole window.
                logger.info(
                    "AWS BQ billing (per_month): no CUR table for "
                    "month %s — skipping (table %s)",
                    month_first.strftime("%Y-%m"), table,
                )
                continue
            picks.append(_BqTablePick(
                label=month_first.strftime("%Y-%m"),
                fq_table=table,
                account_filter=account_filter,
                # CUR 2.0 — uniform schema across all per-month tables.
                # Setting None signals callers to skip the per-table
                # schema-drop logic that ``per_account`` requires.
                columns=None,
            ))
        return picks

    def _bq_probe_dataset_exists(
        self, client: Any, dataset: str,
    ) -> None:
        """Probe ``client.get_dataset(dataset)``; raise loudly on
        NotFound; cache-pass on instance scope.

        Caching keyed on the dataset STRING (not on the client identity)
        so two scopes resolving the same dataset within one cron tick
        only pay one round trip. We don't include ``client`` in the
        key because the driver instance only ever serves one
        integration → one credential set → one effective project
        scope; if anything changes, a new driver instance gets built
        (cost_service / lifecycle_service builds one per call).

        Recoverable errors (Forbidden, transient 5xx, throttle) MUST
        propagate unchanged so the operator sees the real failure
        surface in ``scheduled_job_runs.errors`` rather than a
        misleading "dataset missing" wrapper. Only ``NotFound`` (the
        unambiguous "doesn't exist" signal) gets re-wrapped.

        BQ-API NotFound vs Forbidden semantics (verified against the
        PingCAP prod SA setup): a SA with project-level
        ``bigquery.dataViewer`` resolves a typo'd dataset name as
        ``NotFound`` (the most common operator-facing failure mode).
        A SA WITHOUT project-level visibility on the target project
        gets ``Forbidden`` regardless of whether the dataset exists —
        we deliberately don't wrap Forbidden so an SRE seeing the
        real error knows to fix the IAM grant, not chase a phantom
        typo.
        """
        from google.api_core.exceptions import NotFound  # type: ignore

        if dataset in self._bq_probed_datasets:
            return
        try:
            client.get_dataset(dataset)
        except NotFound as exc:
            raise RuntimeError(
                f"AWS BQ billing: dataset {dataset!r} was not found "
                "by the BigQuery API. Most likely cause: typo in "
                "``config_public.billing_bq_dataset`` on the AWS "
                "integration. Verify the value matches "
                "``<project>.<dataset>`` exactly as the dataset "
                "appears in the BQ console, and that the GCP "
                "integration's service account is granted on it "
                "(via IAM ``roles/bigquery.dataViewer`` OR the "
                "dataset's legacy ACL ``access`` array — see "
                "``docs/ops/cloud-lifecycle-ops.md`` section "
                "``9a-aws-bq`` for the prod grant flow). "
                f"Original error: {exc}",
            ) from exc
        # Cache only on success — a transient Forbidden / 5xx must
        # re-probe on the next call so a permission grant or quota
        # recovery doesn't take a worker-restart to take effect.
        self._bq_probed_datasets.add(dataset)

    def _bq_discover_account_tables(
        self, client: Any, *, dataset: str, pattern: str,
    ) -> list[tuple[str, str]]:
        """Discover ``(account_id, fq_table)`` pairs for tenant-wide
        queries (``scope=None``).

        The driver doesn't trust the platform's registered scopes
        for the discovery step — the reconciliation pass in
        ``billing_daily`` exists precisely to surface unregistered
        AWS accounts whose spend would otherwise vanish. We list
        the dataset's tables and pull the account-id token from each
        table name.
        """
        from google.cloud import bigquery  # type: ignore

        like_pattern = pattern.replace("{account_id}", "%")
        q = (
            f"SELECT table_name FROM `{dataset}.INFORMATION_SCHEMA.TABLES` "
            "WHERE table_name LIKE @pat"
        )
        job = client.query(q, job_config=self._bq_query_job_config([
            bigquery.ScalarQueryParameter("pat", "STRING", like_pattern),
        ]))
        # Extract account id by removing the pattern's literal
        # suffix/prefix around the placeholder. Pattern is
        # ``{account_id}_aws_billing_data`` → prefix="" suffix="_aws_billing_data".
        before, _, after = pattern.partition("{account_id}")
        out: list[tuple[str, str]] = []
        for row in job.result():
            name = str(row.table_name)
            if not name.startswith(before) or not name.endswith(after):
                continue
            acct = name[len(before):len(name) - len(after) or None]
            if not (acct.isdigit() and len(acct) == 12):
                # Skip helper / view tables (e.g.
                # ``557537366020_aws_grouped_o11y_billing_cost`` doesn't
                # fit the {12-digit-id}_aws_billing_data shape) — they
                # are not real per-account CURs and would double-count.
                continue
            out.append((acct, f"{dataset}.{name}"))
        return out

    # ── Daily billing (Cost Explorer or BigQuery CUR) ─

    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]:
        if self._billing_source == "bigquery":
            return await asyncio.to_thread(
                self._list_billing_daily_bq_sync,
                start_date, end_date, scope,
                include_owner_breakdown, bq_billing,
            )
        # boto3 is a hard runtime dependency declared in pyproject. If the
        # import flag is False we are in a broken deployment, not a
        # graceful-degradation case — raise so the scheduled job logs the
        # configuration error against the integration.
        if not _BOTO3_AVAILABLE:
            raise RuntimeError(
                "boto3 not importable — AWS billing daily sync cannot run "
                "on this worker. Reinstall the AWS driver dependencies.",
            )
        return await asyncio.to_thread(
            self._list_billing_daily_sync,
            start_date, end_date, scope, include_owner_breakdown,
        )

    def _list_billing_daily_sync(
        self, start: date, end: date, scope: CloudScopeRef | None,
        include_owner_breakdown: bool = False,
    ) -> list[CloudDailyCost]:
        # Cost Explorer is centralized in us-east-1 for all orgs.
        ce = self._boto_client("ce", region="us-east-1")
        # CE end date is EXCLUSIVE — add a day so the caller's semantics
        # (end_date inclusive) holds.
        time_period = {
            "Start": start.isoformat(),
            "End": (end + timedelta(days=1)).isoformat(),
        }
        out: list[CloudDailyCost] = []
        try:
            # First pass: one row per (date, linked_account)
            for page in _ce_paged(
                ce,
                TimePeriod=time_period,
                Granularity="DAILY",
                Metrics=["UnblendedCost"],
                GroupBy=[{"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"}],
            ):
                for bucket in page.get("ResultsByTime") or []:
                    d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                    for grp in bucket.get("Groups") or []:
                        acct_id = (grp.get("Keys") or [""])[0]
                        if scope is not None and acct_id != scope.external_id:
                            continue
                        amount = float(
                            grp["Metrics"]["UnblendedCost"]["Amount"],
                        )
                        currency = grp["Metrics"]["UnblendedCost"].get(
                            "Unit", "USD",
                        )
                        # Same non-USD handling as ``list_cost_facts``
                        # (Round-2 review parity fix). The legacy field
                        # name ``total_cost_usd`` was always meant to be
                        # USD — pre-fix on AWS-China (Sinnet/NWCD,
                        # ``Unit=CNY``) the driver dropped raw CNY into
                        # this field and silently inflated USD daily
                        # totals. AWS is a USD-native driver (no
                        # ``cny_per_usd`` consumption), so non-USD rows
                        # now zero ``total_cost_usd`` and stash the
                        # native amount in ``raw.native_amount`` for
                        # native-currency analytics.
                        is_usd = str(currency).upper() == "USD"
                        out.append(CloudDailyCost(
                            date=d,
                            scope=CloudScopeRef(
                                scope_type="aws_account",
                                external_id=acct_id,
                            ),
                            total_cost_usd=amount if is_usd else 0.0,
                            currency=currency,
                            raw={
                                "ce_group": grp,
                                "native_amount": amount,
                            },
                        ))
        except (ClientError, BotoCoreError) as exc:
            # Surface the real Cost Explorer error to the lifecycle
            # service so it appears in the scheduled_job_runs ``errors``
            # list (visible in Admin → Cloud → Sync history). Silent
            # ``return []`` would let spike-detection run on stale 7-day
            # averages while the operator stays in the dark.
            raise RuntimeError(
                f"AWS Cost Explorer daily fetch failed: {exc}",
            ) from exc

        # 双口径白名单遍(RECORD_TYPE=Usage):CE 无行级明细,list 由
        # 记账类型过滤的第二遍查询得出(SG 无 SP,Usage 即完整白名单;
        # 混在发票口径里的 Credit / Tax / Refund 被过滤掉)。public 恒
        # None —— CE 无公开牌价度量。失败不阻断发票口径(降级:list
        # 落 None + WARN),发票口径失败仍如上硬抛。
        wl_by_key: dict[tuple[date, str], float] = {}
        wl_ok = True
        try:
            for page in _ce_paged(
                ce,
                TimePeriod=time_period,
                Granularity="DAILY",
                Metrics=["UnblendedCost"],
                GroupBy=[{"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"}],
                Filter={"Dimensions": {
                    "Key": "RECORD_TYPE", "Values": ["Usage"],
                }},
            ):
                for bucket in page.get("ResultsByTime") or []:
                    d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                    for grp in bucket.get("Groups") or []:
                        acct_id = (grp.get("Keys") or [""])[0]
                        if scope is not None and acct_id != scope.external_id:
                            continue
                        amt = float(
                            grp["Metrics"]["UnblendedCost"]["Amount"],
                        )
                        wl_by_key[(d, acct_id)] = (
                            wl_by_key.get((d, acct_id), 0.0) + amt
                        )
        except (ClientError, BotoCoreError) as exc:
            wl_ok = False
            logger.warning(
                "AWS Cost Explorer whitelist (RECORD_TYPE=Usage) daily "
                "fetch failed — list-cost measure degrades to NULL for "
                "this window: %s", exc,
            )
        if wl_ok:
            for row in out:
                if row.currency and row.currency.upper() != "USD":
                    continue
                acct = row.scope.external_id if row.scope else ""
                wl_amt = wl_by_key.get((row.date, acct))
                # 无白名单单元 = 该日该账号纯为 credit/税 → list 真 0。
                row.total_cost_usd_list = float(wl_amt or 0.0)

        # Second pass (optional): decompose per-scope daily total into
        # per-service breakdown. Skip for the tenant-wide call (too
        # many rows); include when called per-scope.
        if scope is not None and out:
            by_day: dict[date, dict[str, float]] = {}
            try:
                for page in _ce_paged(
                    ce,
                    TimePeriod=time_period,
                    Granularity="DAILY",
                    Metrics=["UnblendedCost"],
                    GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
                    Filter={
                        "Dimensions": {
                            "Key": "LINKED_ACCOUNT",
                            "Values": [scope.external_id],
                        }
                    },
                ):
                    for bucket in page.get("ResultsByTime") or []:
                        d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                        day_breakdown = by_day.setdefault(d, {})
                        for grp in bucket.get("Groups") or []:
                            svc = (grp.get("Keys") or [""])[0]
                            amt = float(
                                grp["Metrics"]["UnblendedCost"]["Amount"],
                            )
                            day_breakdown[svc] = day_breakdown.get(svc, 0) + amt
            except (ClientError, BotoCoreError) as exc:
                logger.info(
                    "Cost Explorer per-service breakdown failed "
                    "(keeping daily totals): %s", exc,
                )
            else:
                for row in out:
                    row.cost_by_service = by_day.get(row.date) or {}

            # Third pass (optional): iamPrincipal/Owner tag breakdown.
            # Only runs when requested AND scope is set (per-account pin
            # is required — a tenant-wide owner query is meaningless without
            # knowing which account's principals we are attributing).
            #
            # CE returns tag values as "iamPrincipal/Owner$<value>"; we strip
            # the namespace prefix exactly as _list_cost_facts_principal_tag_sync
            # does.  Empty-string tag values (resources with no Owner tag set)
            # are skipped — they are already captured in total_cost_usd and
            # would bloat the breakdown with an uninformative "(untagged)" key.
            if include_owner_breakdown:
                by_day_owner: dict[date, dict[str, float]] = {}
                _owner_tag_key = "iamPrincipal/Owner"
                try:
                    for page in _ce_paged(
                        ce,
                        TimePeriod=time_period,
                        Granularity="DAILY",
                        Metrics=["UnblendedCost"],
                        GroupBy=[{"Type": "TAG", "Key": _owner_tag_key}],
                        Filter={
                            "Dimensions": {
                                "Key": "LINKED_ACCOUNT",
                                "Values": [scope.external_id],
                            }
                        },
                    ):
                        for bucket in page.get("ResultsByTime") or []:
                            d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                            day_map = by_day_owner.setdefault(d, {})
                            for grp in bucket.get("Groups") or []:
                                raw = (grp.get("Keys") or [""])[0]
                                # Strip "iamPrincipal/Owner$" prefix.
                                owner = (
                                    raw.split("$", 1)[1]
                                    if "$" in raw
                                    else raw
                                )
                                if not owner:
                                    continue
                                amt = float(
                                    grp["Metrics"]["UnblendedCost"]["Amount"],
                                )
                                if amt:
                                    day_map[owner] = day_map.get(owner, 0) + amt
                except (ClientError, BotoCoreError) as exc:
                    logger.info(
                        "Cost Explorer iamPrincipal/Owner breakdown failed "
                        "for scope %s (skipping, totals unaffected): %s",
                        scope.external_id, exc,
                    )
                else:
                    for row in out:
                        row.owner_breakdown = by_day_owner.get(row.date) or {}
        return out

    # ── Daily billing (BigQuery CUR) ─────────────────────────────
    #
    # Mirrors ``_list_billing_daily_sync``'s contract: one
    # ``CloudDailyCost`` per (date, linked-account) when ``scope`` is
    # set; one per (date, linked-account) across ALL accounts visible
    # in the dataset when ``scope`` is None (reconciliation pass).
    # ``include_owner_breakdown`` is honoured via the CUR's
    # ``resource_tags_user_owner`` column — semantically the resource-
    # tagged Owner, not the IAM-principal Owner that CE returns, but
    # for prod's needs that's what's actually wanted (and the only
    # cost-allocation surface the CUR exposes; the CUR has no
    # iamPrincipal column).

    def _list_billing_daily_bq_sync(
        self, start: date, end: date,
        scope: CloudScopeRef | None,
        include_owner_breakdown: bool,
        bq_billing: dict[str, Any] | None,
        *, _all_accounts: bool = False,
    ) -> list[CloudDailyCost]:
        from google.cloud import bigquery  # type: ignore

        (
            creds, dataset, pattern, layout, tag_schema, fact_grain,
        ) = self._bq_resolve_options(bq_billing)

        # ── Batched all-accounts dispatch (per_month / CUR 2.0) ──
        # Same scan-once-for-all-accounts optimisation as cost_facts (see
        # ``_cost_facts_all_account_scopes`` in __init__): the per-account
        # ``WHERE line_item_usage_account_id`` filter doesn't prune the
        # DAY-partitioned, non-account-clustered CUR month-table, so one
        # query per scope re-scans the same partitions N×. billing_daily
        # runs up to 3 such queries (totals / service / owner) per scope.
        # When the caller opts in via ``_billing_daily_all_account_scopes``
        # and the window is bounded, scan ONCE for all accounts and serve
        # each scope from a per-run cache. ``_all_accounts`` guards re-entry
        # (the cache builder calls this method with scope=None).
        if not _all_accounts:
            all_scopes = self._billing_daily_all_account_scopes
            if (
                layout == "per_month"
                and scope is not None and scope.external_id
                and all_scopes
                and (end - start).days
                <= self._BQ_COST_FACTS_BATCH_MAX_SPAN_DAYS
            ):
                cache = self._bq_billing_daily_batch_results(
                    start, end, include_owner_breakdown,
                    all_scopes, bq_billing,
                )
                return list(cache.get(str(scope.external_id), ()))

        client = self._bq_client(creds)
        try:
            partition_pred, partition_params = self._bq_partition_predicate(
                layout, start, end,
            )

            picks = self._bq_resolve_tables(
                client, layout=layout, dataset=dataset, pattern=pattern,
                # Batched mode resolves tenant-wide (scope=None → per_month
                # picks carry account_filter=None → no @scope_acct clause)
                # so one scan serves every account; each output
                # CloudDailyCost self-identifies its account via acct_id.
                scope=(None if _all_accounts else scope),
                start=start, end=end,
            )
            if not picks:
                logger.warning(
                    "AWS billing_daily (BQ): no CUR tables matched in "
                    "dataset %s with pattern %s for layout %s",
                    dataset, pattern, layout,
                )
                return []

            # All picks share the same account_filter (derived from
            # ``scope`` at resolve time), so we lift it out of the
            # per-leg WHERE construction and bind a single @scope_acct
            # parameter when any pick wants it.
            account_filter = picks[0].account_filter
            account_filter_clause = (
                " AND line_item_usage_account_id = @scope_acct"
                if account_filter is not None else ""
            )

            # CUR 2.0 nests product fields inside ``product.key_value``;
            # the helper returns a column name for CUR 1.0 (flat) or a
            # correlated subquery for CUR 2.0 (nested). Resolved once
            # because it's stable across every UNION leg.
            product_name_expr = self._bq_product_field_expression(
                tag_schema, "product_name",
            )

            # Per-table SELECT — pre-aggregates to (day, account, currency)
            # plus optional service breakdown via ANY_VALUE? No — service
            # breakdown is a per-(day, account, service, currency) grain
            # so we collect it in a second pass after the daily totals
            # query. ``currency`` MUST appear in every GROUP BY here so
            # a hypothetical mixed-currency day (USD + CNY on AWS-China
            # cross-partition tenants) produces atomic rows rather than
            # blending the per-service totals across currencies — the
            # output loop then matches breakdowns to totals on
            # ``(day, account, currency)`` so each output row carries
            # the breakdown for its own currency.
            def _totals_select(pick: _BqTablePick) -> str:
                # SP-negation / ComputeSavingsPlans rows are intentionally
                # NOT filtered — see the SP semantics block at the top of
                # the class. Filtering them out drops the discount and
                # inflates reported cost ~30%.
                #
                # 双口径:同一次扫描顺带产出 native_list(业务折前成本,
                # 白名单行)与 native_public(按需牌价)——见类顶
                # ``_LIST_WHITELIST_TYPES`` 的口径说明。CUR 1.0(flat)
                # 老表可能缺 pricing_public / line-type 列 → 投影 NULL,
                # 度量落 NULL(=未接入),绝不伪造 0。
                if tag_schema == "flat":
                    cols = pick.columns or set()
                    lt_expr = (
                        "line_item_line_item_type"
                        if "line_item_line_item_type" in cols
                        else "CAST(NULL AS STRING)"
                    )
                    gross_expr = (
                        "pricing_public_on_demand_cost"
                        if "pricing_public_on_demand_cost" in cols
                        else "CAST(NULL AS FLOAT64)"
                    )
                else:
                    lt_expr = "line_item_line_item_type"
                    gross_expr = "pricing_public_on_demand_cost"
                list_expr, public_expr = self._bq_list_measure_exprs(
                    line_type_expr=lt_expr, gross_expr=gross_expr,
                )
                return f"""
                SELECT
                  DATE(line_item_usage_start_date) AS day,
                  line_item_usage_account_id AS account_id,
                  line_item_currency_code AS currency,
                  SUM(line_item_unblended_cost) AS native_cost,
                  {list_expr} AS native_list,
                  {public_expr} AS native_public
                FROM `{pick.fq_table}`
                WHERE {partition_pred}
                  AND DATE(line_item_usage_start_date) BETWEEN @start AND @end
                  {account_filter_clause}
                GROUP BY day, account_id, currency
                """

            totals_sql = " UNION ALL ".join(
                _totals_select(p) for p in picks
            )
            params: list[Any] = [
                bigquery.ScalarQueryParameter("start", "DATE", start),
                bigquery.ScalarQueryParameter("end", "DATE", end),
            ] + partition_params
            if account_filter is not None:
                params.append(bigquery.ScalarQueryParameter(
                    "scope_acct", "STRING", account_filter,
                ))

            totals_job = client.query(
                totals_sql, job_config=self._bq_query_job_config(params),
            )
            # Per-key accumulator: invoice native + the two whitelist
            # measures. ``*_seen`` keeps NULL-vs-0 honest — a key whose
            # every UNION leg returned NULL list (legacy CUR without the
            # needed columns) emits None, not a fabricated $0.
            totals: dict[tuple[date, str, str], dict[str, Any]] = {}
            for row in totals_job.result():
                key = (row.day, str(row.account_id), str(row.currency or "USD"))
                # UNION ALL across month-tables in per_month means the
                # same (day, account, currency) tuple may appear twice
                # at month boundaries — sum rather than overwrite so
                # the cross-month aggregate is correct. (per_account
                # tables don't span months so the +=0 branch is a noop
                # but keeps the contract uniform.)
                acc = totals.setdefault(key, {
                    "native": 0.0,
                    "list": 0.0, "list_seen": False,
                    "public": 0.0, "public_seen": False,
                })
                acc["native"] += float(row.native_cost or 0)
                _list = getattr(row, "native_list", None)
                if _list is not None:
                    acc["list"] += float(_list)
                    acc["list_seen"] = True
                _public = getattr(row, "native_public", None)
                if _public is not None:
                    acc["public"] += float(_public)
                    acc["public_seen"] = True

            # Service breakdown — pre-aggregate per table, then collate.
            # Keyed by (day, account, currency) to match the totals
            # output rows; without ``currency`` in GROUP BY a single
            # mixed-currency day would attach the same blended
            # breakdown to both the USD and CNY output rows.
            def _service_select(pick: _BqTablePick) -> str:
                # SP rows are kept (see SP semantics block at top of class).
                # Bucket every SP-related line item under a single
                # "Savings Plans" pseudo-service so the per-scope donut /
                # ranking has a clearly labelled discount slice (negative
                # on linked accounts, positive on the payer's SP fee row)
                # rather than emitting a CUR-internal product name like
                # ``Savings Plans for AWS Compute usage`` that mixes both
                # signs in the same legend entry.
                #
                # ``EdpDiscount`` / ``PrivateRateDiscount`` get the same
                # treatment — CUR 2.0 / Data Exports v2 surface enterprise
                # discount lines as negative rows with NULL product_code,
                # so without a named bucket they collapse into the
                # ``Other`` / unnamed service slice (legend reads "Other
                # = -$566.39", uninterpretable for the owner). The donut +
                # discountSummary surface picks them up by name as
                # ``EDP Discount`` / ``Private Rate Discount`` after this
                # CASE. CUR 1.0 export had no such rows so the WHEN is
                # a no-op for legacy datasets.
                return f"""
                SELECT
                  DATE(line_item_usage_start_date) AS day,
                  line_item_usage_account_id AS account_id,
                  line_item_currency_code AS currency,
                  CASE
                    WHEN line_item_line_item_type = 'EdpDiscount'
                      THEN 'EDP Discount'
                    WHEN line_item_line_item_type = 'PrivateRateDiscount'
                      THEN 'Private Rate Discount'
                    WHEN line_item_product_code = 'ComputeSavingsPlans'
                      OR line_item_line_item_type IN (
                        'SavingsPlanNegation', 'SavingsPlanCoveredUsage',
                        'SavingsPlanRecurringFee', 'SavingsPlanUpfrontFee'
                      )
                    THEN 'Savings Plans'
                    ELSE IFNULL({product_name_expr}, line_item_product_code)
                  END AS service,
                  SUM(line_item_unblended_cost) AS native_cost
                FROM `{pick.fq_table}`
                WHERE {partition_pred}
                  AND DATE(line_item_usage_start_date) BETWEEN @start AND @end
                  {account_filter_clause}
                GROUP BY day, account_id, currency, service
                """

            services_sql = " UNION ALL ".join(
                _service_select(p) for p in picks
            )
            services_job = client.query(
                services_sql, job_config=self._bq_query_job_config(params),
            )
            service_breakdown: dict[tuple[date, str, str], dict[str, float]] = {}
            for row in services_job.result():
                key = (row.day, str(row.account_id), str(row.currency or "USD"))
                bd = service_breakdown.setdefault(key, {})
                amt = float(row.native_cost or 0)
                if amt:
                    bd[str(row.service or "Other")] = (
                        bd.get(str(row.service or "Other"), 0.0) + amt
                    )

            # Owner breakdown (optional + per-scope only — symmetric with
            # the CE codepath's third pass). Source column / expression
            # is layout-aware (``resource_tags_user_owner`` for CUR 1.0
            # flat, ``UNNEST(resource_tags.key_value) WHERE
            # key='user_owner'`` for CUR 2.0 nested); rows where the tag
            # is empty are dropped from the breakdown so the dashboard
            # doesn't display a giant ``(untagged)`` bucket — same skip
            # as CE. Also keyed by currency for the same reason service
            # breakdown is.
            #
            # For ``flat`` schema we drop UNION legs over tables that
            # DON'T have the owner column (PingCAP's per-account CUR
            # Parquet tables vary — an account whose resources were
            # never tagged with ``Owner`` simply has no such column in
            # its CUR schema, and including it in SELECT would raise
            # ``invalidQuery: Unrecognized name`` — see PR #311
            # follow-up). CUR 2.0 (``nested``) has uniform schema, so
            # every pick contributes.
            owner_breakdown_by_day: dict[
                tuple[date, str, str], dict[str, float]
            ] = {}
            if include_owner_breakdown and (_all_accounts or scope is not None) and picks:
                owner_expr = self._bq_tag_expression(tag_schema, "Owner")
                if tag_schema == "flat":
                    owner_picks = [
                        p for p in picks
                        if "resource_tags_user_owner"
                        in (p.columns or set())
                    ]
                else:
                    # CUR 2.0 — nested ``resource_tags.key_value`` lives
                    # on every per-month table, no per-pick drop needed.
                    owner_picks = list(picks)
                if owner_picks:
                    # SP rows have no resource owner tag — they're
                    # already implicitly dropped by the
                    # ``owner IS NOT NULL`` predicate. No explicit SP
                    # filter needed (and we MUST NOT add one — see SP
                    # semantics block at top of class).
                    owner_sql = " UNION ALL ".join(
                        f"""
                        SELECT
                          DATE(line_item_usage_start_date) AS day,
                          line_item_usage_account_id AS account_id,
                          line_item_currency_code AS currency,
                          {owner_expr} AS owner,
                          SUM(line_item_unblended_cost) AS native_cost
                        FROM `{p.fq_table}`
                        WHERE {partition_pred}
                          AND DATE(line_item_usage_start_date) BETWEEN @start AND @end
                          {account_filter_clause}
                          AND {owner_expr} IS NOT NULL
                          AND {owner_expr} != ''
                        GROUP BY day, account_id, currency, owner
                        """
                        for p in owner_picks
                    )
                    owner_job = client.query(
                        owner_sql,
                        job_config=self._bq_query_job_config(params),
                    )
                    for row in owner_job.result():
                        key = (
                            row.day, str(row.account_id),
                            str(row.currency or "USD"),
                        )
                        bd = owner_breakdown_by_day.setdefault(key, {})
                        amt = float(row.native_cost or 0)
                        if amt:
                            owner = str(row.owner)
                            bd[owner] = bd.get(owner, 0.0) + amt

            # Detect multi-currency days BEFORE emitting. The
            # ``cmdb_cloud_billing_daily`` table's UNIQUE constraint
            # is ``(integration_id, scope_id, date)`` — it cannot
            # carry two rows for the same day with different
            # currencies. The PingCAP CUR is USD-only end-to-end
            # (all 60 accounts return ``USD``) so this is purely a
            # future-proofing guard: if AWS-China (Sinnet/NWCD,
            # CNY-billed) is ever onboarded to BQ mode, the operator
            # gets a loud failure rather than a silent UPSERT
            # collision that drops one of the two currency rows. The
            # fix in that scenario is a migration extending the
            # UNIQUE constraint with ``currency`` — explicitly out of
            # scope for this PR.
            currencies_per_day_account: dict[
                tuple[date, str], set[str],
            ] = {}
            for (day, acct_id, currency) in totals.keys():
                currencies_per_day_account.setdefault(
                    (day, acct_id), set(),
                ).add(currency)
            mixed = {
                k: cs for k, cs in currencies_per_day_account.items()
                if len(cs) > 1
            }
            if mixed:
                example = next(iter(mixed.items()))
                raise RuntimeError(
                    "AWS billing_daily (BQ): multi-currency days "
                    "are not supported by the cmdb_cloud_billing_daily "
                    "UNIQUE constraint (integration, scope, date). "
                    f"Example: day={example[0][0]} account={example[0][1]} "
                    f"currencies={sorted(example[1])}. Extend the "
                    "UNIQUE constraint with ``currency`` before "
                    "onboarding multi-currency AWS partitions.",
                )

            out: list[CloudDailyCost] = []
            for (day, acct_id, currency), acc in totals.items():
                amount = float(acc["native"])
                is_usd = currency.upper() == "USD"
                key3 = (day, acct_id, currency)
                # 双口径:list / public 与 total_cost_usd 同一套 USD 规则
                # (非 USD 行置 None——AWS 是 USD-native 驱动,不换算;
                # native 值进 raw 供审计)。NULL-seen 的键透传 None。
                native_list = float(acc["list"]) if acc["list_seen"] else None
                native_public = (
                    float(acc["public"]) if acc["public_seen"] else None
                )
                row = CloudDailyCost(
                    date=day,
                    scope=CloudScopeRef(
                        scope_type="aws_account",
                        external_id=acct_id,
                    ),
                    # Same non-USD handling as the CE path: ``total_cost_usd``
                    # zero when the row is in a non-USD currency (AWS-China
                    # CUR partitions emit CNY) so the dashboard's USD sum
                    # doesn't silently inflate. The native amount lands in
                    # ``raw.native_amount`` for native-currency analytics.
                    total_cost_usd=amount if is_usd else 0.0,
                    total_cost_usd_list=(
                        native_list if (is_usd and native_list is not None)
                        else None
                    ),
                    total_cost_usd_public=(
                        native_public if (is_usd and native_public is not None)
                        else None
                    ),
                    currency=currency,
                    cost_by_service=service_breakdown.get(key3) or {},
                    # Empty dict (not None) when include_owner_breakdown
                    # was requested but no resource_tags_user_owner rows
                    # matched — matches the CE codepath's semantics so the
                    # daily-billing upsert lands ``{}`` consistently
                    # across providers.
                    owner_breakdown=(
                        owner_breakdown_by_day.get(key3) or {}
                        if include_owner_breakdown else None
                    ),
                    raw={
                        "bq_source": "cur",
                        "native_amount": amount,
                        "native_list": native_list,
                        "native_public": native_public,
                    },
                )
                out.append(row)
            return out
        finally:
            # ``bigquery.Client`` holds pooled HTTP connections + auth
            # state; the cron creates a fresh client per call so
            # leaking them accumulates ~180 unclosed clients per
            # ``cost_facts_sync`` tick (3 BQ ops × 60 scopes). On a
            # long-running worker process that breaches the soft FD
            # limit within hours. Close defensively — any close-side
            # error is non-fatal (best-effort).
            try:
                client.close()
            except Exception:  # noqa: BLE001
                pass

    def _bq_billing_daily_batch_results(
        self, start: date, end: date, include_owner_breakdown: bool,
        all_account_scopes: list[CloudScopeRef],
        bq_billing: dict[str, Any] | None,
    ) -> dict[str, list[CloudDailyCost]]:
        """Scan the per_month CUR window ONCE for every target account
        and return daily-cost rows bucketed by account id.

        Memoised on the instance keyed by (start, end,
        include_owner_breakdown) — constant across one billing-daily cron
        run, so the first scope's call does the scan(s) and the rest hit
        the cache. Rows are byte-identical to the per-account path: the
        same totals / service / owner queries and aggregation run, only
        without the ``line_item_usage_account_id`` filter, and each
        ``CloudDailyCost`` already self-synthesises its scope from the
        row's ``account_id`` — so the per-scope slice equals what the
        filtered query would have returned for that account. Only
        registered target accounts are bucketed; rows for any other
        account sharing the month-table are dropped.
        """
        cache = self._bq_billing_daily_batch_cache
        key = (start, end, include_owner_breakdown)
        cached = cache.get(key)
        if cached is not None:
            return cached

        valid: set[str] = set()
        for s in all_account_scopes:
            if s is None or not s.external_id:
                continue
            ext = str(s.external_id)
            if ext.isdigit() and len(ext) == 12:
                valid.add(ext)

        rows = self._list_billing_daily_bq_sync(
            start, end, None, include_owner_breakdown, bq_billing,
            _all_accounts=True,
        )
        by_account: dict[str, list[CloudDailyCost]] = {}
        for row in rows:
            ext = str(row.scope.external_id)
            if ext in valid:
                by_account.setdefault(ext, []).append(row)
        cache[key] = by_account
        return by_account

    # ── Line-item cost facts (Cost Explorer w/ resources) ───────────
    #
    # Cost Explorer's ``GetCostAndUsageWithResources`` is the ONLY
    # path to per-resource (EC2 instance / EBS volume / Lambda
    # function …) cost data. Hard limits AWS imposes:
    #   * window cannot start more than 14 days before now;
    #   * GROUP BY caps at 2 entries — we use ``[SERVICE, RESOURCE_ID]``
    #     for the primary call;
    #   * a Filter is REQUIRED (we anchor on LINKED_ACCOUNT to scope
    #     to one member account).
    # CUR-via-Athena is the >14-day path; out of scope for this PR
    # (would require S3 export + Athena workgroup config; covered in
    # DECISIONS.md follow-up).
    #
    # When the requested window starts >14d ago, we still issue the
    # call clipped to the last 14 days and log the gap; the caller
    # (cost_facts cron) is expected to bound its window. Returning []
    # for the entire window would silently drop today's data when an
    # operator does an opportunistic 30-day backfill.

    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]:
        # AWS Cost Explorer's ``UnblendedCost.Amount`` is returned in
        # the linked-account's billing currency; for the USD-billed
        # accounts this driver mostly targets, that's USD. AWS China
        # (cn-north-1 / cn-northwest-1, operated by Sinnet / NWCD)
        # is a separate partition that bills in CNY and CE there returns
        # Unit=CNY. Pre-launch hardening 2026-05-16 (LS1 P1): consume
        # ``cny_per_usd`` so CNY rows convert to USD instead of dropping
        # to $0 (which silently undercounted China-partition tenants in
        # every USD aggregate). The kwarg threads down to both sync paths
        # via the ``_cny_per_usd`` instance attr (instance-scoped so a
        # subsequent call with None doesn't clobber a healthy rate).
        if cny_per_usd is not None and cny_per_usd > 0:
            self._cny_per_usd_for_cost_facts = float(cny_per_usd)
        else:
            self._cny_per_usd_for_cost_facts = None
        # ── BigQuery CUR path ──
        # ``billing_source=bigquery`` short-circuits the entire CE
        # ladder below. The principal-tag mode stays CE-only — that
        # flag is only set on the SG integration whose creds keep CE
        # access and whose CUR is unavailable.
        if self._billing_source == "bigquery":
            if principal_tag_keys:
                # Defensive: SG should never set billing_source=bigquery,
                # and the main AWS integration should never have
                # platform_settings.cost_facts.aws_principal_tag.<slug>
                # set. If both somehow appear, refuse loudly rather
                # than silently doing one or the other.
                raise RuntimeError(
                    "AWS list_cost_facts: principal_tag_keys is "
                    "incompatible with billing_source=bigquery — "
                    "the CUR has no iamPrincipal columns. Unset one.",
                )
            return await asyncio.to_thread(
                self._list_cost_facts_bq_sync,
                start_date, end_date, scope, bq_billing,
            )
        if not _BOTO3_AVAILABLE:
            raise RuntimeError(
                "boto3 not importable — AWS cost-facts ingestion cannot run.",
            )
        # ``principal_tag_keys`` switches the driver into IAM-principal-
        # tag granularity (one fact per (date × service × tag-value)
        # cell) instead of the default per-resource shape. Used for
        # AWS SG / Bedrock-only accounts where the primary cost driver
        # has no taggable resource; an admin enables this by
        # activating the IAM principal tag(s) as cost-allocation tags
        # in the AWS Billing Console **and** seeding
        # ``platform_settings.cost_facts.aws_principal_tag.<slug>`` —
        # the cost-ingestion service resolves the setting and passes
        # the value here. Empty / None falls through to the default
        # per-resource path.
        if principal_tag_keys:
            return await asyncio.to_thread(
                self._list_cost_facts_principal_tag_sync,
                start_date, end_date, scope, principal_tag_keys,
            )
        return await asyncio.to_thread(
            self._list_cost_facts_sync, start_date, end_date, scope,
        )

    def _list_cost_facts_sync(
        self, start: date, end: date, scope: CloudScopeRef | None,
    ) -> list[CostFact]:
        from datetime import UTC

        ce = self._boto_client("ce", region="us-east-1")

        # Clip to AWS's 14-day window. Today minus 14 days INCLUSIVE
        # is the earliest allowed start; if the caller wants older,
        # we clip and log so the gap is visible in the cron's
        # scheduled_job_runs row.
        today_utc = datetime.now(UTC).date()
        earliest = today_utc - timedelta(days=14)
        clipped_start = max(start, earliest)
        if clipped_start > end:
            logger.info(
                "AWS cost-facts: requested window %s..%s entirely outside "
                "Cost Explorer's 14-day resource window — returning empty",
                start, end,
            )
            return []
        if clipped_start != start:
            logger.info(
                "AWS cost-facts: clipping start %s → %s "
                "(Cost Explorer 14-day resource limit)",
                start, clipped_start,
            )

        time_period = {
            "Start": clipped_start.isoformat(),
            "End": (end + timedelta(days=1)).isoformat(),
        }

        # Cost Explorer's resource-level call REQUIRES a
        # LINKED_ACCOUNT pin. We refuse the tenant-wide call entirely
        # rather than silently substituting the caller's own account
        # (which on Org-master credentials returns the master's spend,
        # NOT a fanout — easily misleading for multi-account tenants).
        # The cron handler MUST iterate scopes and call us per scope.
        if scope is None or not scope.external_id:
            raise NotImplementedError(
                "AWS list_cost_facts requires an explicit scope "
                "(linked-account); the cron handler fans out per scope.",
            )
        ext = str(scope.external_id)
        if not (ext.isdigit() and len(ext) == 12):
            raise RuntimeError(
                f"AWS cost-facts: scope external_id {ext!r} is not a "
                "12-digit AWS account id; cannot pin LINKED_ACCOUNT.",
            )
        scope_filter: dict[str, Any] = {"Dimensions": {
            "Key": "LINKED_ACCOUNT", "Values": [ext],
        }}

        out: list[CostFact] = []
        integration_id = self.integration_id
        try:
            next_token: str | None = None
            while True:
                kwargs: dict[str, Any] = dict(
                    TimePeriod=time_period,
                    Granularity="DAILY",
                    Metrics=["UnblendedCost", "UsageQuantity"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "SERVICE"},
                        {"Type": "DIMENSION", "Key": "RESOURCE_ID"},
                    ],
                    Filter=scope_filter,
                )
                if next_token:
                    kwargs["NextPageToken"] = next_token
                resp = ce.get_cost_and_usage_with_resources(**kwargs)
                for bucket in resp.get("ResultsByTime") or []:
                    d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                    for grp in bucket.get("Groups") or []:
                        keys = grp.get("Keys") or []
                        service = keys[0] if len(keys) > 0 else "Other"
                        resource_id = keys[1] if len(keys) > 1 else None
                        if not resource_id:
                            continue
                        metrics = grp.get("Metrics") or {}
                        amount = float(
                            (metrics.get("UnblendedCost") or {}).get("Amount") or 0,
                        )
                        if amount == 0:
                            continue
                        currency = (
                            metrics.get("UnblendedCost") or {}
                        ).get("Unit", "USD")
                        # AWS Cost Explorer in standard partitions always
                        # returns Unit=USD. AWS China (cn-north-1 /
                        # cn-northwest-1, operated by Sinnet / NWCD) is a
                        # separate partition that bills in CNY and CE there
                        # returns Unit=CNY. Pre-launch hardening 2026-05-16
                        # (LS1 P1): when currency is CNY and the platform
                        # FX rate is available, convert to USD instead of
                        # the old amount_usd=0 fallback which silently
                        # undercounted China-partition spend in every USD
                        # aggregate. For non-CNY non-USD (e.g. Brazil BRL)
                        # we still fall through to amount_usd=0 + log so
                        # the operator sees a real partial-failure signal.
                        amount_usd_value = _convert_to_usd(
                            amount=amount,
                            currency=currency,
                            cny_per_usd=getattr(
                                self, "_cny_per_usd_for_cost_facts", None,
                            ),
                            integration_id=str(integration_id),
                        )
                        usage = (metrics.get("UsageQuantity") or {})
                        # 双口径(CE per-resource 路径):资源级行天然
                        # 白名单纯净 —— Tax / Credit / Refund / SP 抵扣
                        # / SP 月费行没有 RESOURCE_ID,已被上面的
                        # ``if not resource_id: continue`` 挡掉;SP 覆盖
                        # 行(SavingsPlanCoveredUsage)带资源 id 且 CE 的
                        # unblended 即按需等价(与 CUR 实测一致)。故
                        # list == unblended。public 恒 None:CE 无公开
                        # 牌价度量。已知局限:RI 覆盖行(DiscountedUsage,
                        # 当前无 RI)unblended≈0,CE 无法还原牌价 ——
                        # 若未来买 RI,CE 链路的 list 会低估,须切 CUR。
                        out.append(CostFact(
                            date=d,
                            integration_id=integration_id,
                            service=service or "Other",
                            amount_usd=amount_usd_value,
                            amount_native=amount,
                            amount_usd_list=amount_usd_value,
                            amount_native_list=amount,
                            cost_bucket=COST_BUCKET_USAGE,
                            currency=currency,
                            scope=scope,
                            resource_group=str(scope.external_id),
                            resource_id=resource_id,
                            resource_name=resource_id.rsplit("/", 1)[-1],
                            resource_type=_aws_resource_type_from_id(
                                resource_id,
                            ),
                            region=None,  # CE doesn't return region per group
                            usage_type=None,
                            tags={},
                            usage_quantity=(
                                float(usage.get("Amount") or 0)
                                if usage else None
                            ),
                            usage_unit=usage.get("Unit") if usage else None,
                            raw={"ce_group": grp},
                        ))
                next_token = resp.get("NextPageToken")
                if not next_token:
                    break
        except (ClientError, BotoCoreError) as exc:
            raise RuntimeError(
                f"AWS Cost Explorer GetCostAndUsageWithResources failed: {exc}",
            ) from exc
        return out

    # ── Cost facts (BigQuery CUR) ────────────────────────────────
    #
    # Per-resource cost from the CUR's ``line_item_resource_id``
    # column when that column exists in the per-account CUR table.
    # PingCAP's ~60 linked-account CUR Parquet tables have
    # HETEROGENEOUS schemas — the original PR #311 probe checked
    # only the master-account table and incorrectly generalised
    # "resource_id exists" to all 60 tables. The cron then failed
    # on 2026-05-15 against ≥3 linked accounts whose CUR was built
    # without "Include resource IDs" toggled. This codepath now
    # introspects the table's schema (``client.get_table`` was
    # already called for existence — no extra round trip) and
    # degrades gracefully:
    #   * ``line_item_resource_id`` present → grain is
    #     (day × account × product × resource_id × currency × usage_type)
    #   * ``line_item_resource_id`` absent → grain is
    #     (day × account × product × currency × usage_type) and
    #     resource_id is synthesised as
    #     ``aws-account:<acc>:<product_code>`` (resource_type
    #     ``aws-aggregated``) — the same synthetic key already used
    #     for NULL-resource_id rows on tables that DO have the column.
    # Tag columns (``resource_tags_user_owner``, ``_application``,
    # etc.) are similarly filtered to those actually present; a
    # CUR table whose account never tagged anything simply emits
    # an empty ``tags`` dict, no SQL crash. ``cny_per_usd`` handled
    # identically to the CE path (USD passes through, non-USD lands
    # native + amount_usd=0).
    @classmethod
    def _bq_cost_facts_windows(
        cls, layout: str, start: date, end: date,
    ) -> list[tuple[date, date]]:
        """Split ``[start, end]`` (inclusive) into per-CALENDAR-MONTH
        sub-windows for the day-partitioned ``per_month`` (CUR 2.0)
        layout; return the window whole for ``per_account`` (CUR 1.0).

        Why per-calendar-month and NOT a fixed day count: each CUR 2.0
        month lives in its own DAY-partitioned table, and a single
        query that UNIONs N month-tables scans ~N× a one-month worth of
        partitions — a multi-month admin-override backfill in one call
        would blow past ``_BQ_MAX_BYTES_BILLED`` (a 13-month re-pull ≈
        900 GB). Slicing on calendar-month boundaries bounds every
        query to ~one month (~70 GiB on the prod AWS main account).

        Why MONTH boundaries specifically (not arbitrary ≤15-day
        chunks): the per-fact ``date`` grain is ``DATE(start_date)`` but
        the table is DAY-partitioned on ``end_date``, and the partition-
        prune predicate clamps ``end_date`` to ``window_end + 2 days``.
        A cost-bearing row that STARTS early in a sub-window but ENDS
        well after it (month-boundary SP/RI recurring fees: start =
        month-1st, end = next-month-1st) belongs to that sub-window by
        its start-day yet sits in a partition the tight end-date prune
        would exclude — an interior (mid-month) chunk boundary would
        silently DROP it. A calendar-MONTH boundary cannot: each chunk
        is exactly the window a standalone monthly close would use, so
        the existing ``+2`` end-date pad (which already handles
        month-boundary spanning rows) keeps every chunk's result
        identical to the proven single-month re-pull. Sub-windows are
        disjoint and partition ``[start, end]`` by ``DATE(start_date)``,
        so concatenating per-window facts never collides on
        ``compute_fact_key`` (which includes ``date``).
        """
        if layout != "per_month":
            return [(start, end)]
        windows: list[tuple[date, date]] = []
        cur = start
        while cur <= end:
            # Last day of ``cur``'s calendar month.
            if cur.month == 12:
                next_month_first = date(cur.year + 1, 1, 1)
            else:
                next_month_first = date(cur.year, cur.month + 1, 1)
            w_end = min(next_month_first - timedelta(days=1), end)
            windows.append((cur, w_end))
            cur = w_end + timedelta(days=1)
        return windows

    def _list_cost_facts_bq_sync(
        self, start: date, end: date,
        scope: CloudScopeRef | None,
        bq_billing: dict[str, Any] | None,
    ) -> list[CostFact]:
        # Chunk multi-month windows by calendar month so no single BQ
        # query trips ``_BQ_MAX_BYTES_BILLED``. Resolving options here is
        # pure (no round trip) and gives us the ``layout`` needed to
        # decide whether chunking applies (see ``_bq_cost_facts_windows``).
        # The per-window helper owns its own client lifecycle; windows
        # partition ``[start, end]`` by start-day so the concatenated
        # facts never collide on ``compute_fact_key``. A single-month
        # window (the ``cmdb.cost_facts_monthclose`` case) yields one
        # window → one query, identical to the pre-chunking behaviour.
        _, _, _, layout, _, fact_grain = self._bq_resolve_options(bq_billing)

        # ── Batched all-accounts path (per_month / CUR 2.0 only) ──
        # See ``_cost_facts_all_account_scopes`` (__init__) for why the
        # per-account WHERE filter doesn't prune bytes, making one query
        # per account an N× scan-cost multiplier. When cost_service hands
        # us the full target set AND the window is bounded (so the per-
        # run cache can't hold every account's facts at once on a giant
        # single-shot backfill) AND the grain is the bounded-cardinality
        # ``per_product`` (not per-resource), scan the window ONCE for
        # all accounts and serve this scope from the cache.
        all_scopes = self._cost_facts_all_account_scopes
        if (
            layout == "per_month"
            and fact_grain == "per_product"
            and scope is not None and scope.external_id
            and all_scopes
            and (end - start).days <= self._BQ_COST_FACTS_BATCH_MAX_SPAN_DAYS
        ):
            cache = self._bq_cost_facts_batch_results(
                start, end, all_scopes, bq_billing,
            )
            return list(cache.get(str(scope.external_id), ()))

        # ── Per-account path ──
        # per_account layout (table name already filters by account),
        # tenant-wide call (scope None), no target set supplied, a span
        # too wide for the bounded cache, or per-resource grain. Streams
        # one scope's window at a time — low memory, N queries per_month.
        out: list[CostFact] = []
        for w_start, w_end in self._bq_cost_facts_windows(
            layout, start, end,
        ):
            out.extend(
                self._bq_cost_facts_one_window(
                    w_start, w_end, scope, bq_billing,
                )
            )
        return out

    def _bq_cost_facts_batch_results(
        self, start: date, end: date,
        all_account_scopes: list[CloudScopeRef],
        bq_billing: dict[str, Any] | None,
    ) -> dict[str, list[CostFact]]:
        """Scan the per_month CUR window ONCE for every target account
        and return facts bucketed by account id.

        Memoised on the instance keyed by (start, end) — constant across
        one cron run, so the first scope's call does the scan and the
        rest hit the cache. Only accounts present in
        ``all_account_scopes`` (valid 12-digit AWS ids) are bucketed;
        rows for any other account in the shared month-table are dropped
        (the cron never registered them as scopes). Facts are byte-
        identical to the per-account path: the same SELECT / GROUP BY /
        aggregation runs, only without the ``line_item_usage_account_id``
        equality filter, and each fact carries the matching scope so
        ``compute_fact_key`` (which keys on the scope external id) and
        the downstream per-scope DELETE+UPSERT are unchanged.
        """
        cache = self._bq_cost_facts_batch_cache
        key = (start, end)
        cached = cache.get(key)
        if cached is not None:
            return cached

        scope_by_account: dict[str, CloudScopeRef] = {}
        for s in all_account_scopes:
            if s is None or not s.external_id:
                continue
            ext = str(s.external_id)
            if ext.isdigit() and len(ext) == 12:
                scope_by_account[ext] = s

        by_account: dict[str, list[CostFact]] = {}
        # Per-calendar-month chunking still bounds each QUERY's bytes
        # (``_bq_cost_facts_windows`` + ``_BQ_MAX_BYTES_BILLED``); the
        # cache accumulates the windows for the whole bounded run window.
        for w_start, w_end in self._bq_cost_facts_windows(
            "per_month", start, end,
        ):
            for fact in self._bq_cost_facts_one_window(
                w_start, w_end, None, bq_billing,
                scope_by_account=scope_by_account,
            ):
                by_account.setdefault(
                    str(fact.scope.external_id), [],
                ).append(fact)
        cache[key] = by_account
        return by_account

    def _bq_cost_facts_one_window(
        self, start: date, end: date,
        scope: CloudScopeRef | None,
        bq_billing: dict[str, Any] | None,
        *, scope_by_account: dict[str, CloudScopeRef] | None = None,
    ) -> list[CostFact]:
        """Fetch + parse cost facts for ONE (already-bounded, ≤1
        calendar month for per_month) window.

        Single-scope mode (``scope_by_account is None``): identical to
        the historical single-query ``_list_cost_facts_bq_sync`` body —
        the query pins ``line_item_usage_account_id = @scope_acct`` and
        every emitted fact carries ``scope``.

        Batched mode (``scope_by_account`` provided, per_month only):
        drops the account filter (which doesn't prune bytes — see
        ``_bq_cost_facts_batch_results``) so ONE scan covers all
        accounts, and each emitted fact carries the scope that matches
        its row's ``account_id`` (rows whose account isn't a registered
        target scope are skipped). Facts are otherwise byte-identical to
        what the single-scope path would have produced for that account.
        """
        from google.cloud import bigquery  # type: ignore

        batch_mode = scope_by_account is not None
        if batch_mode:
            # All-accounts scan: the account dimension comes from each
            # row, so there's no single ``scope`` to validate. ``ext`` is
            # only used for the no-tables log line below.
            ext = "<all-accounts>"
        else:
            # Same per-scope contract as the CE codepath: the cron fans
            # out per scope, so tenant-wide cost_facts is never called.
            # Refuse rather than silently fanning out across 60 tables —
            # cost_facts upsert is per-(scope_id) and a tenant-wide call
            # would land facts under the wrong scope.
            if scope is None or not scope.external_id:
                raise NotImplementedError(
                    "AWS list_cost_facts (BQ mode) requires an explicit "
                    "scope (linked-account); the cron handler fans out "
                    "per scope.",
                )
            ext = str(scope.external_id)
            if not (ext.isdigit() and len(ext) == 12):
                raise RuntimeError(
                    f"AWS cost-facts (BQ): scope external_id {ext!r} is "
                    "not a 12-digit AWS account id",
                )

        (
            creds, dataset, pattern, layout, tag_schema, fact_grain,
        ) = self._bq_resolve_options(bq_billing)
        client = self._bq_client(creds)
        try:
            picks = self._bq_resolve_tables(
                client, layout=layout, dataset=dataset, pattern=pattern,
                # Batched mode resolves tables tenant-wide (scope=None →
                # per_month picks carry account_filter=None → no
                # @scope_acct clause), so one scan serves every account.
                scope=(None if batch_mode else scope),
                start=start, end=end,
            )
            if not picks:
                logger.info(
                    "AWS cost_facts (BQ): no CUR tables for account %s "
                    "in window %s..%s (layout=%s)",
                    ext, start, end, layout,
                )
                return []

            partition_pred, partition_params = self._bq_partition_predicate(
                layout, start, end,
            )

            # Tag harvest strategy is schema-dependent:
            #
            # * ``flat`` (CUR 1.0): per-account schema heterogeneity makes
            #   dynamic column discovery expensive (would need a separate
            #   INFORMATION_SCHEMA scan per pick). Stick to the narrow
            #   allowlist — five tags that the chargeback rule ladder
            #   actively keys off. Per-pick projection is gated on column
            #   presence; absent columns project as NULL so every UNION
            #   leg has the same shape.
            #
            # * ``nested`` (CUR 2.0): tags live in a single
            #   ``resource_tags.key_value`` ARRAY<STRUCT<key, value>>.
            #   Project the whole array via ``ANY_VALUE`` and let Python
            #   build the per-fact dict — one SELECT projection covers
            #   every tag the customer attached. Surfaces creator /
            #   owner / cost-center attribution on the dashboard without
            #   an upstream config change. Per-fact cap below bounds the
            #   row payload (AWS already caps at 50 tags / resource —
            #   the cap is defence-in-depth against malformed CUR data).
            tag_aliases = [
                ("owner", "Owner"),
                ("application", "Application"),
                ("environment", "Environment"),
                ("project", "Project"),
                ("department", "Department"),
            ]

            # Product-field expressions. ``product_product_name`` and
            # ``product_region`` move into the nested ``product``
            # RECORD in CUR 2.0.
            product_name_expr = self._bq_product_field_expression(
                tag_schema, "product_name",
            )
            product_region_expr = self._bq_product_field_expression(
                tag_schema, "region",
            )

            # Build the per-pick SELECT. Per-pick schema-drop logic
            # only applies to ``flat`` (CUR 1.0) — ``nested`` CUR 2.0
            # tables share a uniform schema across all per-month
            # tables so every pick projects every requested tag.
            #
            # ``fact_grain`` controls whether ``line_item_resource_id``
            # is in the GROUP BY (one fact per resource) or collapsed
            # via ``CAST(NULL AS STRING)`` (one fact per product).
            # ``per_product`` is the default to keep volume bounded —
            # see config-field docstring for the trade-off.
            def _select_for_pick(pick: _BqTablePick) -> str:
                if fact_grain == "per_product":
                    # Force per-product grain: NULL the resource_id so
                    # GROUP BY collapses every row of the same product
                    # to one bucket. The Python synthesises a stable
                    # ``aws-account:<acc>:<product>`` id downstream.
                    # ``line_item_resource_id`` column may or may not
                    # exist on the table — we never reference it.
                    resource_id_select = (
                        "CAST(NULL AS STRING) AS resource_id,\n              "
                    )
                elif tag_schema == "flat":
                    # per_resource + CUR 1.0 (flat): use line_item_resource_id
                    # when present in the table's schema, NULL when absent
                    # (per-account table heterogeneity, PR #311 follow-up).
                    present = pick.columns or set()
                    has_resource_id = "line_item_resource_id" in present
                    resource_id_select = (
                        "line_item_resource_id AS resource_id,\n              "
                        if has_resource_id else
                        "CAST(NULL AS STRING) AS resource_id,\n              "
                    )
                else:
                    # per_resource + CUR 2.0 (nested): uniform schema
                    # across all per-month tables; line_item_resource_id
                    # always present.
                    resource_id_select = (
                        "line_item_resource_id AS resource_id,\n              "
                    )

                if tag_schema == "flat":
                    present = pick.columns or set()
                    pick_tag_columns: list[tuple[str, str]] = [
                        (alias, self._bq_tag_column(tag_key))
                        for alias, tag_key in tag_aliases
                    ]
                    tag_selects = ",\n              ".join(
                        # For absent tag columns, project NULL so every
                        # UNION leg has the same column count + names.
                        f"ANY_VALUE({col}) AS tag_{alias}"
                        if col in present else
                        f"CAST(NULL AS STRING) AS tag_{alias}"
                        for alias, col in pick_tag_columns
                    )
                else:
                    # CUR 2.0 — surface ALL user-defined tags by
                    # projecting the whole nested ARRAY via ANY_VALUE.
                    # Python iterates ``row.tag_kv`` (list of
                    # STRUCT{key, value}) to build the per-fact dict,
                    # stripping the ``user_`` prefix that CUR 2.0
                    # stamps on user-defined entries.
                    tag_selects = "ANY_VALUE(resource_tags.key_value) AS tag_kv"
                account_filter_clause = (
                    " AND line_item_usage_account_id = @scope_acct"
                    if pick.account_filter is not None else ""
                )
                # 折扣前(原价)= CUR ``pricing_public_on_demand_cost``
                # (该行按公开按需价的成本)。列存在性必须用与 resource_id
                # / tag 完全相同的判定:CUR 2.0(nested,per_month)``pick.
                # columns`` 被显式置 None 表示"统一 schema、跳过 per-table
                # 列探测",该列恒存在 → 无条件投影;CUR 1.0(flat,per_
                # account)按 ``pick.columns`` 探测,缺列(老 CUR)投影 NULL
                # 保持每个 UNION leg 列一致。注:EDP / Private Rate 折扣是
                # **独立行**(service 落 'EDP Discount' 等、被各桶 service
                # 白名单过滤),桶内 gross vs net 只反映行级 RI/SP/私有费率,
                # 不含组织级 EDP —— AWS 账单结构使然,如实呈现。
                if tag_schema == "flat":
                    gross_present = pick.columns or set()
                    gross_native_expr = (
                        "pricing_public_on_demand_cost"
                        if "pricing_public_on_demand_cost" in gross_present
                        else "CAST(NULL AS FLOAT64)"
                    )
                    line_type_expr = (
                        "line_item_line_item_type"
                        if "line_item_line_item_type" in gross_present
                        else "CAST(NULL AS STRING)"
                    )
                else:
                    # CUR 2.0 — uniform schema, column always present.
                    gross_native_expr = "pricing_public_on_demand_cost"
                    line_type_expr = "line_item_line_item_type"
                # 双口径:白名单 list / 牌价 public 度量(见类顶
                # ``_LIST_WHITELIST_TYPES``),外加该组出现过的行类型
                # 集合(供 Python 侧 cost_bucket 分类;不进 GROUP BY,
                # 保持 fact_key 粒度稳定——同键混型组按 usage 优先分类,
                # list 度量已由行级 CASE 隔离,不受影响)。
                list_expr, public_expr = self._bq_list_measure_exprs(
                    line_type_expr=line_type_expr,
                    gross_expr=gross_native_expr,
                )
                # SP rows are kept (see SP semantics block at top of class).
                # SP negation rows land with product_code='ComputeSavingsPlans'
                # and synthesise resource_id='aws-account:<acc>:ComputeSavingsPlans'
                # below, so they appear as a single discount line per scope
                # per day rather than being scattered across underlying
                # services.
                #
                # GROUP BY grain matches the natural-key
                # ``compute_fact_key((date, integration, scope_external_id,
                # resource_id, usage_type))`` — adding ``usage_type`` to
                # GROUP BY keeps the BQ grain at one row per fact_key so
                # the upsert's UNIQUE(fact_key) doesn't collapse legit
                # cross-usage-type cells. ``region`` stays under
                # ANY_VALUE because a single resource_id maps to one
                # region (provider invariant).
                # ``discount_bucket`` mirrors ``_service_select``'s CASE:
                # EDP / Private Rate discounts land on dedicated named
                # buckets so the cost_facts.service column carries a
                # human-readable label (``'EDP Discount'`` /
                # ``'Private Rate Discount'``) instead of falling through
                # to the empty product_code → ``'Other'`` slice. Group-by
                # adds discount_bucket so different discount types stay
                # separable on the same (day, account, currency) — EDP
                # and PRD rows otherwise both have NULL product_code +
                # NULL resource_id and would collide into one fact row.
                # Non-discount rows get discount_bucket=NULL and group
                # exactly as before, preserving fact_key stability for
                # the upsert.
                return f"""
                SELECT
                  DATE(line_item_usage_start_date) AS day,
                  line_item_usage_account_id AS account_id,
                  line_item_product_code AS product_code,
                  ANY_VALUE({product_name_expr}) AS service_name,
                  CASE
                    WHEN line_item_line_item_type = 'EdpDiscount'
                      THEN 'EDP Discount'
                    WHEN line_item_line_item_type = 'PrivateRateDiscount'
                      THEN 'Private Rate Discount'
                    ELSE NULL
                  END AS discount_bucket,
                  {resource_id_select}line_item_currency_code AS currency,
                  line_item_usage_type AS usage_type,
                  ANY_VALUE({product_region_expr}) AS region,
                  ANY_VALUE(pricing_unit) AS usage_unit,
                  {tag_selects},
                  SUM(line_item_unblended_cost) AS native_cost,
                  SUM({gross_native_expr}) AS gross_native,
                  {list_expr} AS native_list,
                  {public_expr} AS native_public,
                  ARRAY_AGG(DISTINCT {line_type_expr} IGNORE NULLS) AS line_types,
                  SUM(line_item_usage_amount) AS usage_amount
                FROM `{pick.fq_table}`
                WHERE {partition_pred}
                  AND DATE(line_item_usage_start_date) BETWEEN @start AND @end
                  {account_filter_clause}
                GROUP BY day, account_id, product_code, discount_bucket, resource_id, currency, usage_type
                """

            sql = " UNION ALL ".join(_select_for_pick(p) for p in picks)
            account_filter = picks[0].account_filter
            params: list[Any] = [
                bigquery.ScalarQueryParameter("start", "DATE", start),
                bigquery.ScalarQueryParameter("end", "DATE", end),
            ] + partition_params
            if account_filter is not None:
                params.append(bigquery.ScalarQueryParameter(
                    "scope_acct", "STRING", account_filter,
                ))
            job = client.query(
                sql, job_config=self._bq_query_job_config(params),
            )

            # Cross-pick post-aggregation. In ``per_month`` layout, a
            # row whose ``line_item_usage_start_date`` lands in one
            # month but ``line_item_usage_end_date`` falls into the
            # next (rare — long-running usage reported at month
            # boundaries) appears in the next-month-table only. The
            # day grain is computed from ``start_date`` so it would
            # land under the FIRST month logically — but the partition
            # predicate matches the SECOND month's table by end_date.
            # Either way the day still uniquely identifies the row in
            # the output; summing into a dict keyed on
            # (day, account, product_code, resource_id, currency,
            # usage_type) is idempotent across UNION legs. ``per_account``
            # layout has 1 pick per scope so this loop's defensive sum
            # is a noop (every key seen once).
            agg: dict[tuple, dict[str, Any]] = {}
            for row in job.result():
                native = float(row.native_cost or 0)
                # NOT ``native == 0`` alone: a group whose invoice
                # (unblended) figure nets to zero can still carry a
                # NON-zero list figure — that is precisely the shape of
                # a resource fully covered by a Reserved Instance, which
                # the class docstring above says this path exists to
                # rescue via ``pricing_public_on_demand_cost``. Dropping
                # the group here threw the list measure away before it
                # was ever read, so the engineering-basis (list) view
                # would silently under-count every RI-covered hour and
                # diverge from the daily path, which has no such skip
                # (#2175).
                _row_list = getattr(row, "native_list", None)
                _row_public = getattr(row, "native_public", None)
                if native == 0.0 and not _row_list and not _row_public:
                    continue
                key = (
                    row.day, str(row.account_id),
                    str(row.product_code or ""),
                    # ``discount_bucket`` is part of the GROUP BY in
                    # ``_select_for_pick`` for EDP / PrivateRateDiscount
                    # rows — include it in the cross-pick key so the
                    # two discount types stay separable when the same
                    # (day, account, NULL product, NULL resource) tuple
                    # appears on multiple per-month UNION legs.
                    str(getattr(row, "discount_bucket", None) or ""),
                    row.resource_id, str(row.currency or "USD"),
                    str(row.usage_type or ""),
                )
                acc = agg.get(key)
                if acc is None:
                    acc = {
                        "service_name": row.service_name,
                        "region": row.region,
                        "usage_unit": row.usage_unit,
                        "native": 0.0,
                        "gross_native": 0.0,
                        "gross_seen": False,
                        "list_native": 0.0,
                        "list_seen": False,
                        "public_native": 0.0,
                        "public_seen": False,
                        "line_types": set(),
                        "usage_amount": 0.0,
                        "usage_amount_seen": False,
                        "row": row,
                    }
                    agg[key] = acc
                acc["native"] += native
                # 折扣前:列缺失时 gross_native 为 NULL,不计入(保持
                # None=未知 vs 0=真零 的区分);任一行有值即标记 seen。
                _gross = getattr(row, "gross_native", None)
                if _gross is not None:
                    acc["gross_native"] += float(_gross)
                    acc["gross_seen"] = True
                # 双口径:白名单 list / 牌价 public,同样 NULL≠0。
                _list = getattr(row, "native_list", None)
                if _list is not None:
                    acc["list_native"] += float(_list)
                    acc["list_seen"] = True
                _public = getattr(row, "native_public", None)
                if _public is not None:
                    acc["public_native"] += float(_public)
                    acc["public_seen"] = True
                for _lt in (getattr(row, "line_types", None) or []):
                    if _lt:
                        acc["line_types"].add(str(_lt))
                if row.usage_amount is not None:
                    acc["usage_amount"] += float(row.usage_amount)
                    acc["usage_amount_seen"] = True

            # Loud mixed-currency guard — mirrors the cmdb_cloud_billing_daily
            # sibling (which fails loudly on a multi-currency day rather than
            # silently dropping a row). ``compute_fact_key`` omits currency,
            # but this BQ aggregation key INCLUDES it (index 5), so two rows
            # identical except currency hash to the SAME fact_key and the
            # ON DUPLICATE KEY UPDATE upsert would silently discard one
            # currency's spend. PingCAP's CUR is USD-only end-to-end, so this
            # never fires today; if an AWS-China (Sinnet/NWCD, CNY-billed)
            # account is ever onboarded to BQ cost-facts mode, the operator
            # gets a loud failure here instead of silent data loss. The fix in
            # that scenario is to add ``currency`` to compute_fact_key's
            # natural key + extend the UNIQUE constraint (migration).
            _cur_by_natkey: dict[tuple, set[str]] = {}
            for _k in agg:
                _natkey = _k[:5] + _k[6:]  # drop currency (index 5)
                _cur_by_natkey.setdefault(_natkey, set()).add(_k[5])
            _mixed = {k: cs for k, cs in _cur_by_natkey.items() if len(cs) > 1}
            if _mixed:
                _ex_k, _ex_cs = next(iter(_mixed.items()))
                raise RuntimeError(
                    "AWS cost_facts (BQ): multi-currency rows collide on "
                    "fact_key (compute_fact_key omits currency). Example "
                    f"day={_ex_k[0]} account={_ex_k[1]} "
                    f"currencies={sorted(_ex_cs)}. Onboarding a non-USD "
                    "account to BQ cost-facts mode requires adding currency "
                    "to the fact_key natural key + UNIQUE (migration) first."
                )

            out: list[CostFact] = []
            integration_id = self.integration_id
            for _key, acc in agg.items():
                row = acc["row"]
                native = float(acc["native"])
                # Drop TRUE-zero rows so the upsert isn't burdened with
                # millions of micro-credit lines that net to 0. Negative
                # rows (credits, refunds) are kept — they're meaningful
                # for monthly reconciliation. "True zero" means zero on
                # EVERY measure: an invoice-zero group with a non-zero
                # list figure is an RI-covered resource, and dropping it
                # loses that usage from the engineering-basis view
                # entirely (#2175).
                if (
                    native == 0.0
                    and not acc.get("list_native")
                    and not acc.get("public_native")
                ):
                    continue
                # Resolve the scope this row's account belongs to. In
                # single-scope mode every row is the pinned account, so
                # ``scope`` is correct. In batched (all-accounts) mode
                # the account varies per row: map it to its registered
                # target scope, skipping rows for accounts the cron never
                # registered (other tenants sharing the month-table).
                if batch_mode:
                    fact_scope = scope_by_account.get(str(row.account_id))
                    if fact_scope is None:
                        continue
                else:
                    fact_scope = scope
                currency = str(row.currency or "USD")
                # When the CUR row has no resource_id (some service-level
                # billing has resource_id=NULL even with "Include resource
                # IDs" — e.g. SavingsPlanCoveredUsage credits, marketplace
                # fees) OR when the CUR table itself lacks the
                # ``line_item_resource_id`` column (PingCAP's older
                # per-account CURs, projected as ``CAST(NULL AS STRING)``
                # in the per-pick SELECT above), synthesise a stable id
                # keyed on (account, product_code) so the natural-key
                # hash is unique and the upsert behaves.
                resource_id: str | None = None
                raw_rid = row.resource_id
                if raw_rid:
                    resource_id = str(raw_rid)
                if not resource_id:
                    # Discount-bucket rows (EDP / Private Rate) both
                    # have NULL product_code so the legacy fallback
                    # ``aws-account:<acc>:None`` would synthesise an
                    # IDENTICAL id for both, colliding on fact_key
                    # (UNIQUE(date,integration,scope,resource,usage_type)
                    # → upsert keeps only one). Use the bucket label as
                    # the synthesised suffix so EDP and PRD land on
                    # distinct fact_keys per (day, account).
                    discount_bucket_raw = getattr(row, "discount_bucket", None)
                    suffix = (
                        str(discount_bucket_raw)
                        if discount_bucket_raw
                        else str(row.product_code or "")
                    )
                    resource_id = f"aws-account:{row.account_id}:{suffix}"
                    resource_type = "aws-aggregated"
                else:
                    resource_type = _aws_resource_type_from_id(resource_id)

                if tag_schema == "flat":
                    tags = self._cur1_extract_tags_from_row(row, tag_aliases)
                else:
                    tags, malformed_struct_count = (
                        self._cur2_extract_tags_from_row(row)
                    )
                    if malformed_struct_count > 0:
                        # Single aggregated WARN per fact-row so a
                        # systemic CUR schema regression surfaces in
                        # logs without per-tag spam. ``resource_id``
                        # (NULL on aggregated rows) and account id
                        # narrow the blast radius for an operator.
                        logger.warning(
                            "AWS cost_facts: dropped %d malformed "
                            "resource_tags STRUCT items (missing "
                            "key/value field) on account=%s "
                            "resource=%s day=%s",
                            malformed_struct_count,
                            getattr(row, "account_id", "<unknown>"),
                            getattr(row, "resource_id", None) or "<aggregate>",
                            row.day,
                        )

                amount_usd_value = _convert_to_usd(
                    amount=native,
                    currency=currency,
                    cny_per_usd=getattr(
                        self, "_cny_per_usd_for_cost_facts", None,
                    ),
                    integration_id=str(integration_id),
                )
                # 折扣前:仅当该 fact 至少有一行带 publicOnDemandCost 时
                # 才有值;USD 走与净额相同的换算路径,保证可比。
                gross_native_value: float | None = (
                    float(acc["gross_native"]) if acc["gross_seen"] else None
                )
                gross_usd_value: float | None = (
                    _convert_to_usd(
                        amount=gross_native_value,
                        currency=currency,
                        cny_per_usd=getattr(
                            self, "_cny_per_usd_for_cost_facts", None,
                        ),
                        integration_id=str(integration_id),
                    )
                    if gross_native_value is not None else None
                )
                # Bucket every SP-flavor row (ComputeSavingsPlans /
                # EC2InstanceSavingsPlans / SageMakerSavingsPlans) under
                # a single ``'Savings Plans'`` service so the per-scope
                # detail drawer's "cost by service" donut (reading
                # cost_facts) renders the same label as the per-day
                # service breakdown stored in cmdb_cloud_billing_daily
                # (which already applies the same CASE in
                # ``_service_select``). Without this rebucketing the two
                # surfaces showed different legend keys for the same
                # row class — caught by the round-2 review. Allocation
                # lookups downstream key off ``fact.service ==
                # 'Savings Plans'`` to skip per-person chargeback for
                # discount lines (see ``cost_service.ingest_cost_facts``).
                product_code_str = str(row.product_code or "")
                # ``discount_bucket`` from the SQL CASE takes priority —
                # CUR 2.0 EdpDiscount / PrivateRateDiscount rows have
                # NULL product_code so the legacy fallback below would
                # otherwise emit ``'Other'`` for them, hiding the
                # enterprise-discount line under a generic legend slice.
                # The SP-flavor check below stays for the per_account
                # CUR 1.0 path where SP rows arrive as
                # product_code='ComputeSavingsPlans' (no discount_bucket
                # set by the SQL CASE there).
                discount_bucket = getattr(row, "discount_bucket", None)
                if discount_bucket:
                    service_label = str(discount_bucket)
                elif product_code_str.endswith("SavingsPlans"):
                    service_label = "Savings Plans"
                else:
                    service_label = str(
                        row.service_name or product_code_str or "Other",
                    )
                # 双口径:list / public 走与净额相同的 USD 换算;
                # NULL-seen 组透传 None(=口径未接入,绝不伪造 0)。
                list_native_value: float | None = (
                    float(acc["list_native"]) if acc["list_seen"] else None
                )
                list_usd_value: float | None = (
                    _convert_to_usd(
                        amount=list_native_value,
                        currency=currency,
                        cny_per_usd=getattr(
                            self, "_cny_per_usd_for_cost_facts", None,
                        ),
                        integration_id=str(integration_id),
                    )
                    if list_native_value is not None else None
                )
                public_native_value: float | None = (
                    float(acc["public_native"]) if acc["public_seen"] else None
                )
                public_usd_value: float | None = (
                    _convert_to_usd(
                        amount=public_native_value,
                        currency=currency,
                        cny_per_usd=getattr(
                            self, "_cny_per_usd_for_cost_facts", None,
                        ),
                        integration_id=str(integration_id),
                    )
                    if public_native_value is not None else None
                )
                bucket = self._classify_cur_bucket(
                    sorted(acc["line_types"]),
                    discount_bucket=(
                        str(discount_bucket) if discount_bucket else None
                    ),
                    product_code=product_code_str,
                    native=native,
                )
                out.append(CostFact(
                    date=row.day,
                    integration_id=integration_id,
                    service=service_label,
                    amount_usd=amount_usd_value,
                    amount_native=native,
                    amount_usd_gross=gross_usd_value,
                    amount_native_gross=gross_native_value,
                    amount_usd_list=list_usd_value,
                    amount_native_list=list_native_value,
                    amount_usd_public=public_usd_value,
                    cost_bucket=bucket,
                    currency=currency,
                    scope=fact_scope,
                    resource_group=str(fact_scope.external_id),
                    resource_id=resource_id,
                    resource_name=(
                        resource_id.rsplit("/", 1)[-1]
                        if "/" in resource_id else resource_id
                    ),
                    resource_type=resource_type,
                    region=str(row.region) if row.region else None,
                    usage_type=str(row.usage_type) if row.usage_type else None,
                    tags=tags,
                    # Cross-pick SUM of ``line_item_usage_amount`` —
                    # exposes the cross-month total in per_month layout.
                    # ``None`` only when every contributing row had NULL
                    # usage_amount (matching the legacy single-pick
                    # semantics where ``row.usage_amount is None``
                    # mapped to ``None``).
                    usage_quantity=(
                        float(acc["usage_amount"])
                        if acc["usage_amount_seen"] else None
                    ),
                    usage_unit=str(row.usage_unit) if row.usage_unit else None,
                    raw={
                        "bq_source": "cur",
                        "product_code": str(row.product_code or ""),
                    },
                ))
            return out
        finally:
            # Close the BQ client deterministically — see
            # ``_list_billing_daily_bq_sync`` for the FD-leak rationale.
            try:
                client.close()
            except Exception:  # noqa: BLE001
                pass

    # ── Cost-facts ingestion: IAM principal-tag mode ─────────────
    #
    # Activated for accounts where the primary cost driver has NO
    # taggable billable resource (Bedrock on-demand InvokeModel is
    # the canonical case — the foundation model itself is shared
    # across customers, so resource-level tags don't reach the bill).
    # AWS exposes a separate cost-allocation surface for those:
    # **IAM principal tags**. Once the admin activates a key (e.g.
    # ``Owner``) in Billing Console → Cost allocation tags, every
    # cost line item incurred by an IAM principal carrying that tag
    # is grouped in Cost Explorer under the dimension
    # ``iamPrincipal/<key>`` (NOT ``aws:PrincipalTag/<key>`` — that
    # one is the IAM-policy evaluator's namespace and is silently
    # rejected by CE GroupBy, returning every row tagged empty).
    #
    # The ingest emits one fact per (date × service × tag-value)
    # cell. ``GetCostAndUsage`` (without WithResources) is used so
    # we keep the full 13-month window — the 14-day limit only
    # applies to per-resource queries. ``resource_id`` is synthetic
    # so the ``compute_fact_key`` natural-key hash stays unique per
    # (owner, service); ``tags[<key>]`` carries the canonical owner
    # value the cost-allocation rule ladder matches against.
    def _list_cost_facts_principal_tag_sync(
        self, start: date, end: date, scope: CloudScopeRef | None,
        principal_tag_keys: list[str],
    ) -> list[CostFact]:
        # Mirrors ``_list_cost_facts_sync`` for the LINKED_ACCOUNT
        # contract — every Cost Explorer call must be pinned to a
        # 12-digit linked-account id; tenant-wide queries are
        # ambiguous on Org-master credentials.
        ce = self._boto_client("ce", region="us-east-1")
        if scope is None or not scope.external_id:
            raise NotImplementedError(
                "AWS list_cost_facts (principal-tag mode) requires an "
                "explicit scope (linked-account); the cron handler fans "
                "out per scope.",
            )
        ext = str(scope.external_id)
        if not (ext.isdigit() and len(ext) == 12):
            raise RuntimeError(
                f"AWS cost-facts: scope external_id {ext!r} is not a "
                "12-digit AWS account id; cannot pin LINKED_ACCOUNT.",
            )
        scope_filter: dict[str, Any] = {"Dimensions": {
            "Key": "LINKED_ACCOUNT", "Values": [ext],
        }}
        time_period = {
            "Start": start.isoformat(),
            "End": (end + timedelta(days=1)).isoformat(),
        }

        out: list[CostFact] = []
        integration_id = self.integration_id

        for tag_key in principal_tag_keys:
            if not tag_key or not tag_key.strip():
                continue
            tag_key = tag_key.strip()
            # AWS Cost Allocation Tag namespace for IAM principal tags
            # is **``iamPrincipal/<Key>``** (no ``aws:`` prefix) — that's
            # the literal key admins see in Billing Console → Cost
            # allocation tags → User-defined cost allocation tags
            # (shown as ``iamPrincipal/Owner`` once activated).
            #
            # **DO NOT use ``aws:PrincipalTag/<Key>``** — that is the
            # IAM-policy-evaluator dimension (used in
            # ``Condition: {StringEquals: {'aws:PrincipalTag/Owner': '...'}}``
            # policy statements). When passed as a Cost Explorer
            # ``GroupBy[Type=TAG]`` key, AWS does NOT error but returns
            # an empty string for every row's tag value, collapsing
            # every line item into the ``(unallocated)`` bucket. The
            # 2026-05-12 audit found this silent fail caused 100% of
            # Bedrock production cost to land unallocated — see
            # follow-up to PR #265. Verified against live CE:
            # ``aws:PrincipalTag/Owner`` → all rows tagged empty;
            # ``iamPrincipal/Owner`` → correctly grouped by the IAM
            # user's tag value.
            ce_tag_key = f"iamPrincipal/{tag_key}"
            # 双口径白名单遍(RECORD_TYPE=Usage):同一分组多跑一遍带
            # 记账类型过滤的查询,得到每 (day × service × owner) 单元的
            # 纯用量金额 → list。单元里混入的 Credit / Tax / Refund 只
            # 存在于发票口径。SG 无 SP(2026-07 实测),CE 的 Usage 记
            # 账类型即完整白名单;public 恒 None(CE 无公开牌价度量)。
            whitelist_cells: dict[tuple[date, str, str], float] = {}
            wl_filter: dict[str, Any] = {"And": [
                scope_filter,
                {"Dimensions": {"Key": "RECORD_TYPE", "Values": ["Usage"]}},
            ]}
            try:
                for resp in _ce_paged(
                    ce,
                    TimePeriod=time_period,
                    Granularity="DAILY",
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "SERVICE"},
                        {"Type": "TAG", "Key": ce_tag_key},
                    ],
                    Filter=wl_filter,
                ):
                    for bucket in resp.get("ResultsByTime") or []:
                        d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                        for grp in bucket.get("Groups") or []:
                            keys = grp.get("Keys") or []
                            service = keys[0] if len(keys) > 0 else "Other"
                            raw_owner = keys[1] if len(keys) > 1 else ""
                            owner_value = (
                                raw_owner.split("$", 1)[1]
                                if "$" in raw_owner else raw_owner
                            ).strip()
                            owner_label = owner_value or "(unallocated)"
                            amt = float(
                                (grp.get("Metrics") or {})
                                .get("UnblendedCost", {}).get("Amount") or 0,
                            )
                            if amt:
                                cell = (d, service or "Other", owner_label)
                                whitelist_cells[cell] = (
                                    whitelist_cells.get(cell, 0.0) + amt
                                )
                for resp in _ce_paged(
                    ce,
                    TimePeriod=time_period,
                    Granularity="DAILY",
                    Metrics=["UnblendedCost", "UsageQuantity"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "SERVICE"},
                        {"Type": "TAG", "Key": ce_tag_key},
                    ],
                    Filter=scope_filter,
                ):
                    for bucket in resp.get("ResultsByTime") or []:
                        d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                        for grp in bucket.get("Groups") or []:
                            keys = grp.get("Keys") or []
                            service = keys[0] if len(keys) > 0 else "Other"
                            # CE returns tag values as ``"key$value"`` —
                            # strip the prefix when present. An empty
                            # value (after strip) means the principal
                            # had no tag set; we keep the cost row but
                            # mark the synthetic resource_id with an
                            # ``(unallocated)`` sentinel so the chargeback
                            # ladder leaves it on the unallocated bucket.
                            raw_owner = keys[1] if len(keys) > 1 else ""
                            owner_value = (
                                raw_owner.split("$", 1)[1]
                                if "$" in raw_owner else raw_owner
                            ).strip()
                            metrics = grp.get("Metrics") or {}
                            amount = float(
                                (metrics.get("UnblendedCost") or {}).get(
                                    "Amount",
                                ) or 0,
                            )
                            # The zero-skip has to know about BOTH
                            # measures. A cell whose Usage is exactly
                            # cancelled by a same-cell Credit nets to
                            # zero on the invoice basis while still
                            # carrying real whitelist (list) usage —
                            # skipping here dropped that owner's usage
                            # for the day with no warning (#2180). Same
                            # rule the BQ path states: keep credits,
                            # drop only TRUE zeros.
                            _wl_probe = whitelist_cells.get(
                                (
                                    d, service or "Other",
                                    owner_value or "(unallocated)",
                                ),
                            )
                            if amount == 0 and not _wl_probe:
                                continue
                            currency = (
                                metrics.get("UnblendedCost") or {}
                            ).get("Unit", "USD")
                            amount_usd_value = _convert_to_usd(
                                amount=amount,
                                currency=currency,
                                cny_per_usd=getattr(
                                    self,
                                    "_cny_per_usd_for_cost_facts",
                                    None,
                                ),
                                integration_id=str(integration_id),
                            )
                            usage = (metrics.get("UsageQuantity") or {})
                            owner_label = owner_value or "(unallocated)"
                            # Synthetic resource_id encodes (key, owner,
                            # service) so the natural-key hash stays
                            # unique per (date × scope × cell). Two
                            # different principal-tag keys (rare — most
                            # tenants run one) coexist without colliding.
                            synthetic_resource_id = (
                                f"aws-principal:{tag_key}={owner_label}"
                                f"|service={service or 'Other'}"
                            )
                            tags_dict = (
                                {tag_key: owner_value} if owner_value else {}
                            )
                            # 双口径:该单元的白名单遍金额 → list(单元
                            # 混入的 Credit/Tax 只在发票口径);无白名单
                            # 金额的单元(纯 credit/税)按符号归桶。
                            wl_amt = whitelist_cells.get(
                                (d, service or "Other", owner_label),
                            )
                            if wl_amt is not None:
                                cell_bucket = COST_BUCKET_USAGE
                                wl_usd = _convert_to_usd(
                                    amount=wl_amt,
                                    currency=currency,
                                    cny_per_usd=getattr(
                                        self,
                                        "_cny_per_usd_for_cost_facts",
                                        None,
                                    ),
                                    integration_id=str(integration_id),
                                )
                            else:
                                cell_bucket = (
                                    COST_BUCKET_CREDIT
                                    if amount < 0 else COST_BUCKET_OTHER
                                )
                                wl_usd = None
                            out.append(CostFact(
                                date=d,
                                integration_id=integration_id,
                                service=service or "Other",
                                amount_usd=amount_usd_value,
                                amount_native=amount,
                                amount_usd_list=wl_usd,
                                amount_native_list=(
                                    wl_amt if wl_amt is not None else None
                                ),
                                cost_bucket=cell_bucket,
                                currency=currency,
                                scope=scope,
                                resource_group=str(scope.external_id),
                                resource_id=synthetic_resource_id,
                                resource_name=(
                                    f"{service or 'Other'} · "
                                    f"{tag_key}={owner_label}"
                                ),
                                resource_type=None,
                                region=None,
                                usage_type=None,
                                tags=tags_dict,
                                usage_quantity=(
                                    float(usage.get("Amount") or 0)
                                    if usage else None
                                ),
                                usage_unit=(
                                    usage.get("Unit") if usage else None
                                ),
                                raw={
                                    "ce_group": grp,
                                    "principal_tag_key": tag_key,
                                },
                            ))
            except (ClientError, BotoCoreError) as exc:
                raise RuntimeError(
                    f"AWS Cost Explorer GetCostAndUsage "
                    f"(principal-tag {tag_key!r}) failed: {exc}",
                ) from exc
        return out

    async def list_bedrock_profile_facts(
        self, start: date, end: date, scope: CloudScopeRef | None = None,
        *, owner_tag_key: str, profile_tag_key: str,
        cny_per_usd: float | None = None,
    ) -> list[dict[str, Any]]:
        """Per-(date × owner × profile × service) Bedrock spend rows.

        The "profile" is a secondary IAM-principal cost-allocation tag
        (e.g. ``UsedBy``) that subdivides one owner's Bedrock spend into
        its workloads. Cost Explorer's GroupBy is capped at 2 dimensions,
        so we fan out: one discovery call grouped by the owner tag, then
        one call per owner filtered to that owner and grouped by
        (profile-tag × SERVICE). Owner count is small (a handful of
        principals per Bedrock account), so the fan-out is a few cheap
        CE calls per window.

        These rows cover ALL services the owner touched (the owner tag is
        an IAM-principal tag, not Bedrock-scoped). The caller restricts to
        the Bedrock service set — the SAME AI-whitelist patterns the per-AK
        ``mtd_usd`` uses — so the two surfaces reconcile by construction.
        ``cny_per_usd`` is passed explicitly (the caller resolves it the
        same way the main cost-facts sync does); AWS is USD-billed today so
        it is normally None, but a CNY line still converts correctly.

        Returns dicts (not ``CostFact``) — these land in the isolated
        ``cmdb_bedrock_profile_facts`` table, not ``cmdb_cost_facts``.
        """
        if not _BOTO3_AVAILABLE:
            raise RuntimeError(
                "boto3 not importable — AWS profile-facts ingestion "
                "cannot run.",
            )
        return await asyncio.to_thread(
            self._list_bedrock_profile_facts_sync,
            start, end, scope, owner_tag_key, profile_tag_key, cny_per_usd,
        )

    def _list_bedrock_profile_facts_sync(
        self, start: date, end: date, scope: CloudScopeRef | None,
        owner_tag_key: str, profile_tag_key: str,
        cny_per_usd: float | None = None,
    ) -> list[dict[str, Any]]:
        ce = self._boto_client("ce", region="us-east-1")
        if scope is None or not scope.external_id:
            raise NotImplementedError(
                "AWS list_bedrock_profile_facts requires an explicit "
                "scope (linked-account); the cron fans out per scope.",
            )
        ext = str(scope.external_id)
        if not (ext.isdigit() and len(ext) == 12):
            raise RuntimeError(
                f"AWS profile-facts: scope external_id {ext!r} is not a "
                "12-digit AWS account id; cannot pin LINKED_ACCOUNT.",
            )
        owner_ce_key = f"iamPrincipal/{owner_tag_key.strip()}"
        # The profile tag key is used VERBATIM as the Cost Explorer
        # cost-allocation tag key — NOT auto-prefixed with ``iamPrincipal/``
        # like the owner tag. The owner (``Owner``) is an IAM-principal
        # session tag, exposed in CE as ``iamPrincipal/Owner``. The profile
        # tag (``UsedBy``) is a plain user-defined resource cost-allocation
        # tag, exposed in CE under its bare key ``UsedBy`` — querying
        # ``iamPrincipal/UsedBy`` returns an empty value for every row and
        # collapses the whole breakdown into ``(unallocated)`` (caught in
        # the 2026-06-17 prod backfill). Admins set ``aws_profile_tag`` to
        # the EXACT key shown in Billing Console → Cost allocation tags
        # (e.g. ``UsedBy``, or ``iamPrincipal/Foo`` if it is a principal
        # tag), so verbatim handles both namespaces.
        profile_ce_key = profile_tag_key.strip()
        time_period = {
            "Start": start.isoformat(),
            "End": (end + timedelta(days=1)).isoformat(),
        }
        scope_dim: dict[str, Any] = {"Dimensions": {
            "Key": "LINKED_ACCOUNT", "Values": [ext],
        }}
        cny = cny_per_usd
        integration_id = self.integration_id

        def _strip(raw: str) -> str:
            # CE returns tag values as ``"key$value"``.
            return (raw.split("$", 1)[1] if "$" in raw else raw).strip()

        try:
            # 1. Discover the owner tag values with spend in the window.
            owners: set[str] = set()
            for resp in _ce_paged(
                ce,
                TimePeriod=time_period,
                Granularity="DAILY",
                Metrics=["UnblendedCost"],
                GroupBy=[{"Type": "TAG", "Key": owner_ce_key}],
                Filter=scope_dim,
            ):
                for bucket in resp.get("ResultsByTime") or []:
                    for grp in bucket.get("Groups") or []:
                        keys = grp.get("Keys") or []
                        owner = _strip(keys[0]) if keys else ""
                        # Skip the empty/unallocated owner — it maps to no
                        # API key, so a per-AK detail page never reads it.
                        if owner:
                            owners.add(owner)

            # 2. Per owner: break spend down by (profile-tag × SERVICE).
            out: list[dict[str, Any]] = []
            for owner in sorted(owners):
                owner_filter = {"And": [
                    scope_dim,
                    {"Tags": {"Key": owner_ce_key, "Values": [owner]}},
                ]}
                for resp in _ce_paged(
                    ce,
                    TimePeriod=time_period,
                    Granularity="DAILY",
                    Metrics=["UnblendedCost", "UsageQuantity"],
                    GroupBy=[
                        {"Type": "TAG", "Key": profile_ce_key},
                        {"Type": "DIMENSION", "Key": "SERVICE"},
                    ],
                    Filter=owner_filter,
                ):
                    for bucket in resp.get("ResultsByTime") or []:
                        d = date.fromisoformat(
                            bucket["TimePeriod"]["Start"],
                        )
                        for grp in bucket.get("Groups") or []:
                            keys = grp.get("Keys") or []
                            raw_profile = keys[0] if len(keys) > 0 else ""
                            profile = _strip(raw_profile) or "(unallocated)"
                            service = keys[1] if len(keys) > 1 else "Other"
                            metrics = grp.get("Metrics") or {}
                            amount = float(
                                (metrics.get("UnblendedCost") or {}).get(
                                    "Amount",
                                ) or 0,
                            )
                            if amount == 0:
                                continue
                            currency = (
                                metrics.get("UnblendedCost") or {}
                            ).get("Unit", "USD")
                            amount_usd_value = _convert_to_usd(
                                amount=amount,
                                currency=currency,
                                cny_per_usd=cny,
                                integration_id=str(integration_id),
                            )
                            usage = metrics.get("UsageQuantity") or {}
                            out.append({
                                "date": d,
                                "owner_email": owner,
                                "profile": profile,
                                "service": service or "Other",
                                "amount_usd": amount_usd_value,
                                "amount_native": amount,
                                "currency": currency,
                                "usage_quantity": (
                                    float(usage.get("Amount") or 0)
                                    if usage else None
                                ),
                                "usage_unit": (
                                    usage.get("Unit") if usage else None
                                ),
                            })
            return out
        except (ClientError, BotoCoreError) as exc:
            raise RuntimeError(
                f"AWS Cost Explorer GetCostAndUsage "
                f"(bedrock-profile {profile_tag_key!r}) failed: {exc}",
            ) from exc

    async def list_bedrock_per_key_profile_facts(
        self, start: date, end: date, scope: CloudScopeRef | None = None,
        *, owner_tag_key: str, key_tag_key: str, sub_tag_key: str,
        cny_per_usd: float | None = None,
    ) -> list[dict[str, Any]]:
        """Per-(date × API-Key × UsedBy × service) Bedrock spend rows —
        the per-API-Key axis CROSSED with the ``UsedBy`` workload tag
        (migration 452).

        The plain :meth:`list_bedrock_profile_facts` fans out by OWNER and
        groups by (secondary-tag × SERVICE), so its per-key rows carry no
        workload dimension: two keys held by one person share the same
        person-level ``UsedBy`` table (the bug xuhuaiyu reported). Cost
        Explorer caps GroupBy at 2 dimensions, but ``Filter`` is free — so
        to get (key × UsedBy × SERVICE) we FIX the key via Filter and
        group by (UsedBy × SERVICE), one call per spending key.

        ``key_tag_key`` is the CE cost-allocation tag for the API Key
        (``iamPrincipal/helix-request-number``, used verbatim — it already
        carries the ``iamPrincipal/`` namespace). ``sub_tag_key`` is the
        workload tag (``UsedBy``, a bare resource tag, verbatim). The
        ``owner_tag_key`` is only used to stamp an informational
        ``owner_email`` on each row (per-key reads filter by key, never by
        owner); it is the same ``iamPrincipal/Owner`` principal tag the
        main sync uses.

        Verified against prod 2026-08-04: for account 841162690310 the
        per-key crosses summed to the owner total to the cent, so this
        table stays reconciled with ``mtd_usd`` by construction.

        Returns dicts (not ``CostFact``) with a ``sub_profile`` key — they
        land in ``cmdb_bedrock_profile_facts`` beside the workload-axis
        rows, stamped ``profile_tag_key = key_tag_key`` by the caller.
        """
        if not _BOTO3_AVAILABLE:
            raise RuntimeError(
                "boto3 not importable — AWS profile-facts ingestion "
                "cannot run.",
            )
        return await asyncio.to_thread(
            self._list_bedrock_per_key_profile_facts_sync,
            start, end, scope, owner_tag_key, key_tag_key, sub_tag_key,
            cny_per_usd,
        )

    def _list_bedrock_per_key_profile_facts_sync(
        self, start: date, end: date, scope: CloudScopeRef | None,
        owner_tag_key: str, key_tag_key: str, sub_tag_key: str,
        cny_per_usd: float | None = None,
    ) -> list[dict[str, Any]]:
        ce = self._boto_client("ce", region="us-east-1")
        if scope is None or not scope.external_id:
            raise NotImplementedError(
                "AWS list_bedrock_per_key_profile_facts requires an "
                "explicit scope (linked-account); the cron fans out per "
                "scope.",
            )
        ext = str(scope.external_id)
        if not (ext.isdigit() and len(ext) == 12):
            raise RuntimeError(
                f"AWS per-key profile-facts: scope external_id {ext!r} is "
                "not a 12-digit AWS account id; cannot pin LINKED_ACCOUNT.",
            )
        owner_ce_key = f"iamPrincipal/{owner_tag_key.strip()}"
        # Verbatim — the API-Key axis tag already carries its namespace
        # (``iamPrincipal/helix-request-number``); the workload tag is a
        # bare user-defined resource tag (``UsedBy``). Same verbatim rule
        # the sibling method documents at length.
        key_ce_key = key_tag_key.strip()
        sub_ce_key = sub_tag_key.strip()
        time_period = {
            "Start": start.isoformat(),
            "End": (end + timedelta(days=1)).isoformat(),
        }
        scope_dim: dict[str, Any] = {"Dimensions": {
            "Key": "LINKED_ACCOUNT", "Values": [ext],
        }}
        cny = cny_per_usd
        integration_id = self.integration_id

        def _strip(raw: str) -> str:
            return (raw.split("$", 1)[1] if "$" in raw else raw).strip()

        try:
            # 1. Discover which API keys have spend in the window AND the
            #    owner tag they carry — ONE call, GroupBy=(key × owner).
            #    Keys with spend but no ``helix-request-number`` tag (bare
            #    console usage, non-Helix IAM users) map to no AK detail
            #    page, so they're skipped here rather than fanned out.
            key_owner_spend: dict[str, dict[str, float]] = {}
            for resp in _ce_paged(
                ce,
                TimePeriod=time_period,
                Granularity="DAILY",
                Metrics=["UnblendedCost"],
                GroupBy=[
                    {"Type": "TAG", "Key": key_ce_key},
                    {"Type": "TAG", "Key": owner_ce_key},
                ],
                Filter=scope_dim,
            ):
                for bucket in resp.get("ResultsByTime") or []:
                    for grp in bucket.get("Groups") or []:
                        keys = grp.get("Keys") or []
                        key_ref = _strip(keys[0]) if keys else ""
                        owner = _strip(keys[1]) if len(keys) > 1 else ""
                        if not key_ref:
                            continue
                        amount = float(
                            (grp.get("Metrics", {}).get("UnblendedCost")
                             or {}).get("Amount") or 0,
                        )
                        key_owner_spend.setdefault(key_ref, {})
                        key_owner_spend[key_ref][owner] = (
                            key_owner_spend[key_ref].get(owner, 0.0) + amount
                        )

            # 2. Per key: (UsedBy × SERVICE) with the key fixed via Filter.
            out: list[dict[str, Any]] = []
            for key_ref in sorted(key_owner_spend):
                owners = key_owner_spend[key_ref]
                # Informational owner stamp: the tag value that carried
                # the most spend for this key, deterministic tiebreak on
                # the owner string. A key that changed hands mid-window
                # keeps ONE stamp — reads never filter per-key rows by
                # owner, so this is a label, not an attribution key.
                owner_email = (
                    max(owners.items(), key=lambda kv: (kv[1], kv[0]))[0]
                    if owners else ""
                )
                key_filter = {"And": [
                    scope_dim,
                    {"Tags": {"Key": key_ce_key, "Values": [key_ref]}},
                ]}
                for resp in _ce_paged(
                    ce,
                    TimePeriod=time_period,
                    Granularity="DAILY",
                    Metrics=["UnblendedCost", "UsageQuantity"],
                    GroupBy=[
                        {"Type": "TAG", "Key": sub_ce_key},
                        {"Type": "DIMENSION", "Key": "SERVICE"},
                    ],
                    Filter=key_filter,
                ):
                    for bucket in resp.get("ResultsByTime") or []:
                        d = date.fromisoformat(
                            bucket["TimePeriod"]["Start"],
                        )
                        for grp in bucket.get("Groups") or []:
                            keys = grp.get("Keys") or []
                            raw_sub = keys[0] if len(keys) > 0 else ""
                            sub_profile = _strip(raw_sub) or "(unallocated)"
                            service = keys[1] if len(keys) > 1 else "Other"
                            metrics = grp.get("Metrics") or {}
                            amount = float(
                                (metrics.get("UnblendedCost") or {}).get(
                                    "Amount",
                                ) or 0,
                            )
                            if amount == 0:
                                continue
                            currency = (
                                metrics.get("UnblendedCost") or {}
                            ).get("Unit", "USD")
                            amount_usd_value = _convert_to_usd(
                                amount=amount,
                                currency=currency,
                                cny_per_usd=cny,
                                integration_id=str(integration_id),
                            )
                            usage = metrics.get("UsageQuantity") or {}
                            out.append({
                                "date": d,
                                "owner_email": owner_email,
                                "profile": key_ref,
                                "sub_profile": sub_profile,
                                "service": service or "Other",
                                "amount_usd": amount_usd_value,
                                "amount_native": amount,
                                "currency": currency,
                                "usage_quantity": (
                                    float(usage.get("Amount") or 0)
                                    if usage else None
                                ),
                                "usage_unit": (
                                    usage.get("Unit") if usage else None
                                ),
                            })
            return out
        except (ClientError, BotoCoreError) as exc:
            raise RuntimeError(
                f"AWS Cost Explorer GetCostAndUsage "
                f"(bedrock per-key × {sub_tag_key!r}) failed: {exc}",
            ) from exc

    # ── Cost Explorer custom-dimension query ─────────────────────
    #
    # Cost Explorer's ``GetCostAndUsage`` accepts up to TWO GroupBy
    # entries per call. Each is either a DIMENSION (SERVICE /
    # LINKED_ACCOUNT / REGION / USAGE_TYPE / RECORD_TYPE /
    # INSTANCE_TYPE_FAMILY / OPERATION / AVAILABILITY_ZONE /
    # PURCHASE_TYPE) or a TAG key. The Helix API surfaces canonical
    # lowercase aliases so the UI doesn't leak AWS-specific
    # spelling — we map them at the boundary.
    _CE_DIMENSION_ALIASES: dict[str, str] = {
        "service": "SERVICE",
        "linked_account": "LINKED_ACCOUNT",
        "account": "LINKED_ACCOUNT",       # convenience alias
        "region": "REGION",
        "usage_type": "USAGE_TYPE",
        "record_type": "RECORD_TYPE",
        "instance_type_family": "INSTANCE_TYPE_FAMILY",
        "operation": "OPERATION",
        "availability_zone": "AVAILABILITY_ZONE",
        "purchase_type": "PURCHASE_TYPE",
    }

    @classmethod
    def supported_billing_dimensions(cls) -> frozenset[str]:
        # ``tag`` declares "any tag:KEY is acceptable" — see
        # CloudDriver.supported_billing_dimensions docstring.
        return frozenset({*cls._CE_DIMENSION_ALIASES.keys(), "tag"})

    async def list_billing_by_dimension(
        self,
        start_date: date,
        end_date: date,
        *,
        group_by: list[str],
        scope: CloudScopeRef | None = None,
        filters: dict[str, list[str]] | None = None,
        bq_billing: dict[str, Any] | None = None,
    ) -> list[BillingByDimensionRow]:
        if not group_by:
            raise ValueError("group_by must contain at least one dimension")
        if self._billing_source == "bigquery":
            # BQ has no hard 2-dimension cap; the CUR is a flat table
            # so any GROUP BY combination is legal. We still cap at
            # 4 to keep the result-row count bounded — beyond that
            # the caller almost certainly wants list_cost_facts.
            if len(group_by) > 4:
                raise ValueError(
                    "AWS billing_by_dimension (BQ): cap of 4 GroupBy "
                    "dimensions — for finer-grained slicing use "
                    "list_cost_facts",
                )
            return await asyncio.to_thread(
                self._list_billing_by_dimension_bq_sync,
                start_date, end_date, group_by, scope,
                filters or {}, bq_billing,
            )
        if not _BOTO3_AVAILABLE:
            # Symmetric with ``list_billing_daily`` and
            # ``list_cost_facts`` which raise on missing boto3 — those
            # contracts treat missing-dep as a deployment defect, and
            # this dimension-query path was the holdout. Returning []
            # silently masked the missing dependency from the dashboard
            # / MCP caller as "no spend on this dimension".
            raise RuntimeError(
                "AWS list_billing_by_dimension: boto3 not importable — "
                "deployment defect. Install boto3 in the worker image.",
            )
        if len(group_by) > 2:
            # Cost Explorer hard cap. We could simulate higher
            # cardinality client-side, but the cost in API calls
            # multiplies — keep the contract honest.
            raise ValueError(
                "AWS Cost Explorer supports at most 2 GroupBy "
                "dimensions per query",
            )
        return await asyncio.to_thread(
            self._list_billing_by_dimension_sync,
            start_date, end_date, group_by, scope, filters or {},
        )

    def _ce_group_spec(self, dim: str) -> dict[str, str]:
        """Translate a Helix canonical dimension name to a CE GroupBy
        spec. ``tag:KEY`` becomes ``{Type: TAG, Key: KEY}``.

        Reject ``aws:PrincipalTag/<Key>`` explicitly — that namespace
        looks valid (AWS docs use it heavily in IAM policy ``Condition``
        block examples) but Cost Explorer ``GroupBy[Type=TAG]`` silently
        returns empty values for every row, collapsing all spend into
        a single bucket. The correct cost-allocation namespace for IAM
        principal tags is ``iamPrincipal/<Key>`` (no ``aws:`` prefix) —
        admin sees this string in Billing Console → Cost allocation
        tags once activated. Surfacing as 400 with a hint is safer than
        rendering a misleading "0 spend" chart on the admin's screen.
        Same root cause as the 2026-05-12 Bedrock production audit.
        """
        if dim.startswith("tag:"):
            tag_key = dim.split(":", 1)[1].strip()
            if not tag_key:
                raise ValueError(
                    "tag dimension requires a key (e.g. 'tag:Project')",
                )
            if tag_key.startswith("aws:PrincipalTag/"):
                suggested = tag_key.replace(
                    "aws:PrincipalTag/", "iamPrincipal/", 1,
                )
                raise ValueError(
                    f"tag key {tag_key!r} is the IAM-policy evaluator's "
                    "namespace and is silently rejected by Cost Explorer "
                    "GroupBy (every row returns an empty value, hiding "
                    "your spend). For IAM principal-tag cost allocation, "
                    f"use {suggested!r} instead (no ``aws:`` prefix) — "
                    "that's the literal key Billing Console surfaces "
                    "once the cost-allocation tag is activated.",
                )
            return {"Type": "TAG", "Key": tag_key}
        ce_dim = self._CE_DIMENSION_ALIASES.get(dim.lower())
        if ce_dim is None:
            raise ValueError(
                f"AWS does not support dimension {dim!r} "
                f"(supported: {sorted(self._CE_DIMENSION_ALIASES)} + tag:KEY)",
            )
        return {"Type": "DIMENSION", "Key": ce_dim}

    def _ce_filter_expr(
        self,
        scope: CloudScopeRef | None,
        filters: dict[str, list[str]],
    ) -> dict[str, Any] | None:
        """Build the CE ``Filter`` expression by AND-ing scope +
        per-dimension allow lists. ``filters`` keys mirror the
        ``group_by`` aliases so the caller can pass
        ``{"service": ["EC2 - Other"]}``.
        """
        clauses: list[dict[str, Any]] = []
        if scope is not None and scope.external_id:
            # New scope rows are always backed by a 12-digit AWS
            # account id since ``_own_account_id`` now raises on STS
            # failure rather than substituting a UUID. This guard
            # remains for read-side safety: any legacy ghost scope
            # row that survived the cleanup window would otherwise
            # produce a CE filter mismatch and kill billing queries.
            ext = str(scope.external_id)
            if ext.isdigit() and len(ext) == 12:
                clauses.append({"Dimensions": {
                    "Key": "LINKED_ACCOUNT", "Values": [ext],
                }})
        for dim, values in (filters or {}).items():
            if not values:
                continue
            if dim.startswith("tag:"):
                tag_key = dim.split(":", 1)[1].strip()
                if tag_key:
                    clauses.append({"Tags": {
                        "Key": tag_key, "Values": list(values),
                    }})
                continue
            ce_dim = self._CE_DIMENSION_ALIASES.get(dim.lower())
            if ce_dim is None:
                raise ValueError(
                    f"AWS filter dimension {dim!r} not supported",
                )
            clauses.append({"Dimensions": {
                "Key": ce_dim, "Values": list(values),
            }})
        if not clauses:
            return None
        if len(clauses) == 1:
            return clauses[0]
        return {"And": clauses}

    def _list_billing_by_dimension_sync(
        self,
        start: date,
        end: date,
        group_by: list[str],
        scope: CloudScopeRef | None,
        filters: dict[str, list[str]],
    ) -> list[BillingByDimensionRow]:
        ce = self._boto_client("ce", region="us-east-1")
        ce_groups = [self._ce_group_spec(d) for d in group_by]
        ce_filter = self._ce_filter_expr(scope, filters)
        kwargs: dict[str, Any] = dict(
            TimePeriod={
                "Start": start.isoformat(),
                "End": (end + timedelta(days=1)).isoformat(),
            },
            Granularity="DAILY",
            Metrics=["UnblendedCost"],
            GroupBy=ce_groups,
        )
        if ce_filter:
            kwargs["Filter"] = ce_filter
        out: list[BillingByDimensionRow] = []
        try:
            for page in _ce_paged(ce, **kwargs):
                for bucket in page.get("ResultsByTime") or []:
                    d = date.fromisoformat(bucket["TimePeriod"]["Start"])
                    for grp in bucket.get("Groups") or []:
                        keys = grp.get("Keys") or []
                        # Re-attach the canonical alias names so the
                        # API caller sees back what they asked for.
                        # CE returns tag values as ``"key$value"`` —
                        # strip the prefix when we know it's a tag.
                        dims_out: dict[str, str] = {}
                        for alias, raw_key in zip(group_by, keys, strict=False):
                            if alias.startswith("tag:") and "$" in raw_key:
                                raw_key = raw_key.split("$", 1)[1]
                            dims_out[alias] = raw_key
                        amount = float(grp["Metrics"]["UnblendedCost"]["Amount"])
                        out.append(BillingByDimensionRow(
                            date=d,
                            dimensions=dims_out,
                            cost=amount,
                            currency=grp["Metrics"]["UnblendedCost"].get(
                                "Unit", "USD",
                            ),
                            raw={"ce_group": grp},
                        ))
        except (ClientError, BotoCoreError) as exc:
            logger.warning(
                "Cost Explorer dimension query failed (%s, group_by=%s): %s",
                self.integration_id, group_by, exc,
            )
            raise
        return out

    # ── Custom-dimension billing query (BigQuery CUR) ───────────
    #
    # Mirrors the CE codepath's contract: per-day rows whose
    # ``dimensions`` dict echoes the canonical aliases passed in
    # ``group_by``, with optional dimension filters AND-ed in.
    # Differences from CE:
    #   * No 2-dimension cap (CE limit, not CUR).
    #   * ``tag:KEY`` resolves to ``resource_tags_user_<lower(key)>``
    #     — supported as long as the tag is exported by the CUR.
    #   * Tenant-wide (``scope=None``) UNION-ALLs every CUR table
    #     in the dataset.
    #
    # **Schema heterogeneity (post 2026-05-15 hotfix)**: per-account
    # CUR Parquet tables vary in which ``resource_tags_user_*``
    # columns they carry — a tag key never set on any resource in
    # that account simply has no column. When a requested tag
    # column (DIM or FILTER) is absent from a table, the table is
    # dropped from the UNION via a bulk INFORMATION_SCHEMA.COLUMNS
    # probe before the data SELECT runs. **Totals divergence
    # caveat for the dashboard team**: in tenant-wide mode this
    # means an account with a missing tag-DIM column contributes
    # NOTHING to the breakdown (its non-tag-related cost also
    # vanishes from the result). The tradeoff is intentional —
    # substituting NULL would silently merge "tag unobservable"
    # rows with "tag legitimately empty" rows, which is worse for
    # cost-attribution honesty. Daily totals from
    # ``_list_billing_daily_bq_sync`` will exceed the by-dimension
    # SUM for any tenant with mixed CUR tag schemas. Surfaced
    # explicitly in PLATFORM.md so /cmdb/cost/breakdown consumers
    # know to expect this.
    def _list_billing_by_dimension_bq_sync(
        self,
        start: date,
        end: date,
        group_by: list[str],
        scope: CloudScopeRef | None,
        filters: dict[str, list[str]],
        bq_billing: dict[str, Any] | None,
    ) -> list[BillingByDimensionRow]:
        from google.cloud import bigquery  # type: ignore

        # Validate per-scope external_id BEFORE allocating the BQ
        # client — parity with ``_list_cost_facts_bq_sync``. Round-3
        # review consistency fix: a non-12-digit scope should fail
        # loudly without paying the SA-credential-parse cost (which
        # produces a confusing MalformedError on a test/dev fixture).
        if scope is not None and scope.external_id:
            ext_validate = str(scope.external_id)
            if not (ext_validate.isdigit() and len(ext_validate) == 12):
                raise RuntimeError(
                    f"AWS billing_by_dimension (BQ): scope "
                    f"external_id {ext_validate!r} is not a 12-digit "
                    "AWS account id",
                )

        (
            creds, dataset, pattern, layout, tag_schema, fact_grain,
        ) = self._bq_resolve_options(bq_billing)

        # Resolve every aliased dimension into ``(alias, sql_expr)`` —
        # sql_expr is the column reference (CUR 1.0) or correlated
        # subquery (CUR 2.0) inserted into SELECT / GROUP BY. Tag
        # dimensions resolve via ``_bq_tag_expression``; non-tag
        # dimensions via ``_bq_dimension_expression``. Raise on
        # unsupported alias so the API can return 400.
        dim_exprs: list[tuple[str, str]] = []
        for alias in group_by:
            if alias.startswith("tag:"):
                tag_key = alias.split(":", 1)[1]
                dim_exprs.append(
                    (alias, self._bq_tag_expression(tag_schema, tag_key)),
                )
                continue
            expr = self._bq_dimension_expression(tag_schema, alias)
            if expr is None:
                raise ValueError(
                    f"AWS does not support dimension {alias!r} in BQ "
                    f"mode (supported: "
                    f"{sorted(self._BQ_DIMENSION_COLUMN_MAP)} + tag:KEY)",
                )
            dim_exprs.append((alias, expr))

        client = self._bq_client(creds)
        try:
            partition_pred, partition_params = self._bq_partition_predicate(
                layout, start, end,
            )

            # Build WHERE filters: dimension allow-lists AND tag-value
            # allow-lists. Each becomes a ``col IN UNNEST(@param)``
            # clause. Empty value lists are silently skipped so callers
            # can pass partially-populated filters without surprises.
            params: list[Any] = [
                bigquery.ScalarQueryParameter("start", "DATE", start),
                bigquery.ScalarQueryParameter("end", "DATE", end),
            ] + partition_params
            filter_clauses: list[str] = []
            for fkey, fvalues in (filters or {}).items():
                if not fvalues:
                    continue
                if fkey.startswith("tag:"):
                    col_expr = self._bq_tag_expression(
                        tag_schema, fkey.split(":", 1)[1],
                    )
                else:
                    col_expr = self._bq_dimension_expression(tag_schema, fkey)
                    if col_expr is None:
                        raise ValueError(
                            f"AWS filter dimension {fkey!r} not supported",
                        )
                pname = f"flt_{len(filter_clauses)}"
                filter_clauses.append(f"{col_expr} IN UNNEST(@{pname})")
                params.append(bigquery.ArrayQueryParameter(
                    pname, "STRING", [str(v) for v in fvalues],
                ))

            # Resolve target tables — layout-aware (per_account routes
            # by table name, per_month enumerates month-tables in range
            # and adds an account-id WHERE filter for scoped calls).
            picks = self._bq_resolve_tables(
                client, layout=layout, dataset=dataset, pattern=pattern,
                scope=scope, start=start, end=end,
            )
            if not picks:
                return []

            account_filter = picks[0].account_filter
            account_filter_clause = (
                " AND line_item_usage_account_id = @scope_acct"
                if account_filter is not None else ""
            )
            if account_filter is not None:
                params.append(bigquery.ScalarQueryParameter(
                    "scope_acct", "STRING", account_filter,
                ))

            # Schema-aware tag handling — only applies to CUR 1.0
            # (flat). Per-account CUR Parquet tables vary in which
            # ``resource_tags_user_*`` columns they carry (a tag key
            # never set on any resource in an account simply has no
            # column there). When ANY requested tag column (dimension
            # OR filter) is absent from a per-account table, drop that
            # table from the UNION entirely. The alternative —
            # substituting ``CAST(NULL AS STRING)`` for absent dim
            # columns — would silently merge rows whose owner tag was
            # *unobservable* (column missing) with rows whose owner tag
            # was *legitimately NULL* (column present, value not set).
            # Those two states must not collapse on a cost-attribution
            # dashboard. Round-1 review finding. Same incident family
            # as cost_facts_sync 2026-05-15.
            #
            # CUR 2.0 (nested ``resource_tags.key_value``) has uniform
            # schema — the array is present on every row in every
            # per-month table; a tag key never set on a row manifests
            # as NULL from the correlated subquery. The "column missing
            # vs value NULL" distinction the flat path defends against
            # is structurally unavailable in CUR 2.0: both states fold
            # to the same NULL, an unavoidable consequence of the
            # nested schema documented in
            # ``docs/ops/cloud-lifecycle-ops.md``.
            if tag_schema == "flat":
                tag_cols_needed: set[str] = set()
                for _alias, expr in dim_exprs:
                    if expr.startswith("resource_tags_user_"):
                        tag_cols_needed.add(expr)
                for fkey in (filters or {}).keys():
                    if fkey.startswith("tag:") and (filters or {}).get(fkey):
                        tag_cols_needed.add(
                            self._bq_tag_column(fkey.split(":", 1)[1]),
                        )
                if tag_cols_needed:
                    all_tables = [p.fq_table for p in picks]
                    table_columns_map = self._bq_resolve_table_columns(
                        client, dataset=dataset,
                        table_names=all_tables,
                        columns_of_interest=tag_cols_needed,
                    )
                    picks = [
                        p for p in picks
                        if tag_cols_needed.issubset(
                            table_columns_map.get(p.fq_table, set()),
                        )
                    ]
                    if not picks:
                        return []

            group_cols = ", ".join(f"d{i}" for i in range(len(dim_exprs)))

            where_extra = (
                " AND " + " AND ".join(filter_clauses)
            ) if filter_clauses else ""

            # Every table that reaches ``_table_select`` carries all
            # requested tag columns (for ``flat``, the schema-drop step
            # above is the gatekeeper; for ``nested``, the nested
            # array column lives on every row). Non-tag dimensions
            # reference core CUR columns from
            # ``_BQ_DIMENSION_COLUMN_MAP`` (CUR 1.0) or their nested
            # ``product.key_value`` equivalents (CUR 2.0) — both
            # universal across the same-version CUR exports.
            select_cols = ",\n              ".join(
                f"{expr} AS d{i}" for i, (_, expr) in enumerate(dim_exprs)
            )

            def _table_select(pick: _BqTablePick) -> str:
                # SP rows are kept (see SP semantics block at top of class).
                # For ``service``-keyed by-dimension queries the SP rows
                # land under ``product_product_name='Savings Plans for
                # AWS Compute usage'``; callers comparing totals against
                # AWS Cost Explorer's "Group by service" view will see the
                # same line item and matching sign.
                return f"""
                SELECT
                  DATE(line_item_usage_start_date) AS day,
                  {select_cols},
                  line_item_currency_code AS currency,
                  SUM(line_item_unblended_cost) AS partial_cost
                FROM `{pick.fq_table}`
                WHERE {partition_pred}
                  AND DATE(line_item_usage_start_date) BETWEEN @start AND @end
                  {account_filter_clause}
                  {where_extra}
                GROUP BY day, {group_cols}, currency
                """

            # Wrap the per-table UNION in an outer aggregator so a
            # dimension value appearing in multiple linked accounts
            # (rare for ``service``, common for ``region``) collapses
            # to one row. Currency stays in GROUP BY so an org
            # transitioning between USD and CNY billing can keep its
            # rows atomic. The inner alias is ``partial_cost`` (NOT
            # ``native_cost``) so the outer ``SUM(...) AS native_cost``
            # doesn't collide with its own alias — BigQuery rejects
            # "aggregations of aggregations" when the names match.
            inner = " UNION ALL ".join(_table_select(p) for p in picks)
            sql = f"""
            SELECT day, {group_cols}, currency,
                   SUM(partial_cost) AS native_cost
            FROM (
            {inner}
            )
            GROUP BY day, {group_cols}, currency
            HAVING ABS(SUM(partial_cost)) > 0
            ORDER BY day ASC
            """
            job = client.query(
                sql, job_config=self._bq_query_job_config(params),
            )

            out: list[BillingByDimensionRow] = []
            for row in job.result():
                dims_out: dict[str, str] = {}
                for i, (alias, _expr) in enumerate(dim_exprs):
                    v = getattr(row, f"d{i}", None)
                    dims_out[alias] = "" if v is None else str(v)
                out.append(BillingByDimensionRow(
                    date=row.day,
                    dimensions=dims_out,
                    cost=float(row.native_cost or 0),
                    currency=str(row.currency or "USD"),
                    raw={"bq_source": "cur"},
                ))
            return out
        finally:
            # Close the BQ client deterministically — see
            # ``_list_billing_daily_bq_sync`` for the FD-leak rationale.
            try:
                client.close()
            except Exception:  # noqa: BLE001
                pass

    # ── Resource-creator lookup via CloudTrail ──────────────────
    #
    # CloudTrail LookupEvents API limits:
    #   * 90-day retention on the default management-events trail.
    #     Earlier resources can ONLY be attributed by a long-term
    #     S3-backed Trail (operator must configure separately).
    #   * 2 req/s per region (we use one call per resource, so the
    #     cron sweep throttles to 1 req/s for headroom).
    #   * LookupAttributes allows ONE attribute per request — we
    #     filter by ResourceName (most selective) and post-filter
    #     event names in Python.
    #
    # The ResourceName attribute on LookupEvents requires the
    # *short* resource identifier (not the ARN). For ``i-abc`` /
    # ``vol-abc`` / S3 bucket names this is already short; for
    # service ARNs (``arn:aws:rds:...:db:helix-db``) we extract the
    # trailing segment.

    _CLOUDTRAIL_CREATE_EVENTS = frozenset({
        # EC2 + EBS + networking
        "RunInstances", "CreateVolume", "AllocateAddress",
        "CreateSecurityGroup", "CreateVpc", "CreateSubnet",
        "CreateImage", "CreateSnapshot", "CreateNatGateway",
        "CreateInternetGateway", "CreateRouteTable",
        "CreateLaunchTemplate",
        # Load balancing
        "CreateLoadBalancer",
        # S3
        "CreateBucket",
        # RDS
        "CreateDBInstance", "CreateDBCluster", "CreateDBSnapshot",
        # Lambda
        "CreateFunction", "CreateFunction20150331",
        # DynamoDB
        "CreateTable",
        # ElastiCache
        "CreateCacheCluster", "CreateReplicationGroup",
        # CloudFront
        "CreateDistribution",
        # Route53
        "CreateHostedZone",
        # IAM
        "CreateUser", "CreateRole", "CreatePolicy", "CreateGroup",
        # SNS / SQS
        "CreateTopic", "CreateQueue",
        # KMS / Secrets Manager
        "CreateKey", "CreateSecret",
        # EKS / ECS / ECR / Redshift / DMS
        "CreateCluster", "CreateRepository",
        "CreateReplicationInstance",
        # API Gateway / SageMaker / MSK / Glue / CFN
        "CreateRestApi", "CreateNotebookInstance", "CreateEndpoint",
        "CreateClusterV2", "CreateDatabase", "CreateStack",
    })

    _GLOBAL_AWS_RESOURCE_PREFIXES = (
        "arn:aws:iam:",
        "arn:aws:s3:::",
        "arn:aws:cloudfront:",
        "arn:aws:route53:",
        "arn:aws:waf:",
    )

    @classmethod
    def _cloudtrail_extract_lookup_name(
        cls, resource_id: str,
    ) -> tuple[str, bool] | None:
        """Reduce a Helix-stored resource_id to the
        ``LookupAttributes.ResourceName`` value CloudTrail expects.

        Returns ``(lookup_name, is_global)`` or ``None`` when the
        resource_id is synthetic (``aws-account:`` aggregate / SP
        principal-tag id) and audit-log attribution doesn't apply.
        """
        if not resource_id:
            return None
        # Drop synthetic ids the BQ-CUR ingest path mints for
        # service-level / SP-negation / principal-tag rows — these
        # have no real cloud-side resource to look up.
        if resource_id.startswith(("aws-account:", "aws-principal:")):
            return None

        if resource_id.startswith("arn:aws:"):
            is_global = any(
                resource_id.startswith(p)
                for p in cls._GLOBAL_AWS_RESOURCE_PREFIXES
            )
            if resource_id.startswith("arn:aws:s3:::"):
                return resource_id[len("arn:aws:s3:::"):], True
            parts = resource_id.split(":", 5)
            if len(parts) < 6:
                return None
            tail = parts[5]
            if "/" in tail:
                return tail.rsplit("/", 1)[-1], is_global
            if ":" in tail:
                return tail.rsplit(":", 1)[-1], is_global
            return tail, is_global

        return resource_id, False

    async def get_resource_creator(
        self,
        resource_id: str,
        *,
        resource_type: str | None = None,
        scope: CloudScopeRef | None = None,
    ) -> ResourceCreatorInfo | None:
        """Resolve "who created this resource" via CloudTrail.

        See :meth:`CloudDriver.get_resource_creator` for the
        contract.

        Region selection: tries us-east-1 first (global services
        always land there, and most prod accounts use multi-region
        trails that consolidate to us-east-1). If empty and
        ``scope.extra['region']`` is set, retries in that region.
        """
        from datetime import UTC
        from datetime import datetime as _dt
        from datetime import timedelta as _td

        looked_up = self._cloudtrail_extract_lookup_name(resource_id)
        if looked_up is None:
            return None
        lookup_name, is_global = looked_up

        end_time = _dt.now(UTC)
        start_time = end_time - _td(days=90)

        region_hint: str | None = None
        if scope is not None:
            region_hint = (scope.extra or {}).get("region")
        primary_region = "us-east-1" if is_global else (
            region_hint or "us-east-1"
        )
        secondary_region = (
            None if is_global or primary_region == region_hint
            else region_hint
        )

        result = await asyncio.to_thread(
            self._cloudtrail_lookup_sync,
            lookup_name, primary_region, start_time, end_time,
        )
        if result is None and secondary_region:
            result = await asyncio.to_thread(
                self._cloudtrail_lookup_sync,
                lookup_name, secondary_region,
                start_time, end_time,
            )
        return result

    def _cloudtrail_lookup_sync(
        self,
        lookup_name: str,
        region: str,
        start_time: datetime,
        end_time: datetime,
    ) -> ResourceCreatorInfo | None:
        """Synchronous CloudTrail LookupEvents call.

        Returns the EARLIEST event whose EventName is in
        ``_CLOUDTRAIL_CREATE_EVENTS`` — the original create event
        for the resource within the 90-day window. ``None`` when
        no matching create event was found.

        Raises :class:`ResourceCreatorWindowExceeded` on the
        explicit ``InvalidTimeRangeException``.
        """
        from botocore.exceptions import ClientError

        ct = self._boto_client("cloudtrail", region=region)
        kwargs: dict[str, Any] = {
            "LookupAttributes": [
                {"AttributeKey": "ResourceName", "AttributeValue": lookup_name},
            ],
            "StartTime": start_time,
            "EndTime": end_time,
            "MaxResults": 50,
        }
        earliest: dict[str, Any] | None = None
        earliest_at: datetime | None = None
        pages_walked = 0
        _PAGE_CAP = 5
        try:
            while True:
                resp = ct.lookup_events(**kwargs)
                pages_walked += 1
                for ev in resp.get("Events") or []:
                    event_name = ev.get("EventName") or ""
                    # Prefer explicit allowlist (curated, ~40
                    # services) BUT fall back to prefix match on
                    # Create* / Run* / Allocate* / Provision* to
                    # catch services the allowlist hasn't been
                    # updated for. The post-pick "earliest event"
                    # logic protects against false positives — if
                    # a non-create event sneaks in via prefix
                    # match, the actual create event is still
                    # older and wins.
                    if (
                        event_name not in self._CLOUDTRAIL_CREATE_EVENTS
                        and not event_name.startswith((
                            "Create", "Run", "Allocate", "Provision",
                        ))
                    ):
                        continue
                    ev_time = ev.get("EventTime")
                    if ev_time is None:
                        continue
                    if earliest_at is None or ev_time < earliest_at:
                        earliest = ev
                        earliest_at = ev_time
                next_token = resp.get("NextToken")
                if not next_token or pages_walked >= _PAGE_CAP:
                    break
                kwargs["NextToken"] = next_token
        except ClientError as exc:
            code = exc.response.get("Error", {}).get("Code", "")
            if code == "InvalidTimeRangeException":
                raise ResourceCreatorWindowExceeded(
                    f"CloudTrail {region!r} rejected 90-day window for "
                    f"{lookup_name!r}: {exc}",
                ) from exc
            raise

        if earliest is None:
            return None
        return self._cloudtrail_event_to_creator_info(earliest)

    @classmethod
    def _cloudtrail_event_to_creator_info(
        cls, event: dict[str, Any],
    ) -> ResourceCreatorInfo:
        """Parse one CloudTrail LookupEvents row → ResourceCreatorInfo.

        Pulls the principal from ``CloudTrailEvent`` (the JSON-string
        payload) rather than top-level ``Username`` — the userIdentity
        block inside is canonical; top-level Username is best-effort
        and missing for many federated/SSO sessions.
        """
        import json as _json

        principal_id = ""
        email: str | None = None
        display_name: str | None = None

        raw_event_json = event.get("CloudTrailEvent")
        raw_event: dict[str, Any] = {}
        if isinstance(raw_event_json, str):
            try:
                raw_event = _json.loads(raw_event_json)
            except (ValueError, TypeError):
                raw_event = {}

        user_identity = raw_event.get("userIdentity") or {}
        arn = user_identity.get("arn") or ""
        principal_id = str(arn) if arn else str(event.get("Username") or "")

        # AWS SSO assumed-role ARNs embed the email after the last `/`:
        #   ``arn:aws:sts::<acct>:assumed-role/AWSReservedSSO_X_Y/<email>``
        if "/" in arn:
            tail = arn.rsplit("/", 1)[-1]
            if "@" in tail:
                email = tail
        if email is None:
            extra = raw_event.get("additionalEventData") or {}
            cand = extra.get("principalEmail")
            if isinstance(cand, str) and "@" in cand:
                email = cand

        if email:
            local_part = email.split("@", 1)[0]
            display_name = local_part.replace(".", " ").title() or None

        _RAW_CAP_BYTES = 16 * 1024
        raw_to_store: dict[str, Any] = {
            "event_name": event.get("EventName"),
            "event_source": event.get("EventSource"),
            "event_id": event.get("EventId"),
            "event_time": (
                event["EventTime"].isoformat()
                if event.get("EventTime") else None
            ),
            "username": event.get("Username"),
            "principal_arn": principal_id,
        }
        if user_identity:
            raw_to_store["user_identity"] = {
                k: user_identity.get(k)
                for k in (
                    "type", "arn", "accountId", "userName",
                    "principalId", "sessionContext",
                )
                if user_identity.get(k) is not None
            }
        if len(_json.dumps(raw_to_store, default=str)) > _RAW_CAP_BYTES:
            raw_to_store.pop("user_identity", None)

        return ResourceCreatorInfo(
            principal_id=principal_id or "<unknown>",
            email=email,
            display_name=display_name,
            created_at_cloud=event.get("EventTime"),
            source="cloudtrail",
            raw=raw_to_store,
        )