"""Per-AK product-attributable spend computation.

Replaces the legacy per-applicant ``cost_owner_person_id`` sum (the
panel and budget cron previously summed every fact whose owner was
the AK's applicant) that conflated a single AK's spend with the
applicant's entire scope-owner fallback attribution. PR #251 isolates
each AK to the cost it actually drives, by dispatching the query per
upstream product:

* **Foundry** (Azure OpenAI provider): facts whose
  ``integration_id`` is the AK's cloud-scope integration AND whose
  ``resource_group`` equals the AK's RG (the dedicated RG the
  Azure OpenAI auto-grant minted on its behalf).
* **Bedrock** (AWS BedRock provider): facts whose
  ``integration_id`` is the AK's IAM integration AND whose
  ``service`` matches the Bedrock SKU family AND whose principal-tag
  resource_id encodes the IAM user's ``Owner`` cost-allocation tag
  value (resolved from ``cmdb_iam_users.tags['Owner']`` → assignee
  ``work_email`` → IAM username, in that order; see
  :func:`_resolve_bedrock_owner_tag_value`).
* **Vertex AI** (Google Vertex AI provider): facts whose
  ``integration_id`` is the AK's cloud-scope integration AND whose
  ``service`` matches the Vertex SKU family.
* **Bailian** (Alibaba Cloud Model Studio): per-key spend **is**
  resolvable — BSS ``DescribeInstanceBill`` encodes the API-key id in
  the composite ``InstanceID``, which migration 299 lands in
  ``bailian_usage_daily`` at per-(day × API key × model) grain.
  ``data_available=False`` is returned **only** when the assignment has
  no linked Bailian key at all (caveat ``bailian_no_key``), so the UI
  explains itself instead of showing an erroneous ``$0`` budget bar.
  Keys living in Resource-Directory MEMBER accounts cannot be
  enumerated via ``ListApiKeys``; their rows keep ``api_key_pk=NULL``
  and stay account-attributed. (This paragraph claimed per-key spend
  was impossible until 2026-08-15 — it was superseded by migration 299
  on 2026-06-17 and contradicted by the code 130 lines below.)

Both :meth:`ApiKeyService.get_assignment_spend` (the panel) and
:func:`_compute_assignment_mtd_spend` in ``budget_alerts`` (the cron)
call through here so the displayed bar and the threshold alert agree
to the cent on every AK.

:func:`compute_assignment_spend_detail` extends the same dispatch with
two extra reductions — a 6-month MTD series and a per-``usage_type``
"by model" breakdown for the requested month — used by the per-AK
detail page (``/me/api-keys/:id``). Both reductions reuse the bucket
filter-builders below so a CR3-class drift between panel and detail
("the panel says $X but the model table sums to $Y") is impossible by
construction: same integration, same service whitelist, same extra-
where on every bucket.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Any, NamedTuple
from zoneinfo import ZoneInfo

from sqlalchemy import and_, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.ai_spending.whitelist import load_whitelist
from app.domains.apikey.models import (
    ApiKeyAssignment,
    ApiKeyProvider,
    normalize_provider_match_name,
)
from app.domains.bailian.models import (
    BAILIAN_AGGREGATE_MODEL,
    BailianApiKey,
    BailianUsageDaily,
)
from app.domains.cmdb.models import (
    CmdbBedrockProfileFact,
    CmdbCloudScope,
    CmdbCostFact,
    CmdbIamUser,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

# Every caller derives ``month_start`` / ``month_end`` from Asia/Shanghai
# (``service.py::get_assignment_spend`` / ``get_assignment_spend_detail``,
# ``analytics.py::_resolve_month``), so any "today" this module computes
# must use the same anchor. A UTC one runs up to 8h behind the window:
# between 00:00 and 08:00 CST it names yesterday, and on the 1st of a
# month that is BEFORE ``month_start`` — which silently dropped the whole
# open transfer epoch (the previous holder's spend vanished and reappeared
# 8h later).
_SHANGHAI = ZoneInfo("Asia/Shanghai")


def _today_shanghai() -> date:
    """Today's date in the timezone the spend windows are anchored to."""
    return datetime.now(_SHANGHAI).date()


def _esc_like(p: str) -> str:
    """Escape LIKE metacharacters in a pattern so % and _ match literally."""
    return p.replace("\\", "\\\\").replace("%", r"\%").replace("_", r"\_")


# ── Bucket identifiers ─────────────────────────────────────────────
#
# These are stable strings the wire payload carries in the ``bucket``
# field. The frontend keys i18n labels off them
# (``apikey.spend.bucket.<key>.label``).

BUCKET_FOUNDRY = "foundry"
BUCKET_BEDROCK = "bedrock"
BUCKET_VERTEX = "vertex"
BUCKET_BAILIAN = "bailian"
BUCKET_UNKNOWN = "unknown"


# ── Machine-readable caveat codes ──────────────────────────────────
#
# Every ``data_caveat`` prose string below has a stable code the wire
# payload carries in ``data_caveat_code`` (plus ``data_caveat_params``
# for interpolation). The frontend keys i18n copy off the code
# (``apikey.spend.caveat.<code>``) so a zh-CN user reads the ACTUAL
# reason, not a static per-bucket guess — pre-fix the UI showed
# "iam_user_arn 为空" for a key whose ARN was set but whose spend was
# pinned to a sibling key (#1776), which read as a data bug to the
# key owner. The English prose stays for audit-log / MCP consumers.

CAVEAT_BAILIAN_NO_KEY = "bailian_no_key"
CAVEAT_FOUNDRY_NO_RG = "foundry_no_rg"
CAVEAT_BEDROCK_NO_IAM = "bedrock_no_iam"
CAVEAT_BEDROCK_NO_OWNER_TAG = "bedrock_no_owner_tag"
CAVEAT_BEDROCK_OWNER_UNALLOCATED = "bedrock_owner_unallocated"
CAVEAT_BEDROCK_OWNER_SHARED = "bedrock_owner_shared"
CAVEAT_BEDROCK_TRANSFER_EPOCHS = "bedrock_transfer_epochs"
CAVEAT_BEDROCK_PER_KEY_TAG = "bedrock_per_key_tag"
# Same attribution, but the person also holds a key carrying the
# Owner-tag total these dollars are a subset of — worth saying so the
# reader doesn't add the two numbers together.
CAVEAT_BEDROCK_PER_KEY_TAG_SHARED = "bedrock_per_key_tag_shared"
CAVEAT_VERTEX_NO_SCOPE = "vertex_no_scope"
CAVEAT_WHITELIST_NOT_CONFIGURED = "whitelist_not_configured"
CAVEAT_UNKNOWN_PROVIDER = "unknown_provider"


# ── Service-name substring patterns per bucket (case-insensitive) ──
#
# Resolved from the same ``platform_settings.ai_spending.service_whitelist``
# row the AI Spending dashboard reads (migration 149, Phase 2). Every
# call site — this module, ``ai_spending.service``, MCP tools — routes
# through ``load_whitelist`` so an admin edit on the Settings page
# propagates to BOTH the dashboard's bucket totals AND the per-AK
# budget panel / cron on the next request. If the two ever drift, an
# admin who adds a new AI service sees the dashboard pick it up while
# the per-AK panel under-reports — exactly the silent-disagreement
# class of bug Phase 1 set out to close.
#
# Bucket-to-provider mapping:
#   foundry → whitelist['azure']
#   bedrock → whitelist['aws']
#   vertex  → whitelist['gcp']


# ── Caveats (returned verbatim to the caller; the UI shows them in
#   the data_caveat slot of AssignmentSpendPanel). Bilingual handling
#   is done at the i18n layer — these strings are display-ready in
#   English; the frontend's i18n table maps the bucket key to the
#   matching Chinese label).

# Per-API-key Bailian cost IS resolvable (migration 299, 2026-06-17): BSS
# DescribeInstanceBill's InstanceID composite → per-(day × API key × model)
# rows in ``bailian_usage_daily``, summed by the assignment's linked
# ``bailian_api_keys`` row (see :func:`_spend_bailian`). This caveat is only
# returned when the assignment has no synced Bailian key yet (key sync runs
# nightly; or the key lives in an Aliyun member account Helix can't
# enumerate, where spend stays account-level on the AI Spending dashboard).
_CAVEAT_BAILIAN_NO_KEY = (
    "This Bailian (Model Studio) assignment has no synced API key yet, so "
    "per-key spend can't be attributed. The key sync runs nightly; if the "
    "key lives in an Aliyun member account Helix doesn't manage, its spend "
    "is attributable only at the account level on the AI Spending dashboard."
)

_CAVEAT_FOUNDRY_NO_RG_BINDING = (
    "This Azure OpenAI assignment has no cloud_scope binding — the "
    "auto-grant did not mint a dedicated resource group. Per-AK "
    "spend cannot be resolved; the applicant's overall Foundry "
    "spend is visible on the AI Spending dashboard."
)

_CAVEAT_BEDROCK_NO_IAM_BINDING = (
    "This AWS Bedrock assignment has no IAM user binding (iam_user_arn "
    "is unset) — per-AK cost attribution requires an IAM user with the "
    "Owner cost-allocation tag activated in AWS Billing console."
)

_CAVEAT_BEDROCK_NO_OWNER_TAG = (
    "This AWS Bedrock assignment's IAM user has no resolvable Owner "
    "cost-allocation tag value (neither the IAM-user tag mirror in "
    "cmdb_iam_users nor the assignee's work_email is available). "
    "Re-run the Bedrock auto-grant or backfill via "
    "`scripts/backfill_bedrock_owner_tags.py` so AWS Cost Explorer can "
    "attribute spend to this key."
)

_CAVEAT_BEDROCK_OWNER_UNALLOCATED = (
    "AWS Cost Explorer rows for this account are all tagged "
    "Owner=(unallocated). The most common cause is that the `Owner` "
    "cost-allocation tag has not been activated in AWS Billing Console "
    "→ Cost allocation tags → User-defined. Activation is forward-"
    "looking; historic days remain (unallocated). See "
    "docs/ops/cloud-lifecycle-ops.md §AWS Bedrock cost attribution."
)

def _caveat_bedrock_owner_shared(canonical_ref: str, sibling_count: int) -> str:
    """#1776: this AK shares its ``Owner`` cost-allocation tag with
    ``sibling_count`` other live Bedrock keys held by the same person.

    AWS Cost Explorer attributes Bedrock spend at the ``Owner`` tag (=
    the person's work_email) granularity, NOT per IAM user / per API key,
    so every one of the person's keys resolves to the SAME cost total.
    Reporting that total on each key would N×-inflate the fleet rollup
    and trip false budget breaches, so the total is pinned to a single
    deterministic canonical key (earliest-created) and the siblings show
    no isolated number.
    """
    return (
        f"This person holds {sibling_count + 1} live AWS Bedrock keys that "
        f"share one Owner cost-allocation tag. AWS attributes Bedrock spend "
        f"per person (Owner tag), not per key, so the combined total is "
        f"shown on the earliest-created key ({canonical_ref}); this key's "
        f"isolated spend is not separable. To split spend per key, activate "
        f"a per-assignment cost-allocation tag in AWS Billing (see "
        f"docs/ops/cloud-lifecycle-ops.md §AWS Bedrock cost attribution)."
    )


# ── Bedrock per-Key attribution axis ──────────────────────────────
#
# ``cmdb_bedrock_profile_facts`` breaks ONE owner's Bedrock spend down
# by a second cost-allocation tag. Migration 435 made the axis explicit
# (``profile_tag_key``) so the table can hold both the ``UsedBy``
# workload axis it shipped with AND the per-API-Key axis below.
#
# ``bedrock_autogrant`` has always tagged every IAM user it mints with
# ``helix-request-number=<AK number>``, so once an admin activates that
# key as a cost-allocation tag in AWS Billing and adds it to
# ``cost_facts.aws_profile_tag.<slug>``, Cost Explorer returns spend
# grouped per API Key — the only way to give a key that shares its
# ``Owner`` tag with a sibling (#1776) a real number instead of a
# pinned ``$0``. Both the bare and the ``iamPrincipal/``-prefixed
# spellings are accepted: CE wants the prefix, but an admin who
# configures the bare key shouldn't silently get no attribution.
PER_KEY_PROFILE_AXES = (
    "iamprincipal/helix-request-number",
    "helix-request-number",
)

# Which cost-allocation axis produced the numbers on this page. Bedrock
# only — every other bucket reports ``None``. The detail page needs it
# because the two axes have genuinely different SHAPES, and a reader
# who can't tell them apart reads a structural property as missing data:
#
# * ``owner`` — CE grouped by (SERVICE × ``iamPrincipal/Owner``). CE
#   allows 2 GroupBy slots and both are spent, so there is no meter
#   dimension left: ``cost_facts.usage_type`` is NULL for every row and
#   the model table collapses to one ``(uncategorised)`` line. The
#   workload breakdown IS available here.
# * ``per_key`` — the ``helix-request-number`` axis in
#   ``cmdb_bedrock_profile_facts``, whose model dimension is CE's
#   SERVICE ("Claude Sonnet 4.6 (Amazon Bedrock Edition)"). Real model
#   names, no token-level split, and the workload rows belong to the
#   holder rather than to this key.
AXIS_OWNER = "owner"
AXIS_PER_KEY = "per_key"

# ``usage_unit`` values that mean "this bucket mixes incomparable
# meters", not "this is the unit". AWS Cost Explorer returns the
# literal string ``"N/A"`` on a group whose rows span input-tokens /
# output-tokens / requests. Dividing spend by such a sum yields a
# number with no unit — a rate that looks quotable and isn't.
_MEANINGLESS_USAGE_UNITS = frozenset({"n/a", "na", "none", ""})


def _is_meaningless_unit(unit: str | None) -> bool:
    return (unit or "").strip().lower() in _MEANINGLESS_USAGE_UNITS


def _caveat_bedrock_per_key_tag(canonical_ref: str | None) -> str:
    """Informational (``data_available`` stays True): this key's number
    comes from the per-API-Key cost-allocation tag rather than the
    person-level ``Owner`` tag.

    Worth saying out loud because the same dollars ALSO sit inside the
    Owner total shown on the person's canonical key — the per-key rows
    are a drill-down of that total, not an additional charge. Summing
    the two would double-count.
    """
    scope = (
        f" These dollars are also included in the Owner-tag total shown "
        f"on {canonical_ref} — this is a per-key breakdown of that "
        f"total, not additional spend."
        if canonical_ref else ""
    )
    return (
        "This key's spend is attributed via its own per-API-Key "
        "cost-allocation tag (helix-request-number), so it is isolated "
        "from other Bedrock keys held by the same person."
        f"{scope} Attribution starts from the day the tag was activated "
        "in AWS Billing; earlier days remain aggregated at the Owner "
        "tag. See docs/ops/cloud-lifecycle-ops.md §AWS Bedrock cost "
        "attribution."
    )


def _caveat_bedrock_transfer_epochs(
    from_emails: list[str], until: str,
) -> str:
    """Informational (``data_available`` stays True): this key changed
    hands mid-window, so the month total stitches two AWS Owner-tag
    epochs together.

    AWS Cost Explorer attributes each day's Bedrock spend to the
    ``Owner`` cost-allocation tag value **at usage time** — re-tagging
    the IAM user on transfer is forward-looking only, so the days
    before the transfer stay recorded under the previous holder's tag
    forever. Without stitching, a transferred key's month-to-date
    silently drops everything before the transfer date (the
    AK-20260428-B9C1 case: $1,287 of July spend vanished from the
    panel the moment the transfer completed).
    """
    return (
        f"This key changed owner during the window. Spend through "
        f"{until} is attributed via the previous holder's Owner "
        f"cost-allocation tag ({', '.join(from_emails)}) because AWS "
        f"Cost Explorer keeps historic days under the tag value at "
        f"usage time; days after the transfer use the current "
        f"holder's tag. The month total combines both epochs."
    )


_CAVEAT_VERTEX_NO_SCOPE_BINDING = (
    "This Vertex AI assignment has no cloud_scope binding — per-AK "
    "spend cannot be resolved."
)

_CAVEAT_WHITELIST_NOT_CONFIGURED = (
    "AI service whitelist has no patterns configured for this "
    "provider. Add patterns under AI Spending → Settings → Service "
    "Whitelist for the per-AK spend bar to populate."
)


# ── Return shape ───────────────────────────────────────────────────


def _zero_spend(
    *,
    bucket: str,
    month_start: date,
    month_end: date,
    data_caveat: str | None,
    data_caveat_code: str | None = None,
    data_caveat_params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Empty-spend payload with ``data_available=False``.

    Used for buckets that cannot resolve a per-AK number (Bailian
    always, Foundry without RG binding, etc.). Daily series is
    zero-filled across the month so the sparkline still has a
    domain to draw on — same row count as a populated response.

    This zero-fill is intentional and OK because ``data_available=False``
    short-circuits the rendering in the FE (the panel renders a caveat
    banner instead of the sparkline, see ``AssignmentSpendPanel.tsx``).
    The "trailing zero-fill is misleading" concern that drove the
    Round-32 truncation of ``_sum_with_service_patterns`` does NOT
    apply here — these zeros never reach the user as bar values.
    """
    daily_series: list[dict[str, Any]] = []
    cur = month_start
    while cur <= month_end:
        daily_series.append({"date": cur.isoformat(), "usd": 0.0})
        cur += timedelta(days=1)
    return {
        "bucket": bucket,
        "mtd_usd": Decimal("0"),
        "daily_series": daily_series,
        "last_fact_date": None,
        "data_available": False,
        "data_caveat": data_caveat,
        "data_caveat_code": data_caveat_code,
        "data_caveat_params": data_caveat_params or {},
    }


# ── Bucket classification ──────────────────────────────────────────


def classify_bucket(provider_name: str | None) -> str:
    """Map an :class:`ApiKeyProvider` display name to a bucket id.

    Uses :func:`normalize_provider_match_name` so admin-typed quirks
    (double space, NBSP, etc.) don't break the substring match — the
    same normalisation the cloud auto-grant routers use, so a name
    that minted an RG via autogrant matches the same bucket here.

    Returns ``BUCKET_UNKNOWN`` for unrecognised providers; the caller
    falls through to ``_zero_spend`` with an explanatory caveat so
    the UI surfaces a clear "not classified" state rather than
    silently failing.
    """
    n = normalize_provider_match_name(provider_name)
    if not n:
        return BUCKET_UNKNOWN
    # Foundry: matches "azure openai", "azure  openai", "foundry", etc.
    if "foundry" in n or ("azure" in n and "openai" in n):
        return BUCKET_FOUNDRY
    if "bedrock" in n:
        return BUCKET_BEDROCK
    if "vertex" in n:
        return BUCKET_VERTEX
    # Bailian / Aliyun Model Studio. ``百炼`` is the canonical
    # mainland brand; ``model studio`` is the intl brand the Helix
    # provider catalog uses. ``通义`` is the umbrella Qwen brand and
    # is intentionally NOT a Bailian marker — a future Qwen-direct AK
    # (not via Model Studio) would otherwise be wrongly labelled
    # "Bailian / data not available" when it might have per-key data
    # from a different source. CR2 #8.
    if "bailian" in n or "model studio" in n or "百炼" in n:
        return BUCKET_BAILIAN
    return BUCKET_UNKNOWN


# ── IAM-user parsing for Bedrock attribution ───────────────────────


def parse_iam_user_name(iam_user_arn: str | None) -> str | None:
    """Extract the IAM user name from an ARN.

    AWS ARNs are ``arn:aws:iam::<account>:user/<name>``. Returns
    ``None`` for empty / malformed inputs.

    Today the only caller is :func:`_spend_bedrock`, which uses the
    parsed name to match the AWS cost-explorer principal-tag value
    encoded in ``cost_facts.resource_id``. The parser also handles
    Aliyun RAM ARNs (``acs:ram::*:user/<name>``) with the same
    split-on-``:user/`` rule — that's currently dead code (Bailian
    short-circuits to ``data_available=False`` at the bucket layer
    before any IAM parse happens) but kept symmetric so a future
    per-RAM Bailian sync can call this without re-implementing the
    parser.

    AWS cost-allocation tag values are NOT path-aware — the tag
    stores the short user name, not the path-prefixed form. We slice
    on the LAST segment so an ARN of ``arn:aws:iam::a:user/team/alice``
    returns ``"alice"`` (matching how the cost-tag landed when the
    AWS admin set ``Owner=alice`` on the user).
    """
    if not iam_user_arn:
        return None
    # Anchored shape match so a malformed value like ``garbage:user/x``
    # or a ``role/`` ARN whose path segment happens to contain ``:user/``
    # cannot squeak through. AWS and Aliyun RAM are both accepted; path
    # segments allowed.
    import re

    m = re.match(
        r"^(?:arn:aws:iam::\d+|acs:ram::[\d*]+):user/(?:[^/]+/)*([^/]+)$",
        iam_user_arn,
    )
    if m is None:
        return None
    return m.group(1).strip() or None


# ── Public entry point ─────────────────────────────────────────────


class _BucketFilter:
    """Resolved filter recipe for a bucket-specific cost-facts query.

    Returned by :func:`_resolve_bucket_filter`. Two-state:

    * ``zero_caveat`` is non-None → the bucket cannot resolve a number
      (Bailian / missing IAM / missing RG / whitelist blank). The caller
      returns the standard zero-spend payload tagged with this caveat.
      The other fields are unset in that case.
    * ``zero_caveat`` is None → the filter is buildable. The other
      fields carry the integration id, service-pattern tuple, and the
      bucket-specific extra-where clause to AND into the cost-facts
      query. Both :func:`compute_assignment_spend` (basic panel +
      cron) and :func:`compute_assignment_spend_detail` (detail page)
      route through here so the two surfaces cannot disagree on
      filter shape.
    """
    __slots__ = (
        "bucket",
        "zero_caveat",
        "zero_caveat_code",
        "zero_caveat_params",
        "integration_id",
        "service_patterns",
        "extra_where",
        "owner_tag",
        "owner_epochs",
        "info_caveat",
        "info_caveat_code",
        "info_caveat_params",
        "per_key_profile",
        "deduct_per_key",
        "deduct_per_key_monthly",
    )

    def __init__(
        self,
        *,
        bucket: str,
        zero_caveat: str | None,
        zero_caveat_code: str | None = None,
        zero_caveat_params: dict[str, Any] | None = None,
        integration_id: str | None = None,
        service_patterns: tuple[str, ...] = (),
        extra_where: Any = None,
        owner_tag: str | None = None,
        owner_epochs: list[tuple[str, date | None, date]] | None = None,
        info_caveat: str | None = None,
        info_caveat_code: str | None = None,
        info_caveat_params: dict[str, Any] | None = None,
        per_key_profile: str | None = None,
        deduct_per_key: tuple[Decimal, dict[str, float], list[str]] | None = None,
        deduct_per_key_monthly: dict[str, float] | None = None,
    ) -> None:
        self.bucket = bucket
        self.zero_caveat = zero_caveat
        self.zero_caveat_code = zero_caveat_code
        self.zero_caveat_params = zero_caveat_params or {}
        self.integration_id = integration_id
        self.service_patterns = service_patterns
        self.extra_where = extra_where
        # Bedrock-only extras: the resolved Owner cost-allocation tag
        # value, plus previous-owner tag epochs from completed ownership
        # transfers — ``[(owner_value, start_inclusive, end_inclusive)]``.
        # ``compute_assignment_spend_detail`` reuses both for the
        # by-profile breakdown so it can't drift from the main filter.
        self.owner_tag = owner_tag
        self.owner_epochs = owner_epochs or []
        # Informational caveat attached to an OTHERWISE-resolvable
        # filter (``data_available`` stays True) — e.g. "this month
        # includes pre-transfer spend recorded under the previous
        # holder's Owner tag".
        self.info_caveat = info_caveat
        self.info_caveat_code = info_caveat_code
        self.info_caveat_params = info_caveat_params or {}
        # Bedrock-only: when set, this AK has its own rows on the
        # per-API-Key cost-allocation axis of
        # ``cmdb_bedrock_profile_facts`` (value = the ``profile`` those
        # rows are keyed by, i.e. the AK's request_number). The entry
        # points then read the number from THAT table instead of the
        # Owner-tag cost_facts filter, which is what lets a key pinned
        # to $0 by the shared-Owner rule (#1776) show real money.
        self.per_key_profile = per_key_profile
        self.deduct_per_key = deduct_per_key
        # Per-MONTH sibling totals over the detail page's 6-month bar
        # window — deducted separately because ``monthly_series`` is
        # built after the headline (#2225).
        self.deduct_per_key_monthly = deduct_per_key_monthly


def _owner_prefix_match(value: str):
    """``resource_id LIKE 'aws-principal:Owner=<value>|%'``.

    The ``|`` terminator is what keeps two Owner values mutually
    exclusive; LIKE metacharacters in the email are escaped so a value
    containing ``_`` cannot match a different owner.
    """
    safe = (
        value.replace("\\", "\\\\")
             .replace("%", "\\%")
             .replace("_", "\\_")
    )
    return CmdbCostFact.resource_id.ilike(
        f"aws-principal:Owner={safe}|%", escape="\\",
    )


def _bedrock_owner_where(
    owner_tag: str, epochs: list[tuple[str, date | None, date]],
):
    """Current-owner clause OR one date-bounded clause per prior epoch.

    Built from whichever epoch set was resolved for the window being
    queried — the requested month for the MTD reductions, the full
    6-month span for the monthly series (see
    :func:`_bedrock_series_extra_where`). Reusing the month's clause for
    the series would apply the month's epoch gate to the earlier bars and
    drop the previous holder's spend out of them.

    Both epoch bounds are INCLUSIVE. The hand-over day carries rows under
    the outgoing AND the incoming tag (the retag lands mid-day), and each
    tag's rows belong to the holder that tag names, so the day has to be
    inside both adjacent epochs. Same-owner epochs are pre-merged by
    :func:`_coalesced_epochs`, which is what keeps two inclusive ranges
    for ONE tag value from summing its rows twice.
    """
    clauses: list[Any] = [_owner_prefix_match(owner_tag)]
    for epoch_owner, epoch_start, epoch_end in epochs:
        cond = and_(
            _owner_prefix_match(epoch_owner),
            CmdbCostFact.date <= epoch_end,
        )
        if epoch_start is not None:
            cond = and_(cond, CmdbCostFact.date >= epoch_start)
        clauses.append(cond)
    return or_(*clauses) if len(clauses) > 1 else clauses[0]


async def _resolve_bucket_filter(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    month_start: date,
    month_end: date,
) -> _BucketFilter:
    """Classify the AK and resolve the filter recipe.

    Centralised per-bucket logic that the basic-spend and detail-spend
    entry points share. Mirrors the legacy ``_spend_foundry`` /
    ``_spend_bedrock`` / ``_spend_vertex`` short-circuits but returns
    the recipe instead of running the query, so the caller can build
    additional reductions (by_model, monthly_series) on top of the
    same filter without re-implementing the per-bucket guard logic.
    """
    provider_name: str | None = (
        await session.execute(
            select(ApiKeyProvider.name).where(
                ApiKeyProvider.id == assignment.provider_id,
            )
        )
    ).scalar_one_or_none()
    bucket = classify_bucket(provider_name)

    if bucket == BUCKET_BAILIAN:
        # Bailian doesn't use the cost_facts filter recipe — its source is
        # ``bailian_usage_daily`` keyed by the assignment's linked Bailian
        # API key. Signalled by a sentinel (no caveat, no integration_id);
        # the entry points short-circuit to :func:`_spend_bailian`.
        return _BucketFilter(bucket=bucket, zero_caveat=None)

    if bucket == BUCKET_FOUNDRY:
        if not assignment.cloud_scope_id:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_FOUNDRY_NO_RG_BINDING,
                zero_caveat_code=CAVEAT_FOUNDRY_NO_RG,
            )
        scope = (
            await session.execute(
                select(
                    CmdbCloudScope.integration_id,
                    CmdbCloudScope.external_id,
                ).where(CmdbCloudScope.id == assignment.cloud_scope_id)
            )
        ).one_or_none()
        if scope is None or not scope.external_id:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_FOUNDRY_NO_RG_BINDING,
                zero_caveat_code=CAVEAT_FOUNDRY_NO_RG,
            )
        patterns = await _whitelist_patterns_for(session, "azure")
        if not patterns:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_WHITELIST_NOT_CONFIGURED,
                zero_caveat_code=CAVEAT_WHITELIST_NOT_CONFIGURED,
            )
        return _BucketFilter(
            bucket=bucket,
            zero_caveat=None,
            integration_id=str(scope.integration_id),
            service_patterns=tuple(patterns),
            extra_where=(CmdbCostFact.resource_group == scope.external_id),
        )

    if bucket == BUCKET_BEDROCK:
        if not assignment.iam_integration_id:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_BEDROCK_NO_IAM_BINDING,
                zero_caveat_code=CAVEAT_BEDROCK_NO_IAM,
            )
        owner_tag = await _resolve_bedrock_owner_tag_value(session, assignment)
        if owner_tag is None:
            if not assignment.iam_user_arn and not assignment.iam_user_id:
                caveat = _CAVEAT_BEDROCK_NO_IAM_BINDING
                caveat_code = CAVEAT_BEDROCK_NO_IAM
            else:
                caveat = _CAVEAT_BEDROCK_NO_OWNER_TAG
                caveat_code = CAVEAT_BEDROCK_NO_OWNER_TAG
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=caveat,
                zero_caveat_code=caveat_code,
            )
        patterns = await _whitelist_patterns_for(session, "aws")
        if not patterns:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_WHITELIST_NOT_CONFIGURED,
                zero_caveat_code=CAVEAT_WHITELIST_NOT_CONFIGURED,
            )
        # One cohort scan for everything below (#2239): the canonical
        # pin, the per-key axis probe and the sibling deduction all used
        # to walk the same key set — twice, each with a per-candidate
        # owner-tag await and a per-candidate profile probe. The fleet
        # rollup computes spend per key, so that was O(K×N) serial
        # queries per integration, doubled.
        cohort = await _bedrock_cohort(
            session,
            assignment=assignment,
            integration_id=str(assignment.iam_integration_id),
            month_start=month_start,
            month_end=month_end,
        )
        # Per-API-Key axis FIRST: when this AK has its own rows on the
        # per-key cost-allocation tag, its spend is genuinely isolated
        # and none of the shared-Owner reasoning below applies. Checked
        # before the canonical pin so a sibling that used to render a
        # pinned "$0.00" starts showing real money the moment the tag is
        # active — that pinned zero (AK-20260710-B5F7, 2026-07-29) is
        # exactly what read as "Helix lost my spend" to the key holder.
        per_key = next(
            (
                pk for cand, _owner, pk in cohort
                if cand.id == assignment.id and pk
            ),
            None,
        )
        if per_key is not None:
            _, _, canonical_ref = await _bedrock_owner_canonical(
                session,
                assignment=assignment,
                owner_tag=owner_tag,
                month_start=month_start,
                month_end=month_end,
                cohort=cohort,
            )
            # The per-key HEADLINE never reads these — it sums
            # ``cmdb_bedrock_profile_facts`` through ``_per_key_where``.
            # They exist so the detail page can still render the HOLDER's
            # workload (``UsedBy``) breakdown next to a per-key number:
            # that breakdown is an Owner-tag reduction, so it needs the
            # same transfer-epoch stitch and the same cost-facts owner
            # predicate the canonical key's page uses, or a key that
            # changed hands mid-window would show a workload table
            # covering only the days since the hand-over.
            per_key_epochs = await _bedrock_owner_epochs(
                session,
                assignment=assignment,
                current_owner_tag=owner_tag,
                window_start=month_start,
                window_end=month_end,
                eligibility_month_start=month_start,
            )
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=None,
                integration_id=str(assignment.iam_integration_id),
                service_patterns=tuple(patterns),
                extra_where=_bedrock_owner_where(owner_tag, per_key_epochs),
                owner_tag=owner_tag,
                owner_epochs=per_key_epochs,
                per_key_profile=per_key,
                info_caveat=_caveat_bedrock_per_key_tag(canonical_ref),
                info_caveat_code=(
                    CAVEAT_BEDROCK_PER_KEY_TAG_SHARED if canonical_ref
                    else CAVEAT_BEDROCK_PER_KEY_TAG
                ),
                info_caveat_params={
                    "key_ref": per_key,
                    "canonical_ref": canonical_ref,
                },
            )
        # #1776: AWS attributes Bedrock cost at the Owner tag (person),
        # not per key. When this person holds >1 live Bedrock key, pin the
        # shared Owner total to a single canonical key so it isn't counted
        # once per key (fleet-rollup inflation + false budget breaches).
        is_canonical, sibling_count, canonical_ref = (
            await _bedrock_owner_canonical(
                session,
                assignment=assignment,
                owner_tag=owner_tag,
                month_start=month_start,
                month_end=month_end,
                cohort=cohort,
            )
        )
        if not is_canonical:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_caveat_bedrock_owner_shared(
                    canonical_ref or "the earliest-created key",
                    sibling_count,
                ),
                zero_caveat_code=CAVEAT_BEDROCK_OWNER_SHARED,
                zero_caveat_params={
                    "canonical_ref": canonical_ref,
                    "sibling_count": sibling_count,
                    # Lets the UI link straight to where the money is
                    # instead of leaving the reader on a $0 dead end.
                    "canonical_assignment_id": (
                        await _bedrock_canonical_assignment_id(
                            session,
                            assignment=assignment,
                            canonical_ref=canonical_ref,
                        )
                    ),
                },
            )

        # Ownership-transfer epochs: AWS keeps historic days under the
        # Owner-tag value at usage time, so a mid-window transfer splits
        # this key's spend across two tag values. Stitch the previous
        # holders' epochs (bounded by their transfer-completion dates)
        # into the filter so the month total doesn't silently drop
        # everything before the hand-over. Epochs that don't overlap the
        # queried window are dropped up-front — a window entirely after
        # the last hand-over behaves exactly like a never-transferred
        # key (no dead OR-clauses, no permanent transfer caveat).
        epochs = await _bedrock_owner_epochs(
            session,
            assignment=assignment,
            current_owner_tag=owner_tag,
            window_start=month_start,
            window_end=month_end,
            eligibility_month_start=month_start,
        )

        info_caveat = None
        info_caveat_code = None
        info_caveat_params: dict[str, Any] | None = None
        if epochs:
            from_emails = sorted({e[0] for e in epochs})
            until = max(e[2] for e in epochs).isoformat()
            info_caveat = _caveat_bedrock_transfer_epochs(from_emails, until)
            info_caveat_code = CAVEAT_BEDROCK_TRANSFER_EPOCHS
            info_caveat_params = {"from_emails": from_emails, "until": until}

        # Siblings that report on the per-key axis are INSIDE this
        # Owner-tag total; deduct them so canonical + siblings == the
        # owner total exactly once (#2132).
        deduct_total, deduct_daily, deduct_monthly, deduct_refs = (
            await _bedrock_sibling_per_key_totals(
                session,
                assignment=assignment,
                owner_tag=owner_tag,
                integration_id=str(assignment.iam_integration_id),
                month_start=month_start,
                month_end=month_end,
                cohort=cohort,
            )
        )
        return _BucketFilter(
            bucket=bucket,
            zero_caveat=None,
            integration_id=str(assignment.iam_integration_id),
            service_patterns=tuple(patterns),
            extra_where=_bedrock_owner_where(owner_tag, epochs),
            owner_tag=owner_tag,
            owner_epochs=epochs,
            info_caveat=info_caveat,
            info_caveat_code=info_caveat_code,
            info_caveat_params=info_caveat_params,
            deduct_per_key=(
                (deduct_total, deduct_daily, deduct_refs)
                if deduct_refs else None
            ),
            deduct_per_key_monthly=deduct_monthly if deduct_refs else None,
        )

    if bucket == BUCKET_VERTEX:
        if not assignment.cloud_scope_id:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_VERTEX_NO_SCOPE_BINDING,
                zero_caveat_code=CAVEAT_VERTEX_NO_SCOPE,
            )
        scope = (
            await session.execute(
                select(CmdbCloudScope.integration_id).where(
                    CmdbCloudScope.id == assignment.cloud_scope_id,
                )
            )
        ).one_or_none()
        if scope is None:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_VERTEX_NO_SCOPE_BINDING,
                zero_caveat_code=CAVEAT_VERTEX_NO_SCOPE,
            )
        patterns = await _whitelist_patterns_for(session, "gcp")
        if not patterns:
            return _BucketFilter(
                bucket=bucket,
                zero_caveat=_CAVEAT_WHITELIST_NOT_CONFIGURED,
                zero_caveat_code=CAVEAT_WHITELIST_NOT_CONFIGURED,
            )
        return _BucketFilter(
            bucket=bucket,
            zero_caveat=None,
            integration_id=str(scope.integration_id),
            service_patterns=tuple(patterns),
            extra_where=(
                CmdbCostFact.scope_id == str(assignment.cloud_scope_id)
            ),
        )

    return _BucketFilter(
        bucket=BUCKET_UNKNOWN,
        zero_caveat=(
            "Assignment provider "
            f"{provider_name!r} is not recognised as an AI product "
            "(Foundry / Bedrock / Vertex AI / Bailian). Per-AK spend "
            "is not available."
        ),
        zero_caveat_code=CAVEAT_UNKNOWN_PROVIDER,
    )


async def compute_assignment_spend(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    month_start: date,
    month_end: date,
) -> dict[str, Any]:
    """Return the per-AK spend payload for the given month window.

    Output keys (consumed by both the panel and the budget cron):

    * ``bucket`` — one of ``foundry`` / ``bedrock`` / ``vertex`` /
      ``bailian`` / ``unknown``.
    * ``mtd_usd`` — Decimal sum for the window. Zero when
      ``data_available=False``.
    * ``daily_series`` — list of ``{date, usd}`` rows from
      ``month_start`` through ``last_fact_date`` (NOT zero-filled past
      the last day with data — 2026-05-18 Round-32). Empty list when
      no rows match OR when ``data_available=False`` for buckets that
      use the ``_zero_spend`` placeholder shape. Mid-window gaps still
      zero-fill (rare; chart "holes" would mislead worse).
    * ``last_fact_date`` — ISO date string of the latest contributing
      row, or ``None`` when no rows match.
    * ``data_available`` — True when the bucket query produced an
      authoritative number; False when the bucket has a structural
      data limitation (Bailian via BSS) OR the AK is missing the
      binding required for attribution (Foundry without RG,
      Bedrock without iam_user_arn).
    * ``data_caveat`` — human-readable explanation when
      ``data_available=False``. ``None`` when the data is fully
      resolved.

    The function does NOT load the provider relationship lazily —
    the caller passes an :class:`ApiKeyAssignment` already joined to
    its provider OR the function performs an explicit select for
    the provider name (one extra round-trip; trivial cost).
    """
    # Resolve provider name → bucket. We fetch the provider name
    # explicitly rather than dereferencing ``assignment.provider`` to
    # avoid triggering lazy-load on an unloaded relationship under
    # the async session (would raise MissingGreenlet).
    flt = await _resolve_bucket_filter(
        session, assignment=assignment,
        month_start=month_start, month_end=month_end,
    )
    if flt.zero_caveat is not None:
        return _zero_spend(
            bucket=flt.bucket,
            month_start=month_start,
            month_end=month_end,
            data_caveat=flt.zero_caveat,
            data_caveat_code=flt.zero_caveat_code,
            data_caveat_params=flt.zero_caveat_params,
        )
    if flt.bucket == BUCKET_BAILIAN:
        return await _spend_bailian(
            session, assignment=assignment,
            month_start=month_start, month_end=month_end,
        )
    assert flt.integration_id is not None
    if flt.per_key_profile is not None:
        # Per-API-Key axis: isolated number, no Owner-tag reasoning.
        payload = await _spend_bedrock_per_key(
            session,
            bucket=flt.bucket,
            integration_id=flt.integration_id,
            key_ref=flt.per_key_profile,
            month_start=month_start,
            month_end=month_end,
        )
        _merge_info_caveat(payload, flt)
        return payload
    payload = await _sum_with_service_patterns(
        session,
        bucket=flt.bucket,
        integration_id=flt.integration_id,
        month_start=month_start,
        month_end=month_end,
        service_patterns=flt.service_patterns,
        extra_where=flt.extra_where,
    )
    # Bedrock-only post-check: when the AK matched no facts (mtd_usd
    # == 0) but the same integration has Bedrock-family ``(unallocated)``
    # rows in the same window, surface the actionable caveat. The most
    # common cause is the ``Owner`` cost-allocation tag has not been
    # activated in AWS Billing Console — IAM users carry the tag but
    # Cost Explorer doesn't group by it, so every row lands in the
    # ``(unallocated)`` bucket. Without this caveat the panel renders
    # a green "$0 / under budget" check that misleads the user into
    # thinking they're spending nothing on AI.
    if (
        flt.bucket == BUCKET_BEDROCK
        and payload["data_available"]
        and not payload["mtd_usd"]
    ):
        unalloc_exists = await _bedrock_has_unallocated_facts(
            session,
            integration_id=flt.integration_id,
            month_start=month_start,
            month_end=month_end,
            service_patterns=flt.service_patterns,
        )
        if unalloc_exists:
            payload["data_caveat"] = _CAVEAT_BEDROCK_OWNER_UNALLOCATED
            payload["data_caveat_code"] = CAVEAT_BEDROCK_OWNER_UNALLOCATED
    # Deduct AFTER the post-check, never before (#2132). The check above
    # asks "did this AK match NO cost facts?" — a question about the
    # RAW Owner-tag total. Deducting first makes it fire whenever the
    # canonical key's own spend is zero but a sibling's is not, which is
    # a perfectly normal state, and the caveat it emits tells the user to
    # go activate a cost-allocation tag that is already working.
    _apply_per_key_deduction(payload, flt)
    _merge_info_caveat(payload, flt)
    return payload


def _merge_info_caveat(payload: dict[str, Any], flt: _BucketFilter) -> None:
    """Attach the filter's informational caveat (transfer-epoch note)
    to a ``data_available=True`` payload — unless a more actionable
    caveat (the unallocated post-check) already claimed the slot."""
    if flt.info_caveat is not None and payload["data_caveat"] is None:
        payload["data_caveat"] = flt.info_caveat
        payload["data_caveat_code"] = flt.info_caveat_code
        payload["data_caveat_params"] = flt.info_caveat_params


async def compute_assignment_spend_detail(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    month_start: date,
    month_end: date,
) -> dict[str, Any]:
    """Extended per-AK spend payload for the detail page.

    Returns every field :func:`compute_assignment_spend` returns plus:

    * ``monthly_series`` — list of ``{month: "YYYY-MM", usd: float}``
      covering six months ending at the requested month, oldest first.
      For the requested month, the row is the MTD value (clamped to
      ``today_shanghai`` by the caller via ``month_end``); prior months
      are the full calendar-month total. Buckets with
      ``data_available=False`` return six zero rows so the bar chart
      still has a domain to draw on.
    * ``by_model`` — list of per-``usage_type`` aggregates for the
      requested month, sorted by ``mtd_usd`` desc:
      ``{model, mtd_usd, usage_quantity, usage_unit, effective_unit_price_usd, percent_of_month}``.
      ``effective_unit_price_usd`` is ``mtd_usd / usage_quantity`` —
      a row-effective rate, **not** the official per-token price — and
      is set to ``None`` when quantity is zero / missing. AWS's
      ``usage_unit`` mixes input-tokens / output-tokens / requests
      across rows in the same model family; the table renders the unit
      alongside the price so the reader doesn't blindly compare rates
      across different units. ``percent_of_month`` is each row's share
      of the bucket's monthly total — rounded to one decimal place,
      zero when the monthly total is zero.
    * ``by_profile`` — Bedrock workload (``UsedBy`` tag) breakdown, plus
      two fields that say WHOSE total those rows divide:

      - ``by_profile_scope`` — ``"self"`` (rows sum to this page's
        ``mtd_usd``), or ``None`` when there are no rows. Both the Owner
        axis and — since the per-key × UsedBy cross (migration 452) —
        the per-key axis are ``"self"``: each key now carries its OWN
        workload split rather than the shared person-level one. (The
        legacy ``"holder"`` value — rows dividing the key HOLDER's month,
        which must not be added to the headline — is no longer produced,
        but stays documented for any cached pre-452 payload.)
      - ``by_profile_total_usd`` — the denominator ``percent_of_month``
        was computed against, so the UI can name it instead of leaving
        the reader to assume it was the headline.
    * ``attribution_axis`` — ``"owner"`` / ``"per_key"`` / ``None``
      (see :data:`AXIS_OWNER`). Lets the UI explain the Owner axis's
      single ``(uncategorised)`` model row as a property of the axis
      rather than letting it read as lost data.

    Same filter recipe as :func:`compute_assignment_spend` —
    routed through :func:`_resolve_bucket_filter` — so the panel
    sum, the cron alert, the detail-page MTD, and the model-table
    sum-of-rows all reconcile to the cent. By construction.
    """
    flt = await _resolve_bucket_filter(
        session, assignment=assignment,
        month_start=month_start, month_end=month_end,
    )
    if flt.zero_caveat is not None:
        zero = _zero_spend(
            bucket=flt.bucket,
            month_start=month_start,
            month_end=month_end,
            data_caveat=flt.zero_caveat,
            data_caveat_code=flt.zero_caveat_code,
            data_caveat_params=flt.zero_caveat_params,
        )
        zero["monthly_series"] = _zero_monthly_series(month_end)
        zero["by_model"] = []
        zero["by_profile"] = []
        zero["by_profile_scope"] = None
        zero["by_profile_total_usd"] = None
        zero["attribution_axis"] = None
        return zero

    if flt.bucket == BUCKET_BAILIAN:
        payload = await _spend_bailian(
            session, assignment=assignment,
            month_start=month_start, month_end=month_end,
        )
        key_pks = await _bailian_key_pks(session, assignment)
        if payload["data_available"] and key_pks:
            payload["by_model"] = await _bailian_by_model(
                session, key_pks=key_pks,
                month_start=month_start, month_end=month_end,
                month_total_usd=payload["mtd_usd"],
            )
            payload["monthly_series"] = await _bailian_monthly_series(
                session, key_pks=key_pks, month_end=month_end,
            )
        else:
            payload["by_model"] = []
            payload["monthly_series"] = _zero_monthly_series(month_end)
        payload["by_profile"] = []
        payload["by_profile_scope"] = None
        payload["by_profile_total_usd"] = None
        payload["attribution_axis"] = None
        return payload

    assert flt.integration_id is not None
    if flt.per_key_profile is not None:
        payload = await _spend_bedrock_per_key(
            session,
            bucket=flt.bucket,
            integration_id=flt.integration_id,
            key_ref=flt.per_key_profile,
            month_start=month_start,
            month_end=month_end,
        )
        _merge_info_caveat(payload, flt)
        payload["by_model"] = await _per_key_by_model(
            session,
            integration_id=flt.integration_id,
            key_ref=flt.per_key_profile,
            month_start=month_start,
            month_end=month_end,
            month_total_usd=payload["mtd_usd"],
        )
        payload["monthly_series"] = await _per_key_monthly_series(
            session,
            integration_id=flt.integration_id,
            key_ref=flt.per_key_profile,
            month_end=month_end,
        )
        # This key's OWN workload (``UsedBy``) breakdown — the per-key ×
        # UsedBy cross (migration 452). Before the cross, the only
        # workload table available on this axis was the person-level one
        # (``_aggregate_by_profile`` over the Owner tag), which every key
        # the person holds shares verbatim — so two keys rendered an
        # identical table (the bug xuhuaiyu reported). The crossed rows
        # carry ``sub_profile`` and sum to THIS key's ``mtd_usd``, so the
        # breakdown is now genuinely per-key and scope is ``self``: the
        # shares are of the headline, no double-count caveat needed.
        payload["by_profile"] = await _per_key_by_profile(
            session,
            integration_id=flt.integration_id,
            key_ref=flt.per_key_profile,
            month_start=month_start,
            month_end=month_end,
            month_total_usd=payload["mtd_usd"],
        )
        payload["by_profile_scope"] = (
            "self" if payload["by_profile"] else None
        )
        payload["by_profile_total_usd"] = (
            float(payload["mtd_usd"]) if payload["by_profile"] else None
        )
        payload["attribution_axis"] = AXIS_PER_KEY
        return payload
    payload = await _sum_with_service_patterns(
        session,
        bucket=flt.bucket,
        integration_id=flt.integration_id,
        month_start=month_start,
        month_end=month_end,
        service_patterns=flt.service_patterns,
        extra_where=flt.extra_where,
    )
    if (
        flt.bucket == BUCKET_BEDROCK
        and payload["data_available"]
        and not payload["mtd_usd"]
    ):
        unalloc_exists = await _bedrock_has_unallocated_facts(
            session,
            integration_id=flt.integration_id,
            month_start=month_start,
            month_end=month_end,
            service_patterns=flt.service_patterns,
        )
        if unalloc_exists:
            payload["data_caveat"] = _CAVEAT_BEDROCK_OWNER_UNALLOCATED
            payload["data_caveat_code"] = CAVEAT_BEDROCK_OWNER_UNALLOCATED
    # The OWNER-TAG total, before any sibling deduction. Every breakdown
    # below (by_model, by_profile) is built from the owner-tag rows, so
    # their percentages have to divide THAT number — dividing them by the
    # deducted headline instead makes the shares sum past 100% whenever a
    # sibling is on the per-key axis. Kept as a separate name so the two
    # meanings can never be confused again.
    owner_total_usd = payload["mtd_usd"]
    # Deduct AFTER the post-check — see the twin in
    # ``compute_assignment_spend`` for why the order matters (#2132).
    _apply_per_key_deduction(payload, flt)
    _merge_info_caveat(payload, flt)

    by_model = await _aggregate_by_model(
        session,
        integration_id=flt.integration_id,
        month_start=month_start,
        month_end=month_end,
        service_patterns=flt.service_patterns,
        extra_where=flt.extra_where,
        month_total_usd=owner_total_usd,
    )
    # The series spans six months, so its transfer-epoch stitch must be
    # resolved against ITS window — ``flt.extra_where`` gates epochs on
    # the requested month, which would drop a previous holder's spend out
    # of the earlier bars (a June transfer read in August rendered June
    # as the current holder's share only, silently changing a bar the
    # user had already seen in July). Only the overlap window widens:
    # ``month_start`` still anchors the sibling-liveness scan, or the
    # requested month's bar would disagree with the MTD above it.
    series_extra_where = flt.extra_where
    if flt.bucket == BUCKET_BEDROCK and flt.owner_tag:
        series_extra_where = await _bedrock_series_extra_where(
            session,
            assignment=assignment,
            owner_tag=flt.owner_tag,
            month_start=month_start,
            month_end=month_end,
        )
    monthly_series = await _aggregate_monthly_series(
        session,
        integration_id=flt.integration_id,
        month_end=month_end,
        service_patterns=flt.service_patterns,
        extra_where=series_extra_where,
    )
    payload["by_model"] = by_model
    payload["monthly_series"] = monthly_series
    # Same sibling deduction as the headline and the daily bars (#2225) —
    # applied here because the series is built after the deduction above.
    _apply_per_key_deduction_monthly(payload, flt)

    # Bedrock-only: per-profile (UsedBy tag) breakdown of THIS owner's
    # spend. Reads the isolated cmdb_bedrock_profile_facts table — empty
    # list when not configured / not yet synced, so the UI degrades to
    # "no profile breakdown" rather than erroring. Reuses the filter's
    # resolved owner tag + transfer epochs so the profile shares cover
    # the same rows the MTD covers.
    by_profile: list[dict[str, Any]] = []
    if (
        flt.bucket == BUCKET_BEDROCK
        and payload["data_available"]
        and flt.owner_tag
    ):
        by_profile = await _aggregate_by_profile(
            session,
            integration_id=flt.integration_id,
            owner_email=flt.owner_tag,
            owner_epochs=flt.owner_epochs,
            month_start=month_start,
            month_end=month_end,
            month_total_usd=owner_total_usd,
        )
    payload["by_profile"] = by_profile
    # Owner axis: these rows divide the OWNER-TAG total, which is the
    # headline only when no sibling reports on the per-key axis. When one
    # does, the headline is smaller by ``per_key_deducted_usd`` and this
    # field — not ``mtd_usd`` — is the number the shares add up to.
    payload["by_profile_scope"] = "self" if by_profile else None
    payload["by_profile_total_usd"] = (
        float(owner_total_usd) if by_profile else None
    )
    payload["attribution_axis"] = (
        AXIS_OWNER if flt.bucket == BUCKET_BEDROCK else None
    )
    return payload


def _monthly_series_months(month_end: date) -> list[tuple[int, int]]:
    """The series' six ``(year, month)`` pairs, NEWEST first — the
    requested month plus the five before it."""
    year, month_num = month_end.year, month_end.month
    months: list[tuple[int, int]] = []
    for _ in range(6):
        months.append((year, month_num))
        month_num -= 1
        if month_num == 0:
            month_num = 12
            year -= 1
    return months


def _month_end_for(day: date) -> date:
    """Last day of ``day``'s calendar month.

    Only used when ``_bedrock_owner_canonical`` is called WITHOUT a
    window (tests / future callers): the per-key axis probe needs an end
    bound, and the month containing ``month_start`` is the same window
    canonicity itself is judged in.
    """
    import calendar

    return day.replace(day=calendar.monthrange(day.year, day.month)[1])


def _monthly_series_window_start(month_end: date) -> date:
    """First day of the oldest month the 6-month series covers.

    The series' own query window is ``[this, month_end]`` — wider than
    the requested month, which is why any window-scoped part of the
    filter has to be re-resolved for it.
    """
    oldest_year, oldest_month = _monthly_series_months(month_end)[-1]
    return date(oldest_year, oldest_month, 1)


def _zero_monthly_series(month_end: date) -> list[dict[str, Any]]:
    """Six zero-filled monthly rows ending at ``month_end``'s month,
    oldest first. Used by :func:`compute_assignment_spend_detail` when
    the bucket cannot resolve a per-AK number so the bar chart still
    has a domain to draw on."""
    return [
        {"month": f"{y:04d}-{m:02d}", "usd": 0.0}
        for y, m in reversed(_monthly_series_months(month_end))
    ]


async def _aggregate_by_model(
    session: AsyncSession,
    *,
    integration_id: str,
    month_start: date,
    month_end: date,
    service_patterns: tuple[str, ...],
    extra_where: Any,
    month_total_usd: Decimal,
) -> list[dict[str, Any]]:
    """GROUP BY ``usage_type`` for the requested month, returning rows
    suitable for the detail page's model-breakdown table.

    ``usage_type`` is what carries the model granularity in cost_facts
    today:

    * **Azure / Foundry** — driver writes ``MeterSubCategory`` (or
      ``MeterCategory`` as fallback) into ``usage_type`` (see
      ``azure_cloud.py::_emit_fact_for_amount``). Example values:
      ``"Azure OpenAI - GPT-4o Input Tokens"``,
      ``"Azure OpenAI - GPT-4 Turbo Output Tokens"``. One row per
      (model × input/output × day) lands a single ``usage_type``
      bucket here.
    * **AWS / Bedrock** — driver writes ``line_item_usage_type`` (e.g.
      ``"USE2-Bedrock:Claude-3-5-Sonnet-Inference-Input-Tokens"``,
      ``"USW2-Bedrock:Llama-3-70B-Instruct-Inference-Output-Tokens"``).
      Per-model usage rows land distinct ``usage_type`` keys here even
      when CUR resource_id is the synthetic ``aws-account:...``
      aggregate. This is why per-model breakdown works for Bedrock
      against PingCAP's no-resource-id CUR tables — the model lives
      on a different CUR column than ``line_item_resource_id``.
    * **GCP / Vertex** — driver writes ``service.description``-ish
      meter info into ``usage_type``; the resulting keys are e.g.
      ``"Vertex AI online prediction - Gemini 1.5 Pro Input"``.

    Rows with ``usage_type IS NULL`` are folded into a single
    ``"(uncategorised)"`` bucket — this prevents the table from
    losing rows that the upstream driver couldn't classify, and the
    reader sees explicit accounting for the gap.

    ``usage_quantity`` is summed alongside ``amount_usd``;
    ``usage_unit`` is taken from the bucket's MAX (the canonical
    unit for that meter). When quantity is zero or missing, the
    effective unit price is ``None`` — the UI must render "—"
    rather than a misleading ``$0 / unit`` rate.

    The same ``None`` applies when the bucket's unit is meaningless
    (:func:`_is_meaningless_unit` — AWS returns the literal ``"N/A"``
    for a group spanning input-tokens / output-tokens / requests). The
    Owner-axis ``(uncategorised)`` bucket is exactly that shape: its
    quantity is a sum across incomparable meters, so ``spend ÷ qty``
    is arithmetic without a unit. Production rendered it as
    ``$0.8473`` next to ``2.03k N/A`` — a number a reader can neither
    verify nor compare, sitting in a column headed "effective unit
    price". A dash is the honest cell.
    """
    if not service_patterns:
        return []
    service_filter = or_(
        *(CmdbCostFact.service.ilike(f"%{_esc_like(p)}%", escape="\\") for p in service_patterns),
    )
    base_where = (
        (CmdbCostFact.integration_id == integration_id),
        (CmdbCostFact.date >= month_start),
        (CmdbCostFact.date <= month_end),
        service_filter,
        extra_where,
    )
    # ``func.coalesce(usage_type, "(uncategorised)")`` keeps NULL-meter
    # rows in the breakdown rather than silently dropping them.
    model_label = func.coalesce(CmdbCostFact.usage_type, "(uncategorised)")
    stmt = (
        select(
            model_label.label("model"),
            func.coalesce(func.sum(CmdbCostFact.amount_usd), 0).label("usd"),
            func.coalesce(func.sum(CmdbCostFact.usage_quantity), 0).label("qty"),
            func.max(CmdbCostFact.usage_unit).label("unit"),
        )
        .where(*base_where)
        .group_by(model_label)
    )
    rows = (await session.execute(stmt)).all()
    month_total = float(month_total_usd or 0)
    out: list[dict[str, Any]] = []
    for r in rows:
        usd = float(r.usd or 0)
        qty = float(r.qty or 0)
        unit_price = (
            (usd / qty) if qty > 0 and not _is_meaningless_unit(r.unit)
            else None
        )
        pct = (usd / month_total * 100.0) if month_total > 0 else 0.0
        out.append({
            "model": str(r.model),
            "mtd_usd": round(usd, 6),
            "usage_quantity": round(qty, 6),
            "usage_unit": r.unit,
            # Pricing decimals: AI tokens are sub-cent — at 6 dp a
            # GPT-4o-mini input price ($0.000150 / 1k tokens) still
            # has resolution. Below that we'd need scientific notation.
            "effective_unit_price_usd": (
                round(unit_price, 8) if unit_price is not None else None
            ),
            "percent_of_month": round(pct, 1),
        })
    # Sort by spend desc so the top consumer lands first. Within a
    # tie (e.g. two models with $0 — should never both lead, but it
    # happens during early-month days before AWS posts billing) sort
    # by model name asc for deterministic ordering.
    out.sort(key=lambda r: (-r["mtd_usd"], r["model"]))
    return out


async def _bedrock_per_key_profile(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    integration_id: str,
    month_start: date,
    month_end: date,
) -> str | None:
    """The ``profile`` value identifying this AK on the per-API-Key
    cost-allocation axis, or None when per-key attribution isn't
    available for this AK / window.

    Available means: ``cmdb_bedrock_profile_facts`` actually holds rows
    for (this integration × the per-key axis × this AK's key ref) inside
    the window. Presence of rows — not presence of the setting — is the
    gate, so the moment the AWS-side tag activation starts producing
    data the AK switches over, and a window that predates activation
    keeps falling back to the Owner-tag path instead of rendering a
    misleading ``$0``.

    The key ref is matched case-insensitively against
    ``assignment.request_number`` (what ``bedrock_autogrant`` writes as
    the IAM user's ``helix-request-number`` tag value).
    """
    key_ref = (assignment.request_number or "").strip()
    if not key_ref:
        return None
    hit = (await session.execute(
        select(CmdbBedrockProfileFact.profile)
        .where(
            CmdbBedrockProfileFact.integration_id == integration_id,
            func.lower(CmdbBedrockProfileFact.profile_tag_key).in_(
                PER_KEY_PROFILE_AXES,
            ),
            func.lower(CmdbBedrockProfileFact.profile) == key_ref.lower(),
            CmdbBedrockProfileFact.date >= month_start,
            CmdbBedrockProfileFact.date <= month_end,
        )
        .limit(1)
    )).scalar_one_or_none()
    return key_ref if hit is not None else None


def _per_key_where(
    *, integration_id: str, key_ref: str, month_start: date, month_end: date,
) -> tuple[Any, ...]:
    """Shared WHERE for every per-Key reduction (total / daily / model /
    monthly), so the detail page's tables cannot disagree with its
    headline number — the same by-construction guarantee
    ``_resolve_bucket_filter`` gives the Owner-tag path."""
    return (
        CmdbBedrockProfileFact.integration_id == integration_id,
        func.lower(CmdbBedrockProfileFact.profile_tag_key).in_(
            PER_KEY_PROFILE_AXES,
        ),
        func.lower(CmdbBedrockProfileFact.profile) == key_ref.lower(),
        CmdbBedrockProfileFact.date >= month_start,
        CmdbBedrockProfileFact.date <= month_end,
    )


async def _spend_bedrock_per_key(
    session: AsyncSession,
    *,
    bucket: str,
    integration_id: str,
    key_ref: str,
    month_start: date,
    month_end: date,
) -> dict[str, Any]:
    """Standard spend payload built from the per-API-Key axis rows.

    Same shape (and same trailing-zero-fill discipline) as
    :func:`_sum_with_service_patterns` — the service whitelist is
    already applied at ingest time for this table, so no service filter
    is re-applied here.
    """
    where = _per_key_where(
        integration_id=integration_id, key_ref=key_ref,
        month_start=month_start, month_end=month_end,
    )
    mtd_raw = (await session.execute(
        select(func.coalesce(func.sum(CmdbBedrockProfileFact.amount_usd), 0))
        .where(*where)
    )).scalar_one()

    raw_daily: dict[str, float] = {
        r.d.isoformat(): float(r.u or 0)
        for r in (await session.execute(
            select(
                CmdbBedrockProfileFact.date.label("d"),
                func.coalesce(
                    func.sum(CmdbBedrockProfileFact.amount_usd), 0,
                ).label("u"),
            )
            .where(*where)
            .group_by(CmdbBedrockProfileFact.date)
        )).all()
    }
    last_fact_date = max(raw_daily.keys()) if raw_daily else None
    daily_series: list[dict[str, Any]] = []
    if last_fact_date is not None:
        last_dt = date.fromisoformat(last_fact_date)
        cur = month_start
        while cur <= last_dt:
            iso = cur.isoformat()
            daily_series.append(
                {"date": iso, "usd": round(raw_daily.get(iso, 0.0), 6)},
            )
            cur += timedelta(days=1)

    return {
        "bucket": bucket,
        "mtd_usd": Decimal(str(mtd_raw or 0)),
        "daily_series": daily_series,
        "last_fact_date": last_fact_date,
        "data_available": True,
        "data_caveat": None,
        "data_caveat_code": None,
        "data_caveat_params": {},
    }


async def _per_key_by_model(
    session: AsyncSession,
    *,
    integration_id: str,
    key_ref: str,
    month_start: date,
    month_end: date,
    month_total_usd: Decimal,
) -> list[dict[str, Any]]:
    """Per-model breakdown for a per-Key-attributed AK.

    The profile table's model axis is ``service`` — for Bedrock, Cost
    Explorer's SERVICE dimension IS the model (``"Claude Sonnet 4.6
    (Amazon Bedrock Edition)"``), verified against production rows. It
    carries no ``usage_type``, so unlike
    :func:`_aggregate_by_model` there is no token-level split and no
    quantity to derive an effective unit price from — both are reported
    as ``None`` rather than a fabricated ``$0 / unit``.
    """
    rows = (await session.execute(
        select(
            CmdbBedrockProfileFact.service.label("model"),
            func.coalesce(
                func.sum(CmdbBedrockProfileFact.amount_usd), 0,
            ).label("usd"),
            func.coalesce(
                func.sum(CmdbBedrockProfileFact.usage_quantity), 0,
            ).label("qty"),
            func.max(CmdbBedrockProfileFact.usage_unit).label("unit"),
        )
        .where(*_per_key_where(
            integration_id=integration_id, key_ref=key_ref,
            month_start=month_start, month_end=month_end,
        ))
        .group_by(CmdbBedrockProfileFact.service)
    )).all()
    month_total = float(month_total_usd or 0)
    out: list[dict[str, Any]] = []
    for r in rows:
        usd = float(r.usd or 0)
        qty = float(r.qty or 0)
        unit_price = (usd / qty) if qty > 0 else None
        out.append({
            "model": str(r.model),
            "mtd_usd": round(usd, 6),
            "usage_quantity": round(qty, 6),
            "usage_unit": r.unit,
            "effective_unit_price_usd": (
                round(unit_price, 8) if unit_price is not None else None
            ),
            "percent_of_month": (
                round(usd / month_total * 100.0, 1)
                if month_total > 0 else 0.0
            ),
        })
    out.sort(key=lambda r: (-r["mtd_usd"], r["model"]))
    return out


async def _per_key_monthly_series(
    session: AsyncSession,
    *,
    integration_id: str,
    key_ref: str,
    month_end: date,
) -> list[dict[str, Any]]:
    """Six-month series for a per-Key-attributed AK, oldest first.

    Window semantics come from the SAME helpers
    :func:`_aggregate_monthly_series` uses — prior months are full
    calendar totals, the last row is clamped to ``month_end`` (the
    caller's MTD boundary) — so the per-Key bar chart can't drift a
    month against the Owner-tag one.
    """
    months = list(reversed(_monthly_series_months(month_end)))
    window_start = _monthly_series_window_start(month_end)

    rows = (await session.execute(
        select(
            func.extract("year", CmdbBedrockProfileFact.date).label("y"),
            func.extract("month", CmdbBedrockProfileFact.date).label("m"),
            func.coalesce(
                func.sum(CmdbBedrockProfileFact.amount_usd), 0,
            ).label("usd"),
        )
        .where(*_per_key_where(
            integration_id=integration_id, key_ref=key_ref,
            month_start=window_start, month_end=month_end,
        ))
        .group_by(
            func.extract("year", CmdbBedrockProfileFact.date),
            func.extract("month", CmdbBedrockProfileFact.date),
        )
    )).all()
    totals = {(int(r.y), int(r.m)): float(r.usd or 0) for r in rows}
    return [
        {"month": f"{y:04d}-{m:02d}", "usd": round(totals.get((y, m), 0.0), 6)}
        for y, m in months
    ]


async def _per_key_by_profile(
    session: AsyncSession,
    *,
    integration_id: str,
    key_ref: str,
    month_start: date,
    month_end: date,
    month_total_usd: Decimal,
) -> list[dict[str, Any]]:
    """This key's OWN workload (``UsedBy``) breakdown — the per-API-Key ×
    UsedBy cross (migration 452).

    Reads the ``sub_profile`` dimension of THIS key's per-key rows, so
    unlike :func:`_aggregate_by_profile` (which divides the person-level
    Owner total and is shared across every key the person holds) these
    rows sum to the key's OWN ``mtd_usd``. That is what lets two keys held
    by one person finally show DIFFERENT workload tables — the bug
    xuhuaiyu reported: B5F7 is all ``dev-external`` while B9C1 is
    ``aiops`` / ``clinic`` / …, even though both share the ``Owner`` tag.

    Grouped case-insensitively on ``sub_profile`` (``aiops`` / ``AIOPS``
    are one workload AWS records under two tag casings — same fold
    :func:`_aggregate_by_profile` does). A NULL ``sub_profile`` (a
    legacy pre-452 per-key row that hasn't been re-synced yet) folds
    into ``(unallocated)`` so the shares still reconcile to ``mtd_usd``.
    ``month_total_usd`` is this key's own MTD, so ``percent_of_month`` is
    a true share of the headline (scope ``self``).

    Each workload carries a nested ``by_model`` (the Bedrock service /
    model it spent on). Returns ``[]`` when the key has no per-key rows
    for the window (not yet crossed with UsedBy, or nothing synced) — the
    UI hides the block rather than drawing an empty table.
    """
    subkey = func.coalesce(
        func.lower(CmdbBedrockProfileFact.sub_profile), "(unallocated)",
    )
    sublabel = func.coalesce(
        CmdbBedrockProfileFact.sub_profile, "(unallocated)",
    )
    stmt = (
        select(
            subkey.label("pkey"),
            func.min(sublabel).label("label"),
            CmdbBedrockProfileFact.service.label("service"),
            func.coalesce(
                func.sum(CmdbBedrockProfileFact.amount_usd), 0,
            ).label("usd"),
        )
        .where(*_per_key_where(
            integration_id=integration_id, key_ref=key_ref,
            month_start=month_start, month_end=month_end,
        ))
        .group_by(subkey, CmdbBedrockProfileFact.service)
    )
    rows = (await session.execute(stmt)).all()
    if not rows:
        return []

    month_total = float(month_total_usd or 0)
    # Fold (sub_profile × service) rows into one entry per workload with a
    # nested model list. Keep negatives (CE credits) so the per-workload
    # total reconciles to ``mtd_usd``; skip only exact-zero noise. Same
    # discipline as :func:`_aggregate_by_profile`.
    profiles: dict[str, dict[str, Any]] = {}
    for r in rows:
        usd = float(r.usd or 0)
        if usd == 0:
            continue
        entry = profiles.setdefault(r.pkey, {
            "profile": str(r.label),
            "mtd_usd": 0.0,
            "by_model": [],
        })
        entry["mtd_usd"] += usd
        entry["by_model"].append({"model": str(r.service), "mtd_usd": usd})

    out: list[dict[str, Any]] = []
    for entry in profiles.values():
        ptotal = entry["mtd_usd"]
        models = entry["by_model"]
        for m in models:
            m["percent_of_profile"] = (
                round(m["mtd_usd"] / ptotal * 100.0, 1) if ptotal > 0 else 0.0
            )
            m["mtd_usd"] = round(m["mtd_usd"], 6)
        models.sort(key=lambda m: (-m["mtd_usd"], m["model"]))
        out.append({
            "profile": entry["profile"],
            "mtd_usd": round(ptotal, 6),
            "percent_of_month": (
                round(ptotal / month_total * 100.0, 1)
                if month_total > 0 else 0.0
            ),
            "by_model": models,
        })
    out.sort(key=lambda r: (-r["mtd_usd"], r["profile"]))
    return out


async def _aggregate_by_profile(
    session: AsyncSession,
    *,
    integration_id: str,
    owner_email: str,
    owner_epochs: list[tuple[str, date | None, date]] | None = None,
    month_start: date,
    month_end: date,
    month_total_usd: Decimal,
) -> list[dict[str, Any]]:
    """Per-profile (``UsedBy`` tag) breakdown of one owner's Bedrock
    spend for the requested month, sorted by ``mtd_usd`` desc.

    Reads the isolated ``cmdb_bedrock_profile_facts`` table — NOT
    ``cmdb_cost_facts`` — so it never perturbs the by-model / MTD /
    dashboard sums. Rows are grouped case-insensitively on ``profile``
    (``aiops`` and ``AIOPS`` are one workload that AWS records under
    two tag casings — merging at read keeps the ingest a clean 1:1 with
    CE groups). ``percent_of_month`` uses the AK's ``mtd_usd`` (from
    ``cmdb_cost_facts``) as the denominator so the profile shares agree
    with the model table's shares.

    Each profile carries a nested ``by_model`` (the Bedrock service /
    model the profile's spend went to), so the UI can drill from a
    profile into which models it used.

    Returns ``[]`` when the profile table has no rows for this owner /
    month (not yet synced, or the integration isn't profile-configured)
    — the UI then hides the block rather than showing an empty table.
    """
    pkey = func.lower(CmdbBedrockProfileFact.profile)
    # Case-insensitive owner match — mirrors the ILIKE the ``mtd_usd``
    # filter uses on cmdb_cost_facts' resource_id, and stays correct
    # even if this table ever lands on a *_bin (case-sensitive)
    # collation on TiDB. Previous-holder transfer epochs are OR-ed in
    # with the same date bounds the cost-facts filter carries so the
    # profile shares reconcile with the stitched MTD.
    owner_clauses: list[Any] = [
        func.lower(CmdbBedrockProfileFact.owner_email)
        == owner_email.lower(),
    ]
    for epoch_owner, epoch_start, epoch_end in (owner_epochs or []):
        cond = and_(
            func.lower(CmdbBedrockProfileFact.owner_email)
            == epoch_owner.lower(),
            CmdbBedrockProfileFact.date <= epoch_end,
        )
        if epoch_start is not None:
            # Inclusive on both ends, exactly like
            # :func:`_bedrock_owner_where` — an intermediate holder's
            # hand-over-day rows are theirs, and dropping them here
            # while the MTD counts them would make the profile shares
            # not add up to ``mtd_usd``.
            cond = and_(cond, CmdbBedrockProfileFact.date >= epoch_start)
        owner_clauses.append(cond)
    stmt = (
        select(
            pkey.label("pkey"),
            func.min(CmdbBedrockProfileFact.profile).label("label"),
            CmdbBedrockProfileFact.service.label("service"),
            func.coalesce(
                func.sum(CmdbBedrockProfileFact.amount_usd), 0,
            ).label("usd"),
        )
        .where(
            CmdbBedrockProfileFact.integration_id == integration_id,
            # Workload axes only. The per-API-Key axis (migration 435) is
            # a DIFFERENT full breakdown of the same owner total — letting
            # its rows in here would list every one of the person's API
            # Key numbers as if they were workloads AND double the
            # by-profile sum against ``mtd_usd``.
            ~func.lower(CmdbBedrockProfileFact.profile_tag_key).in_(
                PER_KEY_PROFILE_AXES,
            ),
            or_(*owner_clauses) if len(owner_clauses) > 1
            else owner_clauses[0],
            CmdbBedrockProfileFact.date >= month_start,
            CmdbBedrockProfileFact.date <= month_end,
        )
        .group_by(pkey, CmdbBedrockProfileFact.service)
    )
    rows = (await session.execute(stmt)).all()
    if not rows:
        return []

    month_total = float(month_total_usd or 0)
    # Fold (profile × service) rows up into one entry per profile with a
    # nested model list. Keep negative cells (CE credits/refunds land as
    # negative UnblendedCost; the driver only drops exact-zero amounts and
    # ``mtd_usd`` sums negatives too) so the per-profile total reconciles
    # to ``mtd_usd``; only skip exact-zero noise.
    profiles: dict[str, dict[str, Any]] = {}
    for r in rows:
        usd = float(r.usd or 0)
        if usd == 0:
            continue
        entry = profiles.setdefault(r.pkey, {
            "profile": str(r.label),
            "mtd_usd": 0.0,
            "by_model": [],
        })
        entry["mtd_usd"] += usd
        entry["by_model"].append({"model": str(r.service), "mtd_usd": usd})

    out: list[dict[str, Any]] = []
    for entry in profiles.values():
        ptotal = entry["mtd_usd"]
        models = entry["by_model"]
        for m in models:
            m["percent_of_profile"] = (
                round(m["mtd_usd"] / ptotal * 100.0, 1) if ptotal > 0 else 0.0
            )
            m["mtd_usd"] = round(m["mtd_usd"], 6)
        models.sort(key=lambda m: (-m["mtd_usd"], m["model"]))
        out.append({
            "profile": entry["profile"],
            "mtd_usd": round(ptotal, 6),
            "percent_of_month": (
                round(ptotal / month_total * 100.0, 1)
                if month_total > 0 else 0.0
            ),
            "by_model": models,
        })
    out.sort(key=lambda r: (-r["mtd_usd"], r["profile"]))
    return out


async def _aggregate_monthly_series(
    session: AsyncSession,
    *,
    integration_id: str,
    month_end: date,
    service_patterns: tuple[str, ...],
    extra_where: Any,
) -> list[dict[str, Any]]:
    """Six monthly totals ending at ``month_end``'s month, oldest first.

    Window: the requested month and the five months immediately before
    it. Each row sums ``amount_usd`` across the calendar month of the
    fact's ``date`` column. The requested month's row reflects whatever
    range ``month_end`` covers — for the current month, the caller
    clamps to ``today_shanghai`` so the bar matches the panel's MTD,
    not the future-end-of-month projection.

    ``extra_where`` must be resolved for the SERIES window, not the
    requested month: a window-scoped predicate (Bedrock's transfer-epoch
    stitch) built for the requested month silently re-writes the earlier
    bars. The caller uses :func:`_bedrock_series_extra_where` for that.

    Implementation: one window-wide aggregation over the full
    [start_of_oldest, month_end] range, grouped on the calendar
    (year, month) of each row's ``date`` — avoids the 6× round-trips
    a naive per-month loop would issue.
    """
    if not service_patterns:
        return _zero_monthly_series(month_end)
    months = _monthly_series_months(month_end)
    window_start = _monthly_series_window_start(month_end)

    service_filter = or_(
        *(CmdbCostFact.service.ilike(f"%{_esc_like(p)}%", escape="\\") for p in service_patterns),
    )
    # Group on the (year, month) of ``date``. ``extract`` works on
    # both TiDB / MySQL and pytest's MySQL fixture.
    year_col = func.extract("year", CmdbCostFact.date).label("y")
    month_col = func.extract("month", CmdbCostFact.date).label("m")
    stmt = (
        select(
            year_col,
            month_col,
            func.coalesce(func.sum(CmdbCostFact.amount_usd), 0).label("usd"),
        )
        .where(
            CmdbCostFact.integration_id == integration_id,
            CmdbCostFact.date >= window_start,
            CmdbCostFact.date <= month_end,
            service_filter,
            extra_where,
        )
        .group_by(year_col, month_col)
    )
    rows = (await session.execute(stmt)).all()
    by_month: dict[str, float] = {}
    for r in rows:
        key = f"{int(r.y):04d}-{int(r.m):02d}"
        by_month[key] = float(r.usd or 0)
    out: list[dict[str, Any]] = []
    for y, m in reversed(months):
        key = f"{y:04d}-{m:02d}"
        out.append({"month": key, "usd": round(by_month.get(key, 0.0), 6)})
    return out


async def _bedrock_has_unallocated_facts(
    session: AsyncSession,
    *,
    integration_id: str,
    month_start: date,
    month_end: date,
    service_patterns: tuple[str, ...],
) -> bool:
    """Return True iff the integration has at least one Bedrock-family
    cost fact in the window whose synthetic resource_id is the
    ``(unallocated)`` sentinel the AWS driver writes when AWS Cost
    Explorer returned an empty principal-tag value (aws_cloud.py:2191,
    where ``owner_value or "(unallocated)"`` is the fan-out). Used by
    :func:`_spend_bedrock` to surface a structured admin caveat when
    an AK genuinely has zero spend but the account-wide chargeback
    pipeline is broken (Owner tag not activated in Billing Console).
    """
    if not service_patterns:
        return False
    service_filter = or_(
        *(CmdbCostFact.service.ilike(f"%{_esc_like(p)}%", escape="\\") for p in service_patterns),
    )
    stmt = select(func.count()).select_from(CmdbCostFact).where(
        CmdbCostFact.integration_id == integration_id,
        CmdbCostFact.date >= month_start,
        CmdbCostFact.date <= month_end,
        service_filter,
        CmdbCostFact.resource_id.ilike(
            "aws-principal:Owner=(unallocated)|%",
        ),
    )
    return bool((await session.execute(stmt)).scalar_one())


async def _resolve_bedrock_owner_tag_value(
    session: AsyncSession, assignment: ApiKeyAssignment,
) -> str | None:
    """Return the value AWS Cost Explorer would have stored under the
    ``iamPrincipal/Owner`` cost-allocation tag for this AK's IAM user,
    or None when nothing is resolvable.

    Cascade documented on :func:`_spend_bedrock`. Returns ``None``
    (not the literal ``"(unallocated)"``) — the caller decides whether
    to surface the no-owner caveat or fall through to a different
    matching path. The sentinel string is never returned because
    matching against it would attribute every untagged AWS principal's
    cost to this AK.
    """
    # 1. IAM-user mirror (authoritative — same row the upsert path
    #    populates from AWS).
    if assignment.iam_user_id is not None:
        tags = (await session.execute(
            select(CmdbIamUser.tags).where(
                CmdbIamUser.id == assignment.iam_user_id,
            )
        )).scalar_one_or_none()
        if tags:
            raw_owner = tags.get("Owner") if isinstance(tags, dict) else None
            if isinstance(raw_owner, str):
                cleaned = raw_owner.strip()
                if cleaned and cleaned != "(unallocated)":
                    return cleaned

    # 2. Assignee work_email — what bedrock_autogrant.py:362 writes to
    #    AWS for new grants. Covers AKs whose IAM mirror row is stale
    #    or missing the Owner tag (e.g. mirror written before the
    #    autogrant added the tag step in PR #220).
    if assignment.assigned_to is not None:
        email = (await session.execute(
            select(IdentityPerson.work_email).where(
                IdentityPerson.id == assignment.assigned_to,
            )
        )).scalar_one_or_none()
        if email:
            cleaned = email.strip()
            if cleaned:
                return cleaned

    # 3. Last-resort legacy fallback: the IAM username itself.
    #    Pre-PR #220 rows tagged ``Owner=<username>`` (or no Owner tag
    #    at all and the admin tagged manually with the IAM username
    #    as a convenience). Returning the username here matches that
    #    historic shape; rows with no Owner tag at all will still
    #    miss (correctly — they're (unallocated) in Cost Explorer).
    return parse_iam_user_name(assignment.iam_user_arn)


# Spend-eligible statuses for the month — mirrors analytics.py's
# ``spend_universe`` membership (live keys + keys revoked within the
# window still carry the month's spend). Kept local to avoid a
# spend→analytics import cycle (analytics imports spend).
_BEDROCK_SPEND_LIVE_STATUSES = (
    "active", "approved", "expired", "pending_release",
)


def _is_spend_eligible(assignment: ApiKeyAssignment, month_start: date) -> bool:
    """Was this key live at any point during the month starting at
    ``month_start``?

    ``month_start`` is always the REQUESTED month — the month whose
    number a surface is about to show — never a wider query window. The
    monthly series widens its epoch-OVERLAP window (see
    :func:`_bedrock_series_extra_where`) but must keep this anchor
    fixed: a sibling key revoked inside the 6-month span but before the
    requested month would otherwise read as live for the series and
    dead for the month, so the contested guard would drop an epoch from
    the series that the MTD headline keeps — the same chart showing $130
    MTD and a $30 bar for that month.
    """
    if assignment.status in _BEDROCK_SPEND_LIVE_STATUSES:
        return True
    if assignment.status == "revoked" and assignment.revoked_at is not None:
        rev = assignment.revoked_at
        rev_date = rev.date() if hasattr(rev, "date") else rev
        return rev_date >= month_start
    return False


def _apply_per_key_deduction(payload: dict[str, Any], flt: Any) -> None:
    """Subtract sibling per-key spend from an Owner-tag payload (#2132).

    Mutates ``mtd_usd`` and the daily series in place. Floored at zero:
    the two axes are independent AWS reductions, so a rounding or
    ingest-lag skew must never produce a negative headline.

    #2225 — the series key is ``daily_series``. It read ``daily``, which
    no builder in this module has ever produced (``daily_series`` is what
    both the header path and ``compute_assignment_spend_detail`` emit,
    and what ``apikey/service.py`` + ``AssignmentSpendPanel.tsx``
    consume), so the per-day loop was dead code: the headline deducted
    correctly while every sparkline bar still showed the Owner total —
    the chart openly contradicting the number above it, which is the very
    "one question, two answers" state #2204 set out to close.
    """
    if not getattr(flt, "deduct_per_key", None):
        return
    total, daily, refs = flt.deduct_per_key
    if total <= 0 and not daily:
        return
    mtd = payload.get("mtd_usd")
    deducted = Decimal(0)
    if mtd is not None:
        before = Decimal(str(mtd))
        payload["mtd_usd"] = max(Decimal(0), before - total)
        # Report what was ACTUALLY taken off, not what we wanted to take
        # off: when the floor bites (ingest lag leaves the owner total
        # below the sibling total) the old code still claimed the full
        # amount, overstating the deduction on a page whose whole job is
        # to explain where the money went.
        deducted = before - payload["mtd_usd"]
    for point in payload.get("daily_series") or []:
        cut = daily.get(point.get("date"))
        if cut:
            point["usd"] = max(0.0, float(point.get("usd") or 0) - cut)
    payload["per_key_deducted_refs"] = refs
    payload["per_key_deducted_usd"] = str(deducted)


def _apply_per_key_deduction_monthly(
    payload: dict[str, Any], flt: Any,
) -> None:
    """Same deduction for the 6-month bar series (#2225).

    ``monthly_series`` is built AFTER the headline deduction (it needs
    its own wider transfer-epoch stitch), so it needs its own pass —
    otherwise the detail page's month bars keep the Owner total while
    the headline above them is the deducted figure, which is the same
    contradiction the daily series had, one zoom level out.
    """
    monthly = getattr(flt, "deduct_per_key_monthly", None)
    if not monthly:
        return
    for point in payload.get("monthly_series") or []:
        cut = monthly.get(point.get("month"))
        if cut:
            point["usd"] = max(0.0, float(point.get("usd") or 0) - cut)


async def _bedrock_cohort(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    integration_id: str,
    month_start: date,
    month_end: date,
) -> list[tuple[ApiKeyAssignment, str | None, str | None]]:
    """Every spend-eligible Bedrock key in the same IAM account, with its
    resolved Owner tag and its per-key axis ref — scanned ONCE.

    #2239: ``_bedrock_owner_canonical`` and
    ``_bedrock_sibling_per_key_totals`` each ran this same O(K) walk with
    a per-candidate ``_resolve_bedrock_owner_tag_value`` await, and the
    second one added a per-candidate profile-facts probe on top. The
    fleet rollup computes spend per key, so a single integration was
    O(K×N) serial round-trips — twice. The probe is now ONE grouped
    query over the whole cohort instead of one per candidate.

    Returns ``[(assignment, owner_tag, per_key_ref)]``; ``per_key_ref``
    is non-None only when that key actually has rows on the per-key
    cost-allocation axis in this window.
    """
    rows = (await session.execute(
        select(ApiKeyAssignment, ApiKeyProvider.name)
        .join(ApiKeyProvider, ApiKeyProvider.id == ApiKeyAssignment.provider_id)
        .where(
            ApiKeyAssignment.iam_integration_id
            == assignment.iam_integration_id,
        )
    )).all()

    candidates: list[ApiKeyAssignment] = [
        cand for cand, provider_name in rows
        if classify_bucket(provider_name) == BUCKET_BEDROCK
        and _is_spend_eligible(cand, month_start)
    ]
    if not candidates:
        return []

    # One query for the whole cohort's per-key axis presence.
    refs = {
        (a.request_number or "").strip().lower(): (a.request_number or "").strip()
        for a in candidates if (a.request_number or "").strip()
    }
    present: set[str] = set()
    if refs:
        present = {
            str(p).lower()
            for (p,) in (await session.execute(
                select(CmdbBedrockProfileFact.profile).where(
                    CmdbBedrockProfileFact.integration_id == integration_id,
                    func.lower(CmdbBedrockProfileFact.profile_tag_key).in_(
                        PER_KEY_PROFILE_AXES,
                    ),
                    func.lower(CmdbBedrockProfileFact.profile).in_(
                        list(refs),
                    ),
                    CmdbBedrockProfileFact.date >= month_start,
                    CmdbBedrockProfileFact.date <= month_end,
                ).distinct()
            )).all()
        }

    out: list[tuple[ApiKeyAssignment, str | None, str | None]] = []
    for cand in candidates:
        owner = await _resolve_bedrock_owner_tag_value(session, cand)
        raw_ref = (cand.request_number or "").strip()
        per_key = raw_ref if raw_ref.lower() in present else None
        out.append((cand, owner, per_key))
    return out


async def _bedrock_sibling_per_key_totals(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    owner_tag: str,
    integration_id: str,
    month_start: date,
    month_end: date,
    cohort: list[tuple[ApiKeyAssignment, str | None, str | None]] | None = None,
) -> tuple[Decimal, dict[str, float], dict[str, float], list[str]]:
    """Spend that SIBLING keys already report on the per-key axis.

    The two Bedrock attribution axes overlap (#2132). The Owner-tag
    reduction this canonical key runs matches every cost-fact row
    carrying the owner's tag — including the rows generated by a sibling
    key that is ALSO reporting its own number off
    ``cmdb_bedrock_profile_facts``. ``_caveat_bedrock_per_key_tag``
    already says so in words ("Summing the two would double-count"), but
    only to a human reader: the fleet rollup sums both, and the
    canonical key's budget is then judged against a sibling's spend.

    Deducting here fixes both consumers at once:

    * the canonical key's headline becomes ITS OWN spend, so its budget
      comparison is about its own usage again;
    * canonical + siblings once more adds up to exactly the owner total,
      so the fleet rollup stops double-counting.

    Returns ``(total, {iso_date: amount}, {YYYY-MM: amount}, key_refs)``;
    all zero/empty when no sibling is on the per-key axis, which is the
    common case.
    """
    if cohort is None:
        cohort = await _bedrock_cohort(
            session, assignment=assignment, integration_id=integration_id,
            month_start=month_start, month_end=month_end,
        )
    refs: list[str] = [
        per_key for cand, cand_owner, per_key in cohort
        if cand.id != assignment.id
        and cand_owner == owner_tag
        and per_key
    ]

    if not refs:
        return Decimal(0), {}, {}, []

    lowered = [r.lower() for r in refs]
    where = (
        CmdbBedrockProfileFact.integration_id == integration_id,
        func.lower(CmdbBedrockProfileFact.profile_tag_key).in_(
            PER_KEY_PROFILE_AXES,
        ),
        func.lower(CmdbBedrockProfileFact.profile).in_(lowered),
        CmdbBedrockProfileFact.date >= month_start,
        CmdbBedrockProfileFact.date <= month_end,
    )
    total = Decimal(str((await session.execute(
        select(func.coalesce(func.sum(CmdbBedrockProfileFact.amount_usd), 0))
        .where(*where)
    )).scalar_one() or 0))
    daily = {
        r.d.isoformat(): float(r.u or 0)
        for r in (await session.execute(
            select(
                CmdbBedrockProfileFact.date.label("d"),
                func.coalesce(
                    func.sum(CmdbBedrockProfileFact.amount_usd), 0,
                ).label("u"),
            ).where(*where).group_by(CmdbBedrockProfileFact.date)
        )).all()
    }
    # Per-MONTH sibling totals over the detail page's 6-month bar window
    # (#2225). Same predicate, wider date range, one grouped query — the
    # month bars have to be deducted the same way the headline and the
    # daily bars are, or the page shows three different answers.
    monthly_where = (
        CmdbBedrockProfileFact.integration_id == integration_id,
        func.lower(CmdbBedrockProfileFact.profile_tag_key).in_(
            PER_KEY_PROFILE_AXES,
        ),
        func.lower(CmdbBedrockProfileFact.profile).in_(lowered),
        CmdbBedrockProfileFact.date >= _monthly_series_window_start(month_end),
        CmdbBedrockProfileFact.date <= month_end,
    )
    monthly = {
        f"{int(r.y):04d}-{int(r.m):02d}": float(r.u or 0)
        for r in (await session.execute(
            select(
                func.extract("year", CmdbBedrockProfileFact.date).label("y"),
                func.extract("month", CmdbBedrockProfileFact.date).label("m"),
                func.coalesce(
                    func.sum(CmdbBedrockProfileFact.amount_usd), 0,
                ).label("u"),
            ).where(*monthly_where).group_by(
                func.extract("year", CmdbBedrockProfileFact.date),
                func.extract("month", CmdbBedrockProfileFact.date),
            )
        )).all()
    }
    return total, daily, monthly, sorted(refs)


async def _bedrock_owner_canonical(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    owner_tag: str,
    month_start: date,
    month_end: date | None = None,
    cohort: list[tuple[ApiKeyAssignment, str | None, str | None]] | None = None,
) -> tuple[bool, int, str | None]:
    """Resolve the shared-Owner-tag collision for a Bedrock AK (#1776).

    AWS Cost Explorer aggregates Bedrock cost at the ``Owner`` tag (=
    work_email), not per IAM user / per key, so when one person holds
    >1 live Bedrock key every key's naive per-Owner match returns the
    person's ENTIRE Bedrock spend — an N× double-count in the fleet
    rollup + false budget breaches. Pin the total to one deterministic
    canonical key so it's counted exactly once.

    Returns ``(is_canonical, sibling_count, canonical_request_number)``:
    * ``is_canonical`` — True when THIS assignment owns the total (the
      only key, or the earliest-created of the colliding set). The
      caller proceeds to the real per-Owner sum.
    * ``sibling_count`` — number of OTHER spend-eligible keys sharing the
      tag (0 ⇒ no collision).
    * ``canonical_request_number`` — the canonical key's request_number
      (for the sibling's caveat); None when this key is canonical.

    Canonicity is decided among keys that are spend-eligible in the SAME
    month window and the SAME IAM account, so the canonical is always a
    key that actually appears in the rollup — no under/over-count at the
    month boundary. Keys that report on the PER-KEY axis are excluded
    from the candidate set (#2226): they never answer for the Owner-tag
    total, so one of them winning canonicity leaves that total unreported
    by anybody.
    """
    # Candidate live Bedrock keys in the same AWS account. Join the
    # provider name so we can classify the bucket without a per-row
    # round-trip; restrict by iam_integration_id (the cost-facts filter
    # is per integration, so a same-email collision only matters within
    # one account).
    if cohort is None:
        # Standalone call (tests / a future caller): scan the cohort here.
        # ``month_end`` only bounds the per-key axis probe, so when the
        # caller did not supply one, the month containing ``month_start``
        # is the right window — the same one canonicity is judged in.
        cohort = await _bedrock_cohort(
            session, assignment=assignment,
            integration_id=str(assignment.iam_integration_id),
            month_start=month_start,
            month_end=month_end or _month_end_for(month_start),
        )

    # #2226 — a key that is on the PER-KEY axis can never answer for the
    # Owner-tag total: it returns early with its own profile-facts sum
    # and never reaches the Owner reduction. Leaving it in the candidate
    # set meant that when it happened to be the earliest-created key it
    # WON canonicity, and then nobody reported the Owner total at all:
    # the real Owner-tag key resolved ``is_canonical=False``, was pinned
    # to $0, and its caveat pointed the holder at a key whose page does
    # not contain that money. Fleet ``total_mtd`` under-counted by the
    # whole amount and that key's budget alerts went dead.
    same_owner = [
        (cand, per_key) for cand, cand_owner, per_key in cohort
        if cand_owner is not None and cand_owner == owner_tag
    ]
    # ``sibling_count`` keeps its documented meaning — OTHER spend-eligible
    # keys sharing the tag, per-key axis included (they DO share it; they
    # just answer on a different axis).
    sibling_count = sum(1 for cand, _pk in same_owner if cand.id != assignment.id)
    # …but only a key that will actually PERFORM the Owner-tag reduction
    # can be canonical for it.
    matches = [cand for cand, per_key in same_owner if per_key is None]

    if not matches:
        # Every key on this tag reports on the per-key axis, so there is
        # no Owner-tag total left for anyone to own.
        return (True, 0, None)

    # Deterministic canonical: earliest created, id as tiebreak.
    canonical = min(matches, key=lambda a: (a.created_at, str(a.id)))
    if canonical.id == assignment.id:
        return (True, sibling_count, None)
    # Reached either by an Owner-tag sibling that lost the pin, or by a
    # per-key key asking "who holds the Owner total?" for its caveat —
    # which is the question this return has always been answering, and
    # which the old candidate set could get wrong by naming the asker.
    return (False, sibling_count, canonical.request_number)


async def _bedrock_canonical_assignment_id(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    canonical_ref: str | None,
) -> str | None:
    """PK of the canonical key named by ``canonical_ref``, for the
    frontend's deep link.

    The pinned sibling's caveat names the canonical key by request
    number; without an id the reader had to go hunt for it in the list —
    a dead end on the very page that just told them their money is
    somewhere else. Scoped to the same IAM integration so a request
    number reused across accounts can't resolve to a stranger's key.
    """
    if not canonical_ref:
        return None
    row = (await session.execute(
        select(ApiKeyAssignment.id).where(
            ApiKeyAssignment.request_number == canonical_ref,
            ApiKeyAssignment.iam_integration_id
            == assignment.iam_integration_id,
        ).limit(1)
    )).scalar_one_or_none()
    return str(row) if row is not None else None


async def _bedrock_epoch_owner_contested(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    owner_value: str,
    eligibility_month_start: date,
) -> bool:
    """True when another spend-eligible Bedrock key on the same IAM
    integration currently resolves ``owner_value`` as its Owner tag.

    Guard for the transfer-epoch stitch: attributing a previous
    holder's tag epoch to this key is only safe when nothing else
    claims that tag today — otherwise the same rows land on two keys
    and the fleet rollup double-counts (the exact failure #1776's
    canonical pinning exists to prevent). Contested ⇒ the epoch is
    skipped; the un-stitched number is under- rather than over-stated.

    ``eligibility_month_start`` is the REQUESTED month, not the window
    the epochs are being overlapped against: the answer to "does another
    key claim this tag" must be the same for the MTD headline and for
    every bar of the 6-month series drawn beside it (see
    :func:`_is_spend_eligible`). It also matches the anchor
    :func:`_bedrock_owner_canonical` decides canonicity on, which is
    resolved once per page from the requested month.
    """
    rows = (await session.execute(
        select(ApiKeyAssignment, ApiKeyProvider.name)
        .join(ApiKeyProvider, ApiKeyProvider.id == ApiKeyAssignment.provider_id)
        .where(
            ApiKeyAssignment.iam_integration_id == assignment.iam_integration_id,
        )
    )).all()
    for cand, provider_name in rows:
        if cand.id == assignment.id:
            continue
        if classify_bucket(provider_name) != BUCKET_BEDROCK:
            continue
        if not _is_spend_eligible(cand, eligibility_month_start):
            continue
        cand_owner = await _resolve_bedrock_owner_tag_value(session, cand)
        if cand_owner is not None and cand_owner.lower() == owner_value.lower():
            return True
    return False


def _transfer_epoch_end_with_retag(
    retag_state: dict[str, Any] | None,
    *,
    completed_d: date,
) -> date:
    """Effective end (inclusive) of a previous holder's tag epoch,
    accounting for a DELAYED cloud-side retag.

    ``apply_transfer``'s IAM tag push is best-effort — when it fails at
    accept-time, the AWS ``Owner`` tag keeps the OLD holder's value
    until an admin re-runs it via ``retry_iam_retag``, so cost-fact
    days between transfer completion and the actual successful retag
    still carry the old value. Bounding the epoch at ``completed_at``
    would leave those days matched by NEITHER the epoch clause (date >
    epoch_end) NOR the current-owner clause (different tag value) —
    spend silently vanishing with no caveat.

    ``retag_state`` is ``assignment.detail.last_transfer`` when that
    blob describes THIS hand-over, else None — retag state is only
    recorded for the most recent transfer, so older events keep the
    completion-date bound, matching pre-retry behaviour (see
    :func:`_retag_state_for_event`). The tag is
    known to have flipped no later than the FIRST retag attempt that
    reported ``tagged``, so that attempt's date bounds the epoch and
    every attempt after it is ignored — an admin re-running the retag
    on an already-correct tag (or a later attempt that errors because
    the IAM user is gone) must not push the epoch end, and the caveat's
    ``until``, off the real hand-over.

    * a retry landed it → ``max(completed_d, first successful
      attempt's date)``;
    * no successful attempt on record, current summary ``tagged`` →
      ``completed_d``: the accept-time push landed and no retry was
      needed;
    * no successful attempt and the summary is failed / skipped → the
      retag is still pending, the old value is live on AWS through
      today → the epoch stays open-ended at today's date (recomputed
      per read, so it keeps extending until the retag succeeds).
      Double-attribution stays impossible: the contested guard already
      drops the epoch when another live key claims the old owner's tag.

    NOTE ``cloud_tag_summary`` is NOT the accept-time outcome once a
    retry has run — ``retry_iam_retag`` overwrites it with the latest
    attempt's result (``service.py``), so the accept-time status is
    unrecoverable and the first recorded attempt's own precondition is
    unknown. Where the record is ambiguous this over-states the epoch
    end rather than under-stating it: an extra day adds an OR-clause
    that matches no rows when the tag really did flip at completion,
    whereas a missing day silently drops the previous holder's spend.

    CALENDARS — the two possible ends are anchored differently, on
    purpose:

    * a CLOSED end is a **UTC** date. ``completed_d`` comes from
      ``AssetTransferRequest.completed_at`` / ``audit_logs.created_at``
      (both UTC, see :func:`_event_ts`) and ``retagged_d`` from
      ``retag_attempts[].at``, written as ``datetime.now(UTC)``. That
      matches the column it is compared against: ``cmdb_cost_facts.date``
      is a UTC billing day (AWS Cost Explorer's day granularity).
    * the OPEN end is a **Shanghai** date, because it is compared against
      a Shanghai-clamped window: the callers pass ``month_end =
      min(month_end, today_shanghai)``. A UTC "today" here would sit one
      day BEHIND that clamp for the eight hours after Shanghai midnight
      and drop the newest day of a still-un-retagged holder's spend.
      Shanghai runs at most one day ahead of the UTC billing day, and an
      end one day early costs nothing (no facts exist for a UTC day that
      has not started), which is the same over-state-don't-under-state
      posture as the paragraph above.
    """
    last = retag_state or {}
    if not last:
        return completed_d
    for attempt in last.get("retag_attempts") or []:
        summary = (attempt.get("summary") or {}).get("bedrock") or {}
        if summary.get("status") != "tagged":
            continue
        try:
            retagged_d = datetime.fromisoformat(attempt["at"]).date()
        except (KeyError, TypeError, ValueError):
            continue
        return max(completed_d, retagged_d)
    bedrock = (last.get("cloud_tag_summary") or {}).get("bedrock") or {}
    status = bedrock.get("status")
    if not status or status == "tagged":
        # No AWS binding at transfer time / pre-feature row, or the
        # accept-time push landed — keep the completion bound.
        return completed_d
    # failed / skipped and never retagged: the old Owner value is still
    # live on AWS.
    return max(completed_d, _today_shanghai())


# Immutable provenance for a FORCE transfer. ``apply_transfer`` writes
# this row (``service.py``) inside the same transaction that flips
# ``assigned_to``, so ``created_at`` IS the hand-over instant.
_AUDIT_ACTION_FORCE_TRANSFER = "apikey.transfer.force_complete"
_AUDIT_RESOURCE_TYPE_ASSIGNMENT = "api_key_assignment"


class _TransferEvent(NamedTuple):
    """One completed hand-over of this assignment, from either
    provenance store.

    ``transfer_request_id`` is None for a FORCE transfer: the
    offboarding T+0 advance, the admin reconciliation fix, and the
    Bailian owner coordinator all call ``apply_transfer`` with
    ``transfer_request_id=None`` and create no ``AssetTransferRequest``
    row, so those hand-overs are recoverable only from the audit row.

    ``from_person_id`` is None when a force row's ``detail`` was
    replaced by the audit service's 16 KiB truncation marker: the date
    still chains the NEXT holder's epoch start, but the outgoing
    holder is unrecoverable and no epoch is emitted for it — a guessed
    hand-over moves real money between two people's segments.
    """

    completed_at: datetime
    from_person_id: Any | None
    transfer_request_id: Any | None


def _event_ts(completed_at: datetime) -> datetime:
    """Naive-UTC ordering key for the merged transfer chain.

    ``AssetTransferRequest.completed_at`` is written as UTC and
    ``audit_logs.created_at`` as the DB's ``NOW()``; MySQL/TiDB hand
    both back naive, but an aware value must not raise on comparison
    against a naive one when the two stores are merged.
    """
    if completed_at.tzinfo is not None:
        return completed_at.astimezone(UTC).replace(tzinfo=None)
    return completed_at


async def _force_transfer_events(
    session: AsyncSession, *, assignment: ApiKeyAssignment,
) -> list[_TransferEvent]:
    """Force-transfer hand-overs of this assignment, read from the
    immutable ``apikey.transfer.force_complete`` audit rows.

    The audit row — not ``detail.last_transfer`` — is the authority for
    a force hand-over date: ``last_transfer`` holds only the LATEST
    transfer (a chain of force transfers is unrecoverable from it) and
    ``retry_iam_retag`` rewrites parts of the blob, whereas audit is
    append-only. Force transfers write no ``AssetTransferRequest`` row,
    so without this read the successor's key silently lost every day of
    pre-hand-over Bedrock spend — the leaver→manager case, i.e. exactly
    when someone is auditing the number.

    Equality on ``(resource_type, resource_id)`` rides the composite
    index from migration 180, so the fleet rollup's per-key loop stays
    an index range scan.
    """
    # Cross-package model read, deferred like the transfer-request one
    # below so the module import graph stays acyclic.
    from app.audit.models import AuditLog

    rows = (await session.execute(
        select(AuditLog.detail, AuditLog.created_at)
        .where(
            AuditLog.action == _AUDIT_ACTION_FORCE_TRANSFER,
            AuditLog.resource_type == _AUDIT_RESOURCE_TYPE_ASSIGNMENT,
            AuditLog.resource_id == str(assignment.id),
            AuditLog.status == "success",
        )
    )).all()
    events: list[_TransferEvent] = []
    for detail, created_at in rows:
        if created_at is None:
            continue
        raw_from = (detail or {}).get("from_person_id")
        try:
            from_id = (
                uuid.UUID(str(raw_from)) if raw_from is not None else None
            )
        except (AttributeError, ValueError):
            # Unparseable id: same posture as a truncated detail — chain
            # the date, emit no epoch.
            from_id = None
        events.append(_TransferEvent(
            completed_at=created_at,
            from_person_id=from_id,
            transfer_request_id=None,
        ))
    return events


def _retag_state_for_event(
    assignment: ApiKeyAssignment,
    *,
    event: _TransferEvent,
    is_latest: bool,
) -> dict[str, Any] | None:
    """``detail.last_transfer`` when it describes ``event``, else None.

    ``last_transfer`` only ever holds the most recent hand-over. A
    voluntary event matches on ``transfer_request_id`` (exact); a force
    event has none, so it matches only when it IS the newest event in
    the chain AND the blob is itself a force record for the same
    outgoing holder — a voluntary transfer chained after a force one
    must not inherit the force row's retag attempts, and vice versa.
    """
    last = (assignment.detail or {}).get("last_transfer") or {}
    if not last:
        return None
    if event.transfer_request_id is not None:
        if str(event.transfer_request_id) == (
            last.get("transfer_request_id") or ""
        ):
            return last
        return None
    if (
        is_latest
        and last.get("force_transfer")
        and not last.get("transfer_request_id")
        and event.from_person_id is not None
        and str(last.get("from_person_id") or "") == str(event.from_person_id)
    ):
        return last
    return None


async def _bedrock_series_extra_where(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    owner_tag: str,
    month_start: date,
    month_end: date,
):
    """``extra_where`` for the 6-month monthly series.

    Same recipe as the requested-month filter, with the transfer epochs
    re-resolved against ``[series_window_start, month_end]`` so every bar
    carries the holder(s) that owned the tag in ITS month. Costs one
    extra epoch resolution on the Bedrock detail page only.

    Only the OVERLAP window widens. ``month_start`` (the requested
    month) is threaded through unchanged as the sibling-liveness anchor
    so the series and the MTD headline agree about which epochs are
    contested — see :func:`_is_spend_eligible`.
    """
    epochs = await _bedrock_owner_epochs(
        session,
        assignment=assignment,
        current_owner_tag=owner_tag,
        window_start=_monthly_series_window_start(month_end),
        window_end=month_end,
        eligibility_month_start=month_start,
    )
    return _bedrock_owner_where(owner_tag, epochs)


async def _bedrock_owner_epochs(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    current_owner_tag: str,
    window_start: date,
    window_end: date,
    eligibility_month_start: date,
) -> list[tuple[str, date | None, date]]:
    """Previous-holder Owner-tag epochs from completed ownership
    transfers of this assignment, oldest first — restricted to epochs
    that overlap the ``[window_start, window_end]`` query window.

    Two windows, deliberately separate. ``[window_start, window_end]``
    is what the caller is about to SUM, so it decides which epochs are
    worth an OR-clause; the monthly series passes its own 6-month span
    here. ``eligibility_month_start`` is the REQUESTED month and decides
    only whether a sibling key counts as live for the contested guard —
    a per-page constant, so the series' bars and the MTD headline keep
    or drop the same epochs (see :func:`_is_spend_eligible`).

    Both provenance stores are merged into ONE ordered chain: voluntary
    transfers from ``AssetTransferRequest`` rows and force transfers
    from ``apikey.transfer.force_complete`` audit rows (see
    :func:`_force_transfer_events`). Force transfers create no request
    row, so reading only the request table returned ``[]`` for a
    force-transferred key and dropped every pre-hand-over day with no
    caveat; it also mis-started the chain on a force→voluntary pair (the
    voluntary epoch opened at ``None`` instead of the force date, so it
    could claim days that belonged to another key holding the same tag).

    Returns ``[(owner_value, start_inclusive, end_inclusive)]`` where
    ``owner_value`` is the previous holder's work_email — the value
    the transfer flow re-tags AWAY from on hand-over (the transfer's
    ``cloud_tag_summary`` records only the new email, and both the
    autogrant and the transfer retag write the holder's work_email,
    so the previous holder's email is the faithful historic value).
    ``end_inclusive`` is the transfer-completion date — pushed later
    when the cloud-side retag lagged the hand-over (see
    :func:`_transfer_epoch_end_with_retag`): AWS Cost Explorer
    attributes each day to the tag value at usage time, so days up to
    and including the day the tag actually flipped carry the old value.
    ``start_inclusive`` is the PREVIOUS transfer's completion date for
    chained transfers (None for the first).

    Both bounds inclusive means adjacent epochs deliberately SHARE the
    hand-over day: the retag lands mid-day, so that day carries rows
    under the outgoing and the incoming tag, and each tag's rows belong
    to the holder that tag names. Nothing is double-counted by the
    overlap — the two clauses match disjoint ``resource_id`` prefixes.
    The one shape where that reasoning breaks is a chain that hands the
    key BACK to an earlier holder (A→B→C→B), whose two epochs for the
    same tag value can overlap once a retag lag stretches the earlier
    one; :func:`_coalesced_epochs` merges those before they reach SQL.
    Do NOT make the start exclusive again to "fix" an overlap: that
    silently drops every INTERMEDIATE holder's hand-over-day rows (only
    the final holder's clause is unbounded and therefore safe).

    Epochs whose ``[start_inclusive, end_inclusive]`` range doesn't
    intersect the query window are dropped — a later month behaves
    exactly like a never-transferred key (no dead OR-clauses, no
    stale transfer caveat, and no per-epoch contested scan). Epochs
    that resolve to the CURRENT owner tag are dropped too (the
    unbounded current-owner clause already matches them), as are
    epochs whose owner value another live key still claims (see
    :func:`_bedrock_epoch_owner_contested`).
    """
    # Cross-domain model read (same precedent as service.py's
    # transfer delegation) — kept inside the function so the module
    # import graph stays acyclic.
    from app.domains.offboarding.models import (
        XFER_ACCEPTED,
        XFER_COMPLETED,
        AssetTransferRequest,
    )

    rows = (await session.execute(
        select(
            AssetTransferRequest.id,
            AssetTransferRequest.initiator_id,
            AssetTransferRequest.completed_at,
        )
        .where(
            AssetTransferRequest.asset_type == "api_key",
            AssetTransferRequest.asset_ref_id == assignment.id,
            AssetTransferRequest.status.in_((XFER_ACCEPTED, XFER_COMPLETED)),
            AssetTransferRequest.completed_at.is_not(None),
        )
        .order_by(AssetTransferRequest.completed_at.asc())
    )).all()

    # ``initiator_id`` is the OUTGOING holder: the current owner is who
    # creates a hand-over request. Force events carry the same identity
    # under the audit row's ``from_person_id``.
    events = [
        _TransferEvent(
            completed_at=completed_at,
            from_person_id=initiator_id,
            transfer_request_id=transfer_id,
        )
        for transfer_id, initiator_id, completed_at in rows
    ]
    events.extend(
        await _force_transfer_events(session, assignment=assignment)
    )
    if not events:
        return []
    events.sort(key=lambda e: _event_ts(e.completed_at))
    latest_idx = len(events) - 1

    epochs: list[tuple[str, date | None, date]] = []
    prev_end: date | None = None
    for idx, event in enumerate(events):
        completed_at = event.completed_at
        completed_d = (
            completed_at.date() if hasattr(completed_at, "date")
            else completed_at
        )
        end_d = _transfer_epoch_end_with_retag(
            _retag_state_for_event(
                assignment, event=event, is_latest=idx == latest_idx,
            ),
            completed_d=completed_d,
        )
        # ``prev_end`` chains on the COMPLETION date (not the retag-
        # extended end) so the next holder's epoch never shrinks — a
        # retag-lag overlap between two epochs is harmless because the
        # owner values are distinct resource_id prefixes. The chained
        # value is the next epoch's INCLUSIVE start: the hand-over day
        # belongs to both holders' tags (see the docstring).
        start_d, prev_end = prev_end, completed_d
        if event.from_person_id is None:
            # Force row with unrecoverable ``from_person_id`` (audit
            # detail truncated): the date above still chains the NEXT
            # holder's epoch start; this holder's own epoch is dropped
            # rather than inferred.
            continue
        # Window-overlap gate BEFORE the email + contested lookups so
        # months with no overlapping epoch pay zero extra queries. Both
        # comparisons are inclusive-vs-inclusive: an epoch that opens ON
        # the last day of the window still owns that day's rows under
        # the outgoing tag.
        if (
            end_d < window_start
            or (start_d is not None and start_d > window_end)
        ):
            continue
        email = (await session.execute(
            select(IdentityPerson.work_email).where(
                IdentityPerson.id == event.from_person_id,
            )
        )).scalar_one_or_none()
        owner_value = (email or "").strip()
        if (
            owner_value
            and owner_value.lower() != current_owner_tag.lower()
            and not await _bedrock_epoch_owner_contested(
                session,
                assignment=assignment,
                owner_value=owner_value,
                eligibility_month_start=eligibility_month_start,
            )
        ):
            epochs.append((owner_value, start_d, end_d))
    return _coalesced_epochs(epochs)


def _coalesced_epochs(
    epochs: list[tuple[str, date | None, date]],
) -> list[tuple[str, date | None, date]]:
    """Merge overlapping ranges that name the SAME Owner tag value.

    The epoch clauses are OR-ed inside one ``SUM``, so two ranges for one
    tag value that share a day sum that day's rows twice. Distinct tag
    values never collide (disjoint ``resource_id`` prefixes), which is
    why only same-owner ranges are merged and the deliberate adjacent-
    epoch overlap on a hand-over day is left alone.

    Reachable via a chain that returns the key to an earlier holder
    (A→B→C→B): B's two epochs touch once the earlier one's retag-lag end
    reaches the later one's start, or when B→C and C→B complete on the
    same day. Output stays oldest-first (open-ended epochs lead).
    """
    grouped: dict[str, tuple[str, list[tuple[date | None, date]]]] = {}
    for owner, start, end in epochs:
        _, ranges = grouped.setdefault(owner.lower(), (owner, []))
        ranges.append((start, end))
    out: list[tuple[str, date | None, date]] = []
    for label, ranges in grouped.values():
        ranges.sort(key=lambda r: (r[0] is not None, r[0] or date.min))
        merged: list[tuple[date | None, date]] = []
        for start, end in ranges:
            if merged and (start is None or start <= merged[-1][1]):
                prev_start, prev_end = merged[-1]
                merged[-1] = (
                    prev_start if start is not None else None,
                    max(prev_end, end),
                )
            else:
                merged.append((start, end))
        out.extend((label, s, e) for s, e in merged)
    out.sort(key=lambda e: (e[1] is not None, e[1] or date.min, e[2], e[0]))
    return out


# ── Whitelist resolver shared with the AI Spending dashboard ───────


async def _whitelist_patterns_for(
    session: AsyncSession, provider: str,
) -> list[str]:
    """Return the admin-configured service substring patterns for one
    provider, freshly read from the platform_setting.

    Pulls the same row the AI Spending dashboard reads via
    :func:`app.domains.ai_spending.whitelist.load_whitelist`, so a
    pattern an admin adds in Settings shows up in BOTH the dashboard's
    Foundry/Bedrock/Vertex bucket total AND every per-AK budget bar
    that uses this module — no drift between the two surfaces.

    Returns an empty list when the provider has no patterns configured
    (admin blanked it, or the seed never populated GCP). The caller's
    ``_sum_with_service_patterns`` then returns the standard zero-spend
    payload with a structured caveat — the panel renders an honest
    "AI service whitelist not configured for this provider" message
    instead of an inflated number.
    """
    wl = await load_whitelist(session)
    return list(wl.get(provider) or [])


# ── SQL helper shared by the bucket-specific paths ─────────────────


async def _bailian_key_pks(
    session: AsyncSession, assignment: ApiKeyAssignment,
) -> list[str]:
    """Local ``bailian_api_keys.id`` PKs minted for this assignment.

    The Bailian auto-grant links each minted DashScope key back to its
    assignment via ``bailian_api_keys.assignment_id``; a reissue can leave
    >1 row (old + new), so we sum spend across all of them. Soft-deleted
    rows are INCLUDED: since #1230 retires the mirror row on revoke, filtering
    ``deleted_at IS NULL`` would zero a key's month-to-date the instant it is
    revoked mid-month — dropping spend that genuinely landed under this
    assignment earlier in the period. The usage rows are only reachable from
    the panel through this lookup, so excluding the deleted row makes that
    spend unreachable, not "captured regardless". Attribution is by
    ``assignment_id``, which a revoke does not clear, so this stays correct.
    """
    return list((await session.execute(
        select(BailianApiKey.id).where(
            BailianApiKey.assignment_id == assignment.id,
        )
    )).scalars().all())


async def _spend_bailian(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    month_start: date,
    month_end: date,
) -> dict[str, Any]:
    """Per-AK Bailian spend from ``bailian_usage_daily`` (migration 299).

    Sums ``cost_usd`` over the rows whose ``api_key_pk`` is one of the
    assignment's linked Bailian keys for the window. Returns the SAME
    payload shape as :func:`_sum_with_service_patterns` (so the panel,
    the budget cron, and the detail page reconcile to the cent). When no
    Bailian key is linked yet, returns the honest no-key zero payload
    (``data_available=False``) instead of a misleading ``$0``.
    """
    key_pks = await _bailian_key_pks(session, assignment)
    if not key_pks:
        return _zero_spend(
            bucket=BUCKET_BAILIAN,
            month_start=month_start,
            month_end=month_end,
            data_caveat=_CAVEAT_BAILIAN_NO_KEY,
            data_caveat_code=CAVEAT_BAILIAN_NO_KEY,
        )
    base_where = (
        BailianUsageDaily.api_key_pk.in_(key_pks),
        BailianUsageDaily.usage_date >= month_start,
        BailianUsageDaily.usage_date <= month_end,
        BailianUsageDaily.model_name != BAILIAN_AGGREGATE_MODEL,
    )
    # 三口径一次取齐。折前(牌价)是**员工看得懂的那个数** —— 百炼的
    # 实付常年被代金券/合同折扣抵到 0,只给实付的话,自助页上一把每天
    # 都在用的 Key 会显示「本月 $0.00」,读起来像「没在用」。预算条
    # 仍然按实付走(预算管的是真花出去的钱),折前只做展示。
    totals = (await session.execute(
        select(
            func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).label(
                "paid",
            ),
            func.sum(BailianUsageDaily.cost_usd_gross).label("gross"),
        )
        .where(*base_where)
    )).one()
    mtd_usd = Decimal(str(totals.paid or 0))
    # NULL(而非 0)= 这段窗口没有任何一行带牌价 —— 前端据此完全不渲染
    # 该档,而不是显示一个断言性的 "$0.00 折前"。
    mtd_gross_usd = (
        Decimal(str(totals.gross)) if totals.gross is not None else None
    )
    # ``cost_usd_after_discount``(合同折扣后、代金券前)**不出这个函数**。
    # 折前 ÷ 折扣后 **就是公司合同折扣率**,而 Key 持有人按定义在保密圈
    # 之外(``_INVOICE_MEASURE_ROLES``)—— 与 ``cost_view`` 开头那条红线
    # 同一件事。自助页要的是「这个月我用掉了多少钱的量」,折前一档就够;
    # 三档同框才是泄露。

    raw_daily: dict[str, float] = {
        r.d.isoformat(): float(r.u or 0)
        for r in (await session.execute(
            select(
                BailianUsageDaily.usage_date.label("d"),
                func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).label("u"),
            )
            .where(*base_where)
            .group_by(BailianUsageDaily.usage_date)
        )).all()
    }
    # Trailing truncation at last day with data — same rationale as the
    # cost_facts path (avoid misleading "$0 on the last N days").
    last_fact_date = max(raw_daily.keys()) if raw_daily else None
    daily_series: list[dict[str, Any]] = []
    if last_fact_date is not None:
        last_dt = date.fromisoformat(last_fact_date)
        cur = month_start
        while cur <= last_dt:
            iso = cur.isoformat()
            daily_series.append(
                {"date": iso, "usd": round(raw_daily.get(iso, 0.0), 6)},
            )
            cur += timedelta(days=1)
    return {
        "bucket": BUCKET_BAILIAN,
        "mtd_usd": mtd_usd,
        "mtd_gross_usd": mtd_gross_usd,
        "daily_series": daily_series,
        "last_fact_date": last_fact_date,
        "data_available": True,
        "data_caveat": None,
        "data_caveat_code": None,
        "data_caveat_params": {},
    }


async def _bailian_by_model(
    session: AsyncSession,
    *,
    key_pks: list[str],
    month_start: date,
    month_end: date,
    month_total_usd: Decimal,
) -> list[dict[str, Any]]:
    """Per-model Bailian breakdown for the detail page (migration 299
    made ``model_name`` a real dimension). BSS exposes cost but not token
    counts, so ``usage_quantity`` is 0 and the effective unit price is
    ``None`` (UI renders "—") — same contract as the cost_facts buckets."""
    rows = (await session.execute(
        select(
            BailianUsageDaily.model_name.label("model"),
            func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).label("usd"),
        )
        .where(
            BailianUsageDaily.api_key_pk.in_(key_pks),
            BailianUsageDaily.usage_date >= month_start,
            BailianUsageDaily.usage_date <= month_end,
            BailianUsageDaily.model_name != BAILIAN_AGGREGATE_MODEL,
        )
        .group_by(BailianUsageDaily.model_name)
    )).all()
    month_total = float(month_total_usd or 0)
    out: list[dict[str, Any]] = []
    for r in rows:
        usd = float(r.usd or 0)
        out.append({
            "model": str(r.model),
            "mtd_usd": round(usd, 6),
            "usage_quantity": 0.0,
            "usage_unit": None,
            "effective_unit_price_usd": None,
            "percent_of_month": (
                round(usd / month_total * 100.0, 1) if month_total > 0 else 0.0
            ),
        })
    out.sort(key=lambda x: x["mtd_usd"], reverse=True)
    return out


async def _bailian_monthly_series(
    session: AsyncSession,
    *,
    key_pks: list[str],
    month_end: date,
) -> list[dict[str, Any]]:
    """Six monthly Bailian totals ending at ``month_end``'s month, oldest
    first — mirrors :func:`_aggregate_monthly_series` for the cost_facts
    buckets."""
    year, month_num = month_end.year, month_end.month
    months: list[tuple[int, int]] = []
    for _ in range(6):
        months.append((year, month_num))
        month_num -= 1
        if month_num == 0:
            month_num = 12
            year -= 1
    oldest_year, oldest_month = months[-1]
    window_start = date(oldest_year, oldest_month, 1)

    year_col = func.extract("year", BailianUsageDaily.usage_date).label("y")
    month_col = func.extract("month", BailianUsageDaily.usage_date).label("m")
    rows = (await session.execute(
        select(
            year_col,
            month_col,
            func.coalesce(func.sum(BailianUsageDaily.cost_usd), 0).label("usd"),
        )
        .where(
            BailianUsageDaily.api_key_pk.in_(key_pks),
            BailianUsageDaily.usage_date >= window_start,
            BailianUsageDaily.usage_date <= month_end,
            BailianUsageDaily.model_name != BAILIAN_AGGREGATE_MODEL,
        )
        .group_by(year_col, month_col)
    )).all()
    by_month: dict[str, float] = {}
    for r in rows:
        by_month[f"{int(r.y):04d}-{int(r.m):02d}"] = float(r.usd or 0)
    return [
        {"month": f"{y:04d}-{m:02d}", "usd": round(by_month.get(f"{y:04d}-{m:02d}", 0.0), 6)}
        for y, m in reversed(months)
    ]


async def _sum_with_service_patterns(
    session: AsyncSession,
    *,
    bucket: str,
    integration_id: str,
    month_start: date,
    month_end: date,
    service_patterns: tuple[str, ...],
    extra_where,
) -> dict[str, Any]:
    """Aggregate cost_facts under (integration, service-family,
    extra-where) for the window. Returns the standard spend payload.

    ``service_patterns`` is a tuple of substrings resolved from the
    admin-editable ``ai_spending.service_whitelist``; the ``service``
    column must match at least one (case-insensitive). When the
    provider's whitelist is blank (admin deliberately blanked it, or
    the migration seed was empty for GCP) we return a zero-spend
    payload tagged with the whitelist-not-configured caveat instead
    of crashing — the panel renders an honest explanation and the
    cron budget compare gracefully skips threshold evaluation.
    """
    if not service_patterns:
        return _zero_spend(
            bucket=bucket,
            month_start=month_start,
            month_end=month_end,
            data_caveat=_CAVEAT_WHITELIST_NOT_CONFIGURED,
            data_caveat_code=CAVEAT_WHITELIST_NOT_CONFIGURED,
        )
    # OR-fold the per-pattern ILIKE clauses. ``or_(single)`` is
    # equivalent to the single clause in SQLAlchemy, so we don't need
    # to special-case the 1-pattern path.
    service_filter = or_(
        *(CmdbCostFact.service.ilike(f"%{_esc_like(p)}%", escape="\\") for p in service_patterns),
    )

    base_where = (
        (CmdbCostFact.integration_id == integration_id),
        (CmdbCostFact.date >= month_start),
        (CmdbCostFact.date <= month_end),
        service_filter,
        extra_where,
    )

    mtd_stmt = select(
        func.coalesce(func.sum(CmdbCostFact.amount_usd), 0),
    ).where(*base_where)
    mtd_raw = (await session.execute(mtd_stmt)).scalar_one()
    mtd_usd = Decimal(str(mtd_raw or 0))

    daily_stmt = (
        select(
            CmdbCostFact.date.label("d"),
            func.coalesce(func.sum(CmdbCostFact.amount_usd), 0).label("u"),
        )
        .where(*base_where)
        .group_by(CmdbCostFact.date)
    )
    raw_daily: dict[str, float] = {
        r.d.isoformat(): float(r.u or 0)
        for r in (await session.execute(daily_stmt)).all()
    }
    # Truncate at the last day we have data for. Trailing zero-fill
    # otherwise renders as "$0 spend on these last days" on the AK
    # spend sparkline — misleading when the cost-facts cron is lagging.
    # Same pattern + rationale as ``scope_billing_detail._daily_trend``.
    last_fact_date = max(raw_daily.keys()) if raw_daily else None
    daily_series: list[dict[str, Any]] = []
    if last_fact_date is not None:
        last_dt = date.fromisoformat(last_fact_date)
        cur = month_start
        while cur <= last_dt:
            iso = cur.isoformat()
            daily_series.append(
                {"date": iso, "usd": round(raw_daily.get(iso, 0.0), 6)},
            )
            cur += timedelta(days=1)

    return {
        "bucket": bucket,
        "mtd_usd": mtd_usd,
        "daily_series": daily_series,
        "last_fact_date": last_fact_date,
        "data_available": True,
        "data_caveat": None,
        "data_caveat_code": None,
        "data_caveat_params": {},
    }


__all__ = [
    "BUCKET_BAILIAN",
    "BUCKET_BEDROCK",
    "BUCKET_FOUNDRY",
    "BUCKET_UNKNOWN",
    "BUCKET_VERTEX",
    "CAVEAT_BAILIAN_NO_KEY",
    "CAVEAT_BEDROCK_NO_IAM",
    "CAVEAT_BEDROCK_NO_OWNER_TAG",
    "CAVEAT_BEDROCK_OWNER_SHARED",
    "CAVEAT_BEDROCK_PER_KEY_TAG",
    "CAVEAT_BEDROCK_PER_KEY_TAG_SHARED",
    "CAVEAT_BEDROCK_OWNER_UNALLOCATED",
    "CAVEAT_BEDROCK_TRANSFER_EPOCHS",
    "CAVEAT_FOUNDRY_NO_RG",
    "CAVEAT_UNKNOWN_PROVIDER",
    "CAVEAT_VERTEX_NO_SCOPE",
    "CAVEAT_WHITELIST_NOT_CONFIGURED",
    "classify_bucket",
    "compute_assignment_spend",
    "compute_assignment_spend_detail",
    "parse_iam_user_name",
]