"""Cloud cost analysis REST endpoints (production-grade FinOps).

Mounted under ``/api/v1/cmdb/cost``. Reads from the
``cmdb_cost_facts`` table (populated by ``cmdb.cost_facts_sync``)
and the ``cmdb_cost_allocation_rules`` chargeback ladder.

Surface:

  * ``GET  /breakdown`` — aggregate by one of (service / resource_group
    / resource_id / region / cost_center / department / owner /
    integration) over a date window. Returns top-N + "other" + an
    explicit "unallocated" KPI.
  * ``GET  /resources`` — top resources by cost, paginated, with a
    14-day spark series per row. Powers the "Resources" tab + drill-
    down drawer.
  * ``GET  /timeseries`` — daily series broken out by ``group_by``
    with top-N + "other" buckets, zero-filled.
  * ``GET  /forecast`` — month-end projection (linear vs trailing-
    14d blend) with MTD context.
  * ``GET  /anomalies`` — IQR-based per-resource spike detection.
  * ``GET/POST/PATCH/DELETE  /allocation-rules`` — chargeback rule
    CRUD.
  * ``POST /allocation-rules/recompute`` — re-stamp owner columns
    on the last 90 days of facts after a rule edit.
  * ``GET  /export.csv`` — bulk dump of fact rows for finance.

RBAC: ``can_view_finance`` (IT admin / cloud admin / finance /
auditor) for reads; ``can_manage_cloud`` for allocation-rule CRUD
and recompute.
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import uuid
import weakref
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import asynccontextmanager
from datetime import UTC, date, datetime, timedelta
from typing import Annotated, Any

from fastapi import APIRouter, Depends, Path, Query, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.config import settings
from app.core.database import get_db
from app.core.dependencies import (
    check_sensitive_endpoint_rate_limit,
    get_current_user,
    get_real_ip,
)
from app.core.dependencies import (
    require_active_employee_for_mutations as _leaver_runtime_gate,
)
from app.core.exceptions import (
    AppException,
    ForbiddenError,
    NotFoundError,
    ValidationError,
)
from app.core.rbac import can_manage_cloud, can_view_finance
from app.core.response import success_response
from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.cost_service import (
    GROUP_BY_COLUMNS,
    breakdown,
    create_allocation_rule,
    delete_allocation_rule,
    detect_anomalies,
    forecast_month_end,
    list_allocation_rules,
    list_tag_keys,
    recompute_allocation,
    stream_facts_csv,
    subordinates_cost_breakdown,
    sync_cost_facts_for_integration,
    timeseries,
    top_resources,
    update_allocation_rule,
)

logger = get_safe_logger(__name__)


router = APIRouter(
    prefix="/api/v1/cmdb/cost",
    tags=["cmdb-cost"],
    dependencies=[Depends(_leaver_runtime_gate)],
)


def _trace_id(request: Request) -> str:
    """Resolve the request's TraceID middleware-stamped value, or
    mint a fresh one for audit rows on requests where the middleware
    didn't run (mostly tests)."""
    return getattr(request.state, "trace_id", None) or str(uuid.uuid4())


def _validate_optional_integration_id(integration_id: str | None) -> None:
    """Round-49: validate ``integration_id`` query strings against a
    UUID shape before they hit the ``String(36)`` filter column.

    SQLAlchemy already parameterises queries (so this is not a SQL
    injection gap), but a 1 MiB garbage query string flooded into
    ``CmdbCostFact.integration_id == ?`` always returns zero rows
    and floods the slow-query log — the canonical DoS shape the
    Round 22 ``subordinates_breakdown`` fix already addressed.
    Six endpoints (``/breakdown``, ``/timeseries``, ``/forecast``,
    ``/anomalies``, ``/resources``, ``/breakdown/by-fact``) lacked
    this validation; ``/export.csv`` and ``/backfill`` had it.
    """
    if integration_id is None:
        return
    try:
        uuid.UUID(integration_id)
    except (TypeError, ValueError) as exc:
        raise AppException(
            code="BAD_INTEGRATION_ID",
            message="integration_id must be a valid UUID",
            status_code=400,
        ) from exc


def _validate_optional_scope_id(scope_id: str | None) -> None:
    """Validate the ``scope_id`` (subscription) query string against a
    UUID shape before it reaches the ``String(36)`` filter column.

    Same DoS rationale as :func:`_validate_optional_integration_id` — a
    garbage scope_id string floods the slow-query log with zero-row
    scans. ``scope_id`` powers the "isolate one subscription" filter on
    the cloud-overview drilldown (the per-Foundry view path)."""
    if scope_id is None:
        return
    try:
        uuid.UUID(scope_id)
    except (TypeError, ValueError) as exc:
        raise AppException(
            code="BAD_SCOPE_ID",
            message="scope_id must be a valid UUID",
            status_code=400,
        ) from exc


def _parse_window(
    start: date | None, end: date | None, *, max_days: int = 365,
) -> tuple[date, date]:
    """Parse + validate a date-range query.

    Defaults to the last 30 days if neither is set; clamps the
    window to ``max_days`` so an open-ended request can't sweep
    years of data and stall the dashboard.
    """
    end_dt = end or (datetime.now(UTC).date() - timedelta(days=1))
    start_dt = start or (end_dt - timedelta(days=29))
    if end_dt < start_dt:
        raise AppException(
            code="BAD_DATE_RANGE", message="end < start",
            status_code=400,
        )
    if (end_dt - start_dt).days > max_days:
        raise AppException(
            code="WINDOW_TOO_LARGE",
            message=f"max {max_days}-day window per request",
            status_code=400,
        )
    return start_dt, end_dt


def _resolve_measure_or_403(measure: str | None, user: AuthUser) -> str:
    """口径路由:默认业务折前成本(list,研发口径);``invoice``(实付)
    仅保密圈(finance / cloud_admin / global_admin)。无权限请求 invoice
    → 403,绝不静默降级(保密圈用户误把研发口径当实付更危险);未知值
    → 400。auditor / management 虽可进本页(can_view_finance),但在
    财务确认扩圈前只看研发口径。"""
    from app.domains.cmdb.cost_view import resolve_measure

    try:
        resolved = resolve_measure(measure, user.role_names)
    except ValueError as exc:
        raise AppException(
            code="UNKNOWN_MEASURE", message=str(exc), status_code=400,
        ) from exc
    if resolved is None:
        raise ForbiddenError()
    return resolved


_MEASURE_QUERY = Query(
    None,
    description=(
        "成本口径:``list``(默认,业务折前成本/研发口径)或 ``invoice``"
        "(实付,仅 finance / cloud_admin / global_admin)。"
    ),
)


def _read_cache_key(
    surface: str, *, measure: str, params: dict[str, Any],
) -> str:
    """Stable Redis key for a cost read: surface + measure + a hash of
    EVERY input that varies the response body.

    保密红线:``measure`` 既进明文前缀也进哈希 —— 一个角色绝不能吃到
    另一个口径的缓存体。本页的读接口在 ``can_view_finance`` 之后只按
    ``measure`` 这一个 RBAC 派生维度分流,其余全是请求参数(窗口 /
    group_by / integration / scope / department / top);任何新增的按人
    收窄过滤必须同时进这个键,否则就是跨角色泄露。
    """
    signature = hashlib.sha256(
        json.dumps(
            {"measure": measure, **params}, sort_keys=True, default=str,
        ).encode("utf-8"),
    ).hexdigest()[:32]
    return f"cost:{surface}:v1:{measure}:{signature}"


# ── Fan-out admission control (2026-07-29 review, F4) ────────────────
#
# 每个分块并发的成本读(``/breakdown`` / ``/timeseries`` /
# ``/finance/waterfall``)除了自己的请求 session,还会额外开最多
# ``cost_service._CHUNK_CONCURRENCY``(3)个池化 session。此前没有任何东西
# 约束**并发请求数**:N 个财务用户同时开 365 天窗口 → 最多 4N 条连接,而池
# 只有 ``pool_size=10 + max_overflow=20 = 30``(``app/core/database.py``)。
# 8 个并发读就能抽干池子,连登录一起卡死。
#
# 闸门只管**分块扇出**这一段。宽度按**整个请求**的连接数算,不是只算「额外」
# 的分块 session:一个准入的读同时占 1 条请求 session(闸内还会用到 ``db``)
# + 最多 3 条分块 session = 4 条。3 × 4 = 12,池(30)还剩 18 条给普通请求
# 路径 —— 登录的角色 SELECT 必须永远拿得到连接。此前写 4 是按「额外连接」
# 记账的(4 × 3 = 12 看着一样),按整请求算其实是 16/30,已经过半。
# 常态零成本:30 分钟响应缓存吸收重复流量,**缓存命中根本不进这个闸**;同键
# 并发由单飞收敛成一次计算,所以宽度 3 不等于「同时只能 3 个人看财务页」。
_COST_FANOUT_MAX_CONCURRENCY = 3
# 排队上限(2026-07-29 复审 F1)。排队本身不占连接(见
# ``_release_request_connection``),但也不许无限排:20s 还没拿到名额说明
# 前面至少还有一整波在算,与其把请求挂到 nginx / 浏览器超时(客户端看到的
# 是一个没有解释的失败),不如诚实地 503 + Retry-After。20s 严格小于
# ``database._POOL_TIMEOUT_SECONDS``(30s),所以过载先在这里被卸掉,而不是
# 退化成「QueuePool timed out」——后者会打在任何一个恰好要取连接的请求上,
# 包括登录的角色 SELECT。
# 20s **短于**最坏的单次扇出(~25s),所以闸满时确实可能在第一波还没算完就
# 卸掉一个请求 —— 这是刻意的,不是漏算:闸量的是**不同键**的真实容量过载,
# 它的正确出口是尽早的 503 + Retry-After 而不是排更久(把预算加到 25s+ 只是
# 让过载的队伍更长)。成本读的前端超时已是 90s 专用值
# (``web/src/api/cost.ts`` 的 ``COST_FANOUT_TIMEOUT_MS``,全局 30s 不适用),
# 所以「诚实的 503」到得了用户眼前。真正不许发生的是**同键**的旁观者互相把
# 队排满 —— 那由单飞收敛(见下方)。
_COST_FANOUT_ACQUIRE_TIMEOUT_SECONDS = 20.0
# 实测最坏一次扇出(24 个月 / 365 天窗口,分块并发 3)的墙上时钟。写成常量而
# 不是只留在注释里:单飞的锁 TTL 与等待上限都必须**长过**它,否则旁观者会在
# 持有者还在算的时候放弃、各自重算一遍,并把后面的排队者顶成 503(2026-07-29
# 三审的 P1 就是 15s 的等待守着 25s 的计算)。
# ``test_singleflight_bounds_are_longer_than_the_compute_they_wait_for`` 盯着
# 这两个不等式,所以这个数与那两个常量不可能再各自漂开。
_COST_WORST_CASE_FANOUT_SECONDS = 25.0
# Per-loop, mirroring ``scheduled_jobs.service._job_concurrency_gate``: a
# module-level semaphore binds to the loop that first blocks on it, so a
# process with more than one loop (pytest-asyncio makes a fresh one per test)
# would raise "bound to a different event loop" the moment it saturates.
# WEAK keys: a strong dict pins every loop that ever reached this gate for
# the life of the process (a fresh loop per test → one dead loop + semaphore
# retained per test), and a retained loop keeps its own callbacks/handles
# alive with it.
_COST_FANOUT_GATES: weakref.WeakKeyDictionary[
    asyncio.AbstractEventLoop, asyncio.Semaphore,
] = weakref.WeakKeyDictionary()


def _cost_fanout_gate() -> asyncio.Semaphore:
    loop = asyncio.get_running_loop()
    gate = _COST_FANOUT_GATES.get(loop)
    if gate is None:
        gate = asyncio.Semaphore(_COST_FANOUT_MAX_CONCURRENCY)
        _COST_FANOUT_GATES[loop] = gate
    return gate


async def _release_request_connection(db: AsyncSession | None) -> None:
    """把请求 session 的池连接交还给池,**然后**才去等待。

    ``get_current_user``(``app/core/dependencies.py``)与端点共用同一个
    session,它的角色 SELECT 已经 autobegin 了事务 —— 连接从此被这个请求钉
    住,直到 ``get_db`` 收尾。等待时还钉着,闸门本身就是死锁源:准入者
    ×4 条 + 排队者 ×1 条,~18 个并发读就凑满整池 30
    (``database.pool_size=10 + max_overflow=20``),于是准入者拿不到分块连
    接、排队者又在等准入 —— 环形等待,只能等 ``pool_timeout``(30s)崩开,
    连登录一起 500(2026-07-29 复审 F1 的实测形状)。commit 之后排队者占
    **0** 条连接,环就断了。

    只读端点上 commit 是安全的:三个扇出读在这个事务里不写任何东西,后续也
    不依赖同一个事务(``db`` 下次被用到时重新取一条连接);分块本来就跑在
    独立 session / 独立事务里(``cost_service._run_month_chunks``),所以
    「请求事务里的未提交行对分块可见」从来不是这条路径的语义。
    """
    if db is not None:
        await db.commit()


@asynccontextmanager
async def _cost_fanout_admission(
    db: AsyncSession | None,
) -> AsyncIterator[None]:
    """扇出准入:先交还连接,再有界排队。

    三个扇出读(``/breakdown`` / ``/timeseries`` / ``/finance/waterfall``)
    只能通过这里进入分块段 —— 见 ``_COST_FANOUT_MAX_CONCURRENCY`` 与
    ``_release_request_connection`` 各自守的不变量。
    """
    await _release_request_connection(db)
    gate = _cost_fanout_gate()
    try:
        await asyncio.wait_for(
            gate.acquire(), _COST_FANOUT_ACQUIRE_TIMEOUT_SECONDS,
        )
    except TimeoutError as exc:
        # 「忙,请重试」而不是「算失败了」:数据没问题,没有任何写入发生,
        # 重试一次就会命中前一波刚回写的缓存。
        raise AppException(
            code="COST_READ_BUSY",
            message=(
                "Cloud-cost aggregation is at capacity right now — nothing "
                "failed and nothing changed; retry in a few seconds."
            ),
            status_code=503,
            headers={
                "Retry-After": str(
                    int(_COST_FANOUT_ACQUIRE_TIMEOUT_SECONDS),
                ),
            },
        ) from exc
    try:
        yield
    finally:
        gate.release()


# 单飞(2026-07-29 复审 F1;等待判据 2026-07-29 三审重做):响应缓存 TTL 30
# 分钟、又没有单飞,过期那一刻所有在线财务用户同时 miss —— N 个请求算同一个
# 键,全部去抢扇出名额。等值的旁观者应该等**持有者**的结果。
#
# 旁观者的放弃判据是「锁还在不在」,不是墙上时钟。上一版是固定 15s,而它等
# 的那次计算实测最坏 ~25s:8 个人打同一个冷键时,7 个旁观者在持有者算完前
# 全部放弃、各自去抢扇出闸(宽度 3、排队上限 20s),于是既跑了 8 份重复聚合,
# 又把其中至少 2 个人顶成 503 —— 而这些请求在没有单飞的年代只是排队然后成
# 功。等待上限**必须**长过它守的那次计算,否则单飞在它唯一存在的意义(同键
# 冷读)上是负收益。
#
# 有界且自愈:
#   * 锁不 ``auto_renew``,所以它最迟在取锁后 TTL 秒消失 —— 持有者被 kill /
#     部署打断也一样。于是「锁还在」⇒「有人还在算」是可信判据
#     (``distributed_lock.is_lock_held``),旁观者不会在持有者明明还在干活
#     的时候放弃;
#   * 锁没了、缓存还是空的(持有者报错 / 猝死)⇒ 回到外层**再抢一次锁**:N
#     个旁观者里仍然只有一个去算,其余继续等它。死持有者既不会把读面挂住,
#     也不会退化成 N 份重复扇出 —— 后者正是把并发读顶成 503 的形状;
#   * 等待预算 ``_COST_SINGLEFLIGHT_MAX_WAIT_SECONDS`` 按**锁纪元**计,不是
#     终身一次性(五审 F-1):死持有者的锁要到 TTL 才消失,接手者从那一刻才
#     开始算 ~25s —— 终身预算只给接手纪元留 5s,其余旁观者会在接手者明明在
#     算的时候集体落进本地兜底,3 个挤进宽度 3 的闸、其余 503,恰好复刻单飞
#     要消灭的踩踏。所以每次回到外层重抢(= 新纪元)都续满预算;
#   * 纪元数封顶 ``_COST_SINGLEFLIGHT_MAX_TAKEOVERS``:连续多任持有者全部
#     猝死是基础设施灾难,不值得再信锁,兜底自己算(仍然过扇出闸——连接
#     封顶是 F4 的不变量);
#   * 等待期间旁观者不占池连接(``_release_request_connection`` 已交还)。
#
# 自己算的那个旁观者仍然走 ``_cost_fanout_admission`` 的 20s 预算,不给它加
# 时:走到那一步时同键只有它一个在算(上面第二条),20s 预算面对的是**别的
# 键**在占闸,那是真实的容量过载。浏览器侧:成本扇出读有 90s 专用超时
# (``web/src/api/cost.ts`` 的 ``COST_FANOUT_TIMEOUT_MS``,不是全局 30s),
# 盖得住健康冷键的最长路径(20s 排队 + ~25s 计算 ≈ 45s;旁观者 ~25s)。
# 死持有者的最长尾部(等满 TTL 65s + 排队 + 计算 ≈ 110s)会先被前端 abort
# —— 已接受:缓存仍会回填,用户重试即命中,不为极端尾部加长全体预算。
#
# TTL 60s:实测最慢的扇出(24 个月 / 365 天窗口)~25s,留一倍余量 —— 锁必须
# 活过它守的那次计算(否则旁观者会在持有者还在算时看到「没人在算」),又不许
# 在计算早就结束后继续挡着(``distributed_lock`` 的 finally 会主动 CAS-删除,
# TTL 只是持有者猝死时的兜底)。
_COST_SINGLEFLIGHT_LOCK_TTL_SECONDS = 60
# 从 TTL 派生,不是第二个手写常量:两个数各自漂移就又回到「等得比算得短」。
# ``+ 5`` 是让锁的过期先发生、旁观者最后一轮轮询还看得见「锁没了」。
_COST_SINGLEFLIGHT_MAX_WAIT_SECONDS = float(
    _COST_SINGLEFLIGHT_LOCK_TTL_SECONDS + 5,
)
_COST_SINGLEFLIGHT_POLL_SECONDS = 0.25


def _epoch_deadline(loop: asyncio.AbstractEventLoop) -> float:
    """本次锁纪元的等待截止点 = 现在 + 满额预算。

    每进入一次等待就调用一次 —— 这正是「按纪元续期」与「终身一次性
    预算」的区别所在(五审 F-1),所以它是一个具名函数而不是内联算式:
    调用**次数**是这条性质的确定性观测点,回归测试据此断言,不必再去
    赌墙上时钟的窗口(而那种窗口受 ``预算 < 死锁存活 + 接手计算`` 的
    约束,挤不出能扛住 CI 抖动的余量 —— 试过两次,都在 CI 上翻)。
    """
    return loop.time() + _COST_SINGLEFLIGHT_MAX_WAIT_SECONDS
# 连续接手纪元的封顶(每个纪元 = 一任持有者 + 满额等待预算)。1 任猝死是
# 部署/OOM 的正常形状;连着 3 任全死说明 Redis/DB 层面已经着火,继续等锁
# 没有意义 —— 兜底本地算,让扇出闸做最后的容量防线。
_COST_SINGLEFLIGHT_MAX_TAKEOVERS = 3


async def _cached_read(
    cache_key: str,
    compute: Callable[[], Awaitable[Any]],
    *,
    db: AsyncSession | None = None,
) -> Any:
    """财务瀑布同款响应缓存 + 单飞:命中即返回,miss 只让一个请求现算。

    Redis 故障降级为直查,绝不 500(缓存故障不许拖垮页面)——锁拿不到、
    读不到、写不进都一样降级。``COST_READ_CACHE_TTL_SECONDS = 0`` 关闭缓存
    (测试默认),此时连单飞一起跳过。

    ``db`` 传进来只为一件事:进入任何等待前把池连接交还(见
    ``_release_request_connection``)。
    """
    ttl = settings.COST_READ_CACHE_TTL_SECONDS
    if ttl <= 0:
        return await compute()

    from app.core.redis_client import get_redis

    async def _read_cache() -> Any | None:
        try:
            raw = await get_redis().get(cache_key)
        except Exception:  # noqa: BLE001 — 缓存故障绝不拖垮页面
            return None
        return json.loads(raw) if raw else None

    async def _compute_and_store() -> Any:
        data = await compute()
        try:
            await get_redis().set(cache_key, json.dumps(data), ex=ttl)
        except Exception:  # noqa: BLE001
            pass
        return data

    cached = await _read_cache()
    if cached is not None:
        return cached

    await _release_request_connection(db)

    from app.core.distributed_lock import (
        LockBusy,
        distributed_lock,
        is_lock_held,
    )

    lock_key = f"cost_read:{cache_key}"
    loop = asyncio.get_running_loop()

    async def _await_holder(deadline: float) -> tuple[Any | None, bool]:
        """等同键持有者回写。返回 ``(命中的数据, 该不该再抢一次锁)``。

        判据是锁的存在性(见 ``_COST_SINGLEFLIGHT_*`` 上方):锁活着就说明
        持有者还在算,旁观者继续等;锁没了却还没有缓存,说明持有者报错或
        猝死,该由**一个**旁观者接手,所以回到外层重抢锁。``deadline`` 由
        调用方按**当前纪元**给出——不是终身预算(五审 F-1)。
        """
        while loop.time() < deadline:
            await asyncio.sleep(_COST_SINGLEFLIGHT_POLL_SECONDS)
            cached_now = await _read_cache()
            if cached_now is not None:
                return cached_now, False
            try:
                held = await is_lock_held(lock_key)
            except Exception:  # noqa: BLE001 — 探测失败按「没人在算」处理
                return None, True
            if not held:
                return None, True
        return None, False

    takeovers = 0
    while True:
        # 只有「取锁失败」才降级;进了 compute 之后的任何异常必须原样冒出去
        # (静默重算会把一次 400/DB 错误变成两次)。
        entered_compute = False
        try:
            async with distributed_lock(
                lock_key, ttl=_COST_SINGLEFLIGHT_LOCK_TTL_SECONDS,
            ):
                # 拿到锁再看一眼:排在锁上的那段时间里持有者可能刚回写。
                cached = await _read_cache()
                if cached is not None:
                    return cached
                entered_compute = True
                return await _compute_and_store()
        except LockBusy:
            pass
        except Exception:
            if entered_compute:
                raise
            logger.warning(
                "cost read single-flight lock unavailable for %s — computing "
                "directly", cache_key, exc_info=True,
            )
            return await _compute_and_store()

        # 有人正在算同一个键 —— 等它回写。预算按**这一任持有者**满额起算
        # (五审 F-1:终身一次性预算在死持有者场景只给接手纪元剩 5s,旁观者
        # 会在接手者还在算时集体落进本地兜底,复刻踩踏)。
        cached, retake_lock = await _await_holder(_epoch_deadline(loop))
        if cached is not None:
            return cached
        if retake_lock:
            takeovers += 1
            if takeovers <= _COST_SINGLEFLIGHT_MAX_TAKEOVERS:
                continue
            logger.warning(
                "cost read single-flight saw %d consecutive dead holders for "
                "%s — the lock is no longer trustworthy; computing locally",
                takeovers, cache_key,
            )
            return await _compute_and_store()
        logger.info(
            "cost read single-flight still locked after %.0fs for %s — the "
            "holder outlived its own lock TTL; computing locally",
            _COST_SINGLEFLIGHT_MAX_WAIT_SECONDS, cache_key,
        )
        return await _compute_and_store()


def _waterfall_start_month(today: date, months: int) -> date:
    """首月(1 日)——``months`` 个自然月窗口含当前月。

    正经月算术:此前的 ``timedelta(days=31*(months-1))`` 再 floor 到
    月初会「多退一个月」(月长 <31 天时跨过两个月边界),months=2 返回
    3 个月行、``le=24`` 的上限实际执行 25 个分块。
    """
    anchor = today.replace(day=1)
    total = anchor.year * 12 + (anchor.month - 1) - (months - 1)
    return date(total // 12, total % 12 + 1, 1)


@router.get("/breakdown")
async def breakdown_endpoint(
    request: Request,
    group_by: str = Query(
        ...,
        description=(
            "One of service / resource_group / resource_id / region / "
            "resource_type / cost_center / department / owner / integration."
        ),
    ),
    start: date | None = None,
    end: date | None = None,
    integration_id: str | None = None,
    scope_id: str | None = Query(
        None,
        description=(
            "Narrow to a single cloud subscription / account / project "
            "(``cmdb_cloud_scopes.id``). Combine with group_by=resource_id "
            "to see per-resource (e.g. per-Foundry) spend within one "
            "subscription. Routing mirrors the subscription billing tab."
        ),
    ),
    department: str | None = Query(
        None,
        max_length=128,
        description=(
            "Narrow to spend attributable to people in this department "
            "(by ``cost_owner_department``). Picking a parent rolls up its "
            "whole subtree. Unattributed facts (NULL owner) are excluded."
        ),
    ),
    top: int = Query(20, ge=1, le=200),
    measure: str | None = _MEASURE_QUERY,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    resolved_measure = _resolve_measure_or_403(measure, user)
    if group_by not in GROUP_BY_COLUMNS and group_by != "provider":
        raise AppException(
            code="UNKNOWN_GROUP_BY",
            message=(
                f"unknown group_by={group_by!r}; "
                f"supported: {sorted([*GROUP_BY_COLUMNS, 'provider'])}"
            ),
            status_code=400,
        )
    _validate_optional_integration_id(integration_id)
    _validate_optional_scope_id(scope_id)
    start_dt, end_dt = _parse_window(start, end)
    # 与财务瀑布同一套执行策略(2026-07-29):按月分块 + 分块间有界并发
    # (独立 session)+ Redis 响应缓存。本端点接受 365 天窗口 → 12 个
    # 分块 × 4 个聚合,顺序执行是分块前单查询延迟的 ~12 倍。
    from app.core.database import async_session_factory

    # ``excluded_buckets`` 的金额按保密圈开放(见
    # ``cost_service.excluded_bucket_summary``)。这是本端点第二个按角色
    # 变体的维度,所以**必须**进缓存键 —— 否则一个 finance 用户填的缓存
    # 会把非 usage 桶的实付额喂给同样请求 ``measure=list`` 的研发用户,
    # 正是 ``_read_cache_key`` docstring 点名的跨角色泄露。
    from app.domains.cmdb.cost_view import can_view_invoice_measure

    excluded_amounts_visible = can_view_invoice_measure(user.role_names)
    cache_key = _read_cache_key("breakdown", measure=resolved_measure, params={
        "start": start_dt, "end": end_dt, "group_by": group_by,
        "integration_id": integration_id, "scope_id": scope_id,
        "department": department, "top": top,
        "excluded_amounts": excluded_amounts_visible,
    })

    async def _compute() -> dict[str, Any]:
        # 只在真的要扇出(cache miss)时排队 —— 见 ``_cost_fanout_admission``。
        async with _cost_fanout_admission(db):
            return await breakdown(
                db,
                start=start_dt,
                end=end_dt,
                group_by=group_by,
                integration_id=integration_id,
                scope_id=scope_id,
                department=department,
                top=top,
                measure=resolved_measure,
                include_excluded_amounts=excluded_amounts_visible,
                session_factory=async_session_factory,
            )

    try:
        data = await _cached_read(cache_key, _compute, db=db)
    except ValueError as exc:
        raise AppException(
            code="UNKNOWN_SCOPE_ID",
            message=str(exc),
            status_code=400,
        ) from exc
    return success_response(
        data={
            "start": start_dt.isoformat(),
            "end": end_dt.isoformat(),
            "group_by": group_by,
            "measure": resolved_measure,
            **data,
        },
        request=request,
    )


@router.get("/resources")
async def top_resources_endpoint(
    request: Request,
    start: date | None = None,
    end: date | None = None,
    integration_id: str | None = None,
    department: str | None = Query(None, max_length=128),
    resource_group: str | None = None,
    search: str | None = None,
    tag_key: str | None = Query(
        None,
        description=(
            "Filter by a single user-defined tag key. Must be supplied "
            "together with tag_value."
        ),
    ),
    tag_value: str | None = Query(
        None,
        description=(
            "Filter value for tag_key. Exact match against the JSON "
            "scalar."
        ),
    ),
    limit: int = Query(100, ge=1, le=500),
    measure: str | None = _MEASURE_QUERY,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    from app.core.database import async_session_factory

    resolved_measure = _resolve_measure_or_403(measure, user)
    _validate_optional_integration_id(integration_id)
    start_dt, end_dt = _parse_window(start, end)
    try:
        rows = await top_resources(
            db,
            start=start_dt,
            end=end_dt,
            integration_id=integration_id,
            department=department,
            resource_group=resource_group,
            search=search,
            tag_key=tag_key,
            tag_value=tag_value,
            limit=limit,
            measure=resolved_measure,
            # Fan the month chunks out concurrently, each on its own
            # session — same execution strategy as the other chunked
            # reads in this router (#2133).
            session_factory=async_session_factory,
        )
    except ValueError as exc:
        raise AppException(
            code="INVALID_TAG_FILTER",
            message=str(exc),
            status_code=400,
        ) from exc
    return success_response(
        data={
            "start": start_dt.isoformat(),
            "end": end_dt.isoformat(),
            "measure": resolved_measure,
            "resources": rows,
        },
        request=request,
    )


@router.get("/resources/{integration_id}/{resource_id:path}/creator")
async def resource_creator_endpoint(
    request: Request,
    integration_id: str,
    resource_id: str,
    refresh: bool = Query(
        False,
        description=(
            "When true, bypass the cache and re-call the cloud audit "
            "API. Use sparingly — drains the audit API's rate budget."
        ),
    ),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Resolve "who created this resource" via the cloud provider's
    audit log (CloudTrail / Cloud Audit Logs / ActionTrail /
    Activity Log).

    Returns the cached attribution row when available; calls the
    cloud audit API on a cache miss + persists. ``refresh=true``
    forces a fresh API call regardless of cache state — gated by
    the same RBAC as the read path because the audit-API rate
    budget is shared org-wide.

    RBAC: ``can_view_finance`` OR ``can_read_cloud`` — finance
    needs this for chargeback, cloud admins need it for
    incident response / orphan-resource attribution.

    Audit-window caveats per provider are documented on
    :meth:`CloudDriver.get_resource_creator`; for AWS / Aliyun /
    Azure the default window is 90 days, GCP is 400.
    """
    from app.core.dependencies import apply_sensitive_rate_limit
    from app.core.rbac import can_read_cloud
    from app.domains.cmdb.resource_creator_service import (
        get_cached_creator,
        resolve_resource_creator,
    )

    if not (
        can_view_finance(user.role_names)
        or can_read_cloud(user.role_names)
    ):
        raise ForbiddenError()

    # ``resource_id`` is a path param with ``:path`` so it carries
    # forward slashes (Azure ARM ids, GCP resource paths). FastAPI
    # already URL-decodes once; ARM ids legitimately start with `/`
    # so we don't strip.
    if not resource_id:
        raise ValidationError("resource_id must be non-empty")
    if not integration_id:
        raise ValidationError("integration_id must be non-empty")
    # Defence-in-depth before any DB write: same shape as POST
    # /retry. Without this guard, a 600-char resource_id flows
    # through ``resolve_resource_creator`` → driver call (burns
    # audit-API quota) → ``_upsert_creator_row`` INSERT against
    # ``String(512)`` → asyncmy DataError → 500. Quota-burn-then-
    # 500 is exactly the failure mode Round 2 closed on POST /retry.
    if len(resource_id) > 512:
        raise ValidationError("resource_id exceeds 512 chars")

    # Audit-API rate-limit. The cooldown inside ``resolve_resource_
    # creator`` only protects the ``force=True`` (refresh) path when
    # ``last_attempt_at`` is populated; it does NOT protect a fresh
    # cache-miss + ``refresh=false`` from firing a driver call (see
    # resource_creator_service.py:395-547 flow — cached is None →
    # falls through to driver). So both (a) refresh=true and (b)
    # cache-miss-with-refresh-false burn provider quota. Cache HIT
    # with refresh=false is a pure SELECT and doesn't need rate-
    # limiting; we peek the cache first to keep dashboard browsing
    # (cache hit, the common case) un-throttled, and gate the
    # 10/h/user cap only when this call actually triggers a driver
    # call. Same Redis key as POST /retry / refresh=true / PAT-
    # reveal so a PAT holder can't bypass by switching surfaces.
    if refresh:
        await apply_sensitive_rate_limit(user.id)
    else:
        cached_peek = await get_cached_creator(db, integration_id, resource_id)
        if cached_peek is None:
            await apply_sensitive_rate_limit(user.id)

    try:
        row = await resolve_resource_creator(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
            force=refresh,
        )
    except LookupError as exc:
        raise NotFoundError(
            "Integration", integration_id,
        ) from exc

    # Audit: any explicit "who created X" lookup is a finance-/
    # security-sensitive read; the trace_id surfaces this user's
    # query in the audit log for compliance review.
    #
    # Do NOT include ``creator_email`` in the audit-row details
    # block — that would write the resolved creator's PII to a
    # row keyed on the CALLER's actor_user_id, drifting GDPR-
    # erasure boundaries (an erasure on the creator would leave
    # their email in this row indefinitely). ``resource_id``
    # already carries the (integration, resource_id) pair which
    # is sufficient to correlate this lookup to the result row.
    #
    # 2026-05-26: the prior call used ``AuditService.write_audit``
    # (static method) which does not exist on this class — every
    # successful GET fell through to AttributeError → 500 after
    # the audit-API call had already burnt quota. PR #466 shipped
    # the broken call; PR #508 corrects it to the established
    # ``AuditService(db).log(...)`` instance pattern used by every
    # other domain (sso / apikey / auditor / hr_tools / …).
    await AuditService(db).log(
        trace_id=_trace_id(request),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.resource_creator.lookup",
        status="success",
        resource_type="cmdb.resource_creator",
        resource_id=f"{integration_id}:{resource_id}"[:128],
        detail={
            "refresh": refresh,
            "result_status": row.status,
            "result_resolved": row.status == "resolved",
        },
    )

    from app.domains.cmdb.resource_creator_service import (
        serialize_resource_creator_row,
    )
    return success_response(
        data=serialize_resource_creator_row(row),
        request=request,
    )


@router.post(
    "/resources/{integration_id}/{resource_id:path}/creator/retry",
    # Audit-API burner — every call fires a fresh CloudTrail /
    # Cloud Logging / ActionTrail / Activity Log query against the
    # provider. The reset wipes the in-row cooldown by design, so
    # this dependency is the ONLY app-level protection against a
    # compromised finance/cloud-admin (or a curious operator
    # scripting it) draining provider rate budget in a loop. Same
    # 10/h/user cap as PAT reveal / asset PII reveal. MCP twin
    # enforces the same cap inline via ``apply_sensitive_rate_limit``.
    dependencies=[Depends(check_sensitive_endpoint_rate_limit)],
)
async def resource_creator_retry_endpoint(
    request: Request,
    integration_id: str,
    resource_id: str,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Operator force-retry — zero ``attempt_count`` + fire a fresh
    audit-API resolve.

    Why a separate endpoint instead of ``GET /creator?refresh=true``:
    when ``attempt_count`` has hit ``RESOURCE_CREATOR_MAX_ATTEMPTS``
    (5), the cron sweep's candidate-list excludes the row forever
    (see :func:`app.domains.cmdb.resource_creator_service.list_unresolved_resources`).
    A bare refresh fires one more driver call but doesn't restore
    cron eligibility — the row stays at attempt-cap and the next
    cron tick still skips it. ``POST /creator/retry`` resets the
    counter so subsequent cron ticks pick up the row if needed, AND
    also fires an immediate ``resolve_resource_creator(force=True)``
    so the operator sees the result in the same response.

    Idempotent — re-calling is harmless (resets again + re-resolves).
    Returns the same shape as ``GET /creator``.

    RBAC: this is a WRITE (resets attempt_count + fires a provider
    audit-API call), so it gates on the *operational* cloud/finance
    roles — ``can_manage_cloud`` (cloud_admin / global_admin) OR finance
    (who owns cost attribution) — NOT the read gate ``GET /creator``
    uses. The read gate (``can_view_finance OR can_read_cloud``) admits
    the read-only governance roles ``auditor`` / ``management``, which
    have no business resetting counters or triggering provider calls.
    Cost: one audit-API call (worth it; the alternative is a DB-level
    intervention).
    """
    from app.core.rbac import FINANCE_ROLE, can_manage_cloud, has_role
    from app.domains.cmdb.resource_creator_service import (
        reset_and_resolve_resource_creator,
    )

    if not (
        can_manage_cloud(user.role_names)
        or has_role(user.role_names, FINANCE_ROLE)
    ):
        raise ForbiddenError()

    if not resource_id:
        raise ValidationError("resource_id must be non-empty")
    if not integration_id:
        raise ValidationError("integration_id must be non-empty")
    # Defence-in-depth — MCP twin enforces UUID shape via
    # ``_validate_uuid``; HTTP needs the same so a non-UUID
    # integration_id surfaces as 400 (not a deeper 500 when the
    # downstream UPDATE / SELECT errors on the asyncmy bind).
    try:
        uuid.UUID(integration_id)
    except (TypeError, ValueError) as exc:
        raise ValidationError(
            "integration_id must be a UUID",
        ) from exc
    # ``resource_id`` is a path param backed by a ``String(512)``
    # column (cmdb_resource_creators.resource_id). Reject 513+ char
    # inputs at the door so the asyncmy DataError becomes a clean
    # 400 instead of a 500 after the row UPDATE has already partly
    # executed.
    if len(resource_id) > 512:
        raise ValidationError(
            "resource_id exceeds 512 chars",
        )

    try:
        row = await reset_and_resolve_resource_creator(
            db,
            integration_id=integration_id,
            resource_id=resource_id,
        )
    except LookupError as exc:
        raise NotFoundError(
            "Integration", integration_id,
        ) from exc

    await AuditService(db).log(
        trace_id=_trace_id(request),
        actor_type="user",
        actor_user_id=user.id,
        action="cmdb.resource_creator.retry",
        status="success",
        resource_type="cmdb.resource_creator",
        resource_id=f"{integration_id}:{resource_id}"[:128],
        detail={
            "result_status": row.status,
            "result_resolved": row.status == "resolved",
        },
    )

    from app.domains.cmdb.resource_creator_service import (
        serialize_resource_creator_row,
    )
    return success_response(
        data=serialize_resource_creator_row(row),
        request=request,
    )


@router.get("/finance/waterfall")
async def finance_waterfall_endpoint(
    request: Request,
    months: int = Query(6, ge=1, le=24),
    integration_id: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """财务折扣瀑布 —— 保密圈专属（finance / cloud_admin / global_admin）。

    每 (月份 × 云) 一行:牌价(public)、业务折前成本(list)、发票净额
    (invoice)、九桶发票金额分布,以及派生项:

    * ``engineer_savings`` = Σ(public − list) **逐行配对差**,只在同时
      携带两个分量的行上累计(spot / 阶梯——部门功劳;GCP 牌价==list
      恒 ≈0 是口径事实)。整月无配对行 → ``None``(口径未接入),绝不
      伪造 0 —— 0 是「一分没省」这个断言。
    * ``public_total`` / ``savings_list_base`` 是该差值的两个基数,取
      **同一配对子集**,所以三个数闭合:
      ``public_total − engineer_savings == savings_list_base``。前端把
      这三列相邻展示,财务据此对账。该等式在本接口输出的 4 位小数精度上
      **精确**成立(不是「约等于」):三列都从同一批未取整累加值出发,
      两个基数各 round 一次,节省项由它们相减得出 —— 每列各自 round 会
      让等式漂 ≤1e-4。
    * ``list_total`` 是**全组** Σlist(含只有 list、没有 public 覆盖的
      行,如 AWS SG 的 CE 取数),因此 ``list_total`` 一般 ≥
      ``savings_list_base``;``contract_inprice_discount`` 与「成本口径」
      对数区都按它计价。
    * ``contract_inprice_discount`` = list_total − usage 桶发票额(GCP /
      阿里云的「价内合同折扣」——不是独立账单行,只能派生)。

    发票合计 = Σ桶,与云厂商账单对账(注意 AWS 的 CUR 导出不含
    payer,Support / AWS 侧 credit 多在 payer——对账须计入该范围差,
    见「成本口径」说明页)。

    执行策略（2026-07-28 生产 8175 修复）：**按月分块**（单查询窗口
    ≤1 个月，封顶 TiDB 单 SQL 内存——一次 GROUP BY 扫 6 个月 ~550 万
    行曾打爆 serverless 内存配额）+ 分块间有界并发（独立 session）+
    Redis 响应缓存 + 同键单飞（``_cached_read``：30 分钟——月度对账页
    容忍该滞后；Redis 故障降级为直查，绝不 500；同一键的并发读只算一
    次）。覆盖索引 ix_facts_date_measure（迁移 431）使每月分块走
    IndexRangeScan。
    """
    import asyncio as _asyncio

    from app.core.database import async_session_factory
    from app.domains.cmdb.cost_service import (
        integration_provider_map,
        monthly_measure_rollup,
    )
    from app.domains.cmdb.cost_view import can_view_invoice_measure

    if not can_view_invoice_measure(user.role_names):
        raise ForbiddenError()
    _validate_optional_integration_id(integration_id)

    today = datetime.now(UTC).date()
    start = _waterfall_start_month(today, months)

    # v3 (2026-07-29):public_total 收窄到配对子集 + 新增
    # savings_list_base(节省项的闭合基数)。语义变了就换代缓存键,
    # 避免 30 分钟内继续吐旧算法的值(v2 = 月窗口修正 + 逐行配对差)。
    cache_key = (
        f"cost:waterfall:v3:{months}:{integration_id or 'all'}:"
        f"{today.isoformat()}"
    )

    async def _compute() -> dict[str, Any]:
        # 月份窗口列表(含当前月),旧→新。
        windows: list[tuple[date, date]] = []
        cursor = start
        while cursor <= today:
            nxt = (cursor + timedelta(days=32)).replace(day=1)
            windows.append((cursor, nxt))
            cursor = nxt

        provider_by_integration = await integration_provider_map(db)

        # 有界并发:每个分块用独立 session(同一 AsyncSession 不能并发),
        # 并发度 3 —— 实测 TiDB serverless 上并发抢 RU 会拖慢单块,3 是
        # 提速与互踩之间的平衡点;24 个月最坏 ~25s,由 Redis 缓存吸收。
        sem = _asyncio.Semaphore(3)

        async def _one_month(ms: date, me: date):
            async with sem:
                async with async_session_factory() as chunk_session:
                    return ms, await monthly_measure_rollup(
                        chunk_session,
                        month_start=ms, month_end=me,
                        integration_id=integration_id,
                    )

        # 与 ``/breakdown`` / ``/timeseries`` 共用同一道扇出闸 —— 三个端点抢的
        # 是同一个连接池,分开计数等于没封顶。``_cost_fanout_admission`` 在排队前
        # 交还请求连接:``provider_by_integration`` 是纯 dict(无 ORM 实例),
        # 上面这段查询的结果不依赖事务继续开着。
        async with _cost_fanout_admission(db):
            chunk_results = await _asyncio.gather(
                *[_one_month(ms, me) for ms, me in windows],
            )

        by_key: dict[tuple[str, str], dict[str, Any]] = {}
        for month_start_d, rows in chunk_results:
            month = month_start_d.strftime("%Y-%m")
            for r in rows:
                provider = provider_by_integration.get(
                    str(r.integration_id), "unknown",
                )
                key = (month, provider)
                entry = by_key.setdefault(key, {
                    "month": month, "provider": provider,
                    "buckets": {}, "invoice_total": 0.0,
                    "list_total": None, "public_total": None,
                    "savings_list_base": None,
                })
                # 累加**不取整**,一律留到出行时一次性 round —— 逐笔取整会让
                # 相邻三列的闭合式各自漂 ≤1e-4(2026-07-29 复审 F5:两个基数逐
                # 桶 round、节省项只在末尾 round 一次,财务照 tooltip 的
                # 「牌价 − 节省 = 折前成本(牌价覆盖段)」去对就是加不起来)。
                inv = float(r.invoice or 0)
                entry["buckets"][str(r.cost_bucket)] = (
                    entry["buckets"].get(str(r.cost_bucket), 0.0) + inv
                )
                if r.list_sum is not None:
                    entry["list_total"] = (
                        entry["list_total"] or 0.0
                    ) + float(r.list_sum)
                # 展示用的牌价合计取**配对子集**(paired_public_sum),与节省
                # 项、savings_list_base 同一覆盖面 —— 三个数才加得起来。只有
                # public 没有 list 的行不进入任何断言(当前每个 driver 的
                # public 都与 list 同源同生,该行数在生产上恒为 0)。
                if r.paired_public_sum is not None:
                    entry["public_total"] = (
                        entry["public_total"] or 0.0
                    ) + float(r.paired_public_sum)
                if r.paired_list_sum is not None:
                    entry["savings_list_base"] = (
                        entry["savings_list_base"] or 0.0
                    ) + float(r.paired_list_sum)
        out_rows = []
        for entry in by_key.values():
            # 桶先定稿,发票合计取**已取整桶值之和**再 round —— 说明文案里的
            # 「发票合计 = Σ桶」在页面上必须字面成立。
            entry["buckets"] = {
                b: round(v, 4) for b, v in entry["buckets"].items()
            }
            entry["invoice_total"] = round(sum(entry["buckets"].values()), 4)
            pub = (
                round(entry["public_total"], 4)
                if entry["public_total"] is not None else None
            )
            base = (
                round(entry["savings_list_base"], 4)
                if entry["savings_list_base"] is not None else None
            )
            lst = (
                round(entry["list_total"], 4)
                if entry["list_total"] is not None else None
            )
            entry["public_total"], entry["savings_list_base"] = pub, base
            entry["list_total"] = lst
            usage_invoice = entry["buckets"].get("usage", 0.0)
            # engineer_savings = Σ(public − list) 只在同时携带两个分量的行
            # 上——不是 public_total − list_total:同一 provider 混合覆盖面
            # (AWS org BQ 带 public + AWS SG CE 只有 list)时,后者会把
            # SG 的整段 list 支出当成负节省。无配对行 → None(口径未接入),
            # 不伪造 0;``pub is not None`` 恒与 ``savings_list_base is not
            # None`` 同真(同一 CASE 条件),所以这一个判据够了。
            #
            # 取值从**已定稿的两个基数**相减,而不是另攒一路
            # ``Σsavings_sum``:rollup 侧 ``paired_public_sum −
            # paired_list_sum ≡ savings_sum``(同一 CASE 子集,见
            # ``cost_service.monthly_measure_rollup``),所以这不是换算法,而是
            # 让「三列闭合」由构造保证 —— 三个数各自 round 出来仍可差 1e-4。
            entry["engineer_savings"] = (
                round(pub - base, 4) if pub is not None and base is not None
                else None
            )
            entry["contract_inprice_discount"] = (
                round(lst - usage_invoice, 4) if lst is not None else None
            )
            out_rows.append(entry)
        out_rows.sort(key=lambda e: (e["month"], e["provider"]))
        return {"start_month": start.strftime("%Y-%m"), "rows": out_rows}

    # 与 ``/breakdown`` / ``/timeseries`` 同一套响应缓存 + 单飞(2026-07-29
    # 三审):本端点是最贵的那个扇出(24 个月实测 ~25s),此前只有自己的内联
    # 缓存,过期那一刻 N 个财务用户同时 miss 就是 N 份 25s 的重复聚合去挤宽度
    # 3 的闸 —— 缓存收窄了闸却没给它收敛同键并发,是纯粹的回退。语义与内联版
    # 一致(同一个键、同一个 TTL、Redis 故障降级为直查),多出来的只有单飞。
    payload = await _cached_read(cache_key, _compute, db=db)
    return success_response(data=payload, request=request)


@router.get("/methodology")
async def cost_methodology_endpoint(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """「成本口径」方法论说明页 —— 2026-07 成本口径改造 PR5。

    规则文字（每朵云的取数链路、白名单规则原文、排除项、禁用列、
    核验状态）来自 :mod:`app.domains.cmdb.cost_methodology` 的静态
    描述符 —— 只含规则与核验记录，不含实付金额、不泄露折扣率，
    ``can_read_cloud`` 即可读。

    「当月两口径对数区」（按 provider 的 list / invoice 合计与差值
    分桶）仅对保密圈（``cost_view.can_view_invoice_measure``）拼进
    响应；非保密圈的响应里 ``reconciliation`` 恒为 ``None`` ——
    保密红线：研发可见响应绝不同框两口径。
    """
    import json as _json

    from app.core.rbac import can_read_cloud
    from app.core.redis_client import get_redis
    from app.domains.cmdb.cost_methodology import (
        SHARED_BOUNDARIES,
        get_methodology_catalog,
    )
    from app.domains.cmdb.cost_service import (
        integration_provider_map,
        monthly_measure_rollup,
    )
    from app.domains.cmdb.cost_view import can_view_invoice_measure

    if not can_read_cloud(user.role_names):
        raise ForbiddenError()

    reconciliation: dict[str, Any] | None = None
    if can_view_invoice_measure(user.role_names):
        # 与财务瀑布同一查询单元(monthly_measure_rollup,单月分块
        # ——TiDB 8175 内存封顶的同款策略),粒度收窄到 provider ——
        # 对数区回答的是「这朵云这个月两口径合计差多少、差在哪些
        # 桶」。Redis 缓存 15 分钟,故障降级直查。
        today = datetime.now(UTC).date()
        month_start = today.replace(day=1)
        month_end = (month_start + timedelta(days=32)).replace(day=1)
        cache_ttl = min(settings.COST_READ_CACHE_TTL_SECONDS, 900)
        cache_key = f"cost:methodology:recon:v1:{month_start.isoformat()}"
        cached = None
        if cache_ttl > 0:
            try:
                cached = await get_redis().get(cache_key)
            except Exception:  # noqa: BLE001 — 缓存故障绝不拖垮页面
                cached = None
        if cached:
            reconciliation = _json.loads(cached)
        else:
            rows = await monthly_measure_rollup(
                db, month_start=month_start, month_end=month_end,
            )
            provider_by_integration = await integration_provider_map(db)
            by_provider: dict[str, dict[str, Any]] = {}
            for r in rows:
                provider = provider_by_integration.get(
                    str(r.integration_id), "unknown",
                )
                entry = by_provider.setdefault(provider, {
                    "provider": provider,
                    "buckets": {}, "invoice_total": 0.0, "list_total": None,
                })
                inv = float(r.invoice or 0)
                entry["buckets"][str(r.cost_bucket)] = round(
                    entry["buckets"].get(str(r.cost_bucket), 0.0) + inv, 4,
                )
                entry["invoice_total"] = round(
                    entry["invoice_total"] + inv, 4,
                )
                if r.list_sum is not None:
                    entry["list_total"] = round(
                        (entry["list_total"] or 0.0) + float(r.list_sum), 4,
                    )
            recon_rows = []
            for entry in by_provider.values():
                lst = entry["list_total"]
                entry["delta"] = (
                    round(entry["invoice_total"] - lst, 4)
                    if lst is not None else None
                )
                recon_rows.append(entry)
            recon_rows.sort(key=lambda e: e["provider"])
            reconciliation = {
                "month": month_start.strftime("%Y-%m"),
                "rows": recon_rows,
            }
            if cache_ttl > 0:
                try:
                    await get_redis().set(
                        cache_key, _json.dumps(reconciliation), ex=cache_ttl,
                    )
                except Exception:  # noqa: BLE001
                    pass

    return success_response(
        data={
            "clouds": get_methodology_catalog(),
            "shared_boundaries": list(SHARED_BOUNDARIES),
            "reconciliation": reconciliation,
        },
        request=request,
    )


@router.get("/tag-keys")
async def tag_keys_endpoint(
    request: Request,
    start: date | None = None,
    end: date | None = None,
    integration_id: str | None = None,
    department: str | None = Query(None, max_length=128),
    limit: int = Query(200, ge=1, le=500),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """List user-defined tag keys observed in the cost window, sorted
    by fact count (most-used first). Powers the dashboard's "filter
    by tag" autocomplete.
    """
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    _validate_optional_integration_id(integration_id)
    start_dt, end_dt = _parse_window(start, end)
    keys = await list_tag_keys(
        db,
        start=start_dt,
        end=end_dt,
        integration_id=integration_id,
        department=department,
        limit=limit,
    )
    return success_response(
        data={
            "start": start_dt.isoformat(),
            "end": end_dt.isoformat(),
            "keys": keys,
        },
        request=request,
    )


@router.get("/timeseries")
async def timeseries_endpoint(
    request: Request,
    group_by: str = Query("service"),
    start: date | None = None,
    end: date | None = None,
    integration_id: str | None = None,
    scope_id: str | None = Query(
        None,
        description=(
            "Narrow to a single cloud subscription / account / project "
            "(``cmdb_cloud_scopes.id``). Combine with group_by=resource_id "
            "for a per-resource (per-Foundry) daily series within one "
            "subscription."
        ),
    ),
    department: str | None = Query(None, max_length=128),
    top: int = Query(10, ge=1, le=50),
    measure: str | None = _MEASURE_QUERY,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    resolved_measure = _resolve_measure_or_403(measure, user)
    if group_by not in GROUP_BY_COLUMNS and group_by != "provider":
        raise AppException(
            code="UNKNOWN_GROUP_BY",
            message=(
                f"unknown group_by={group_by!r}; "
                f"supported: {sorted([*GROUP_BY_COLUMNS, 'provider'])}"
            ),
            status_code=400,
        )
    _validate_optional_integration_id(integration_id)
    _validate_optional_scope_id(scope_id)
    start_dt, end_dt = _parse_window(start, end)
    # 同 ``/breakdown``:按月分块 + 有界并发(独立 session)+ Redis 响应
    # 缓存。365 天窗口 = 12 个分块 × 3 个聚合。
    from app.core.database import async_session_factory

    cache_key = _read_cache_key("timeseries", measure=resolved_measure, params={
        "start": start_dt, "end": end_dt, "group_by": group_by,
        "integration_id": integration_id, "scope_id": scope_id,
        "department": department, "top": top,
    })

    async def _compute() -> dict[str, Any]:
        # 同 ``/breakdown``:只在 cache miss 的扇出段排队。
        async with _cost_fanout_admission(db):
            return await timeseries(
                db,
                start=start_dt,
                end=end_dt,
                group_by=group_by,
                integration_id=integration_id,
                scope_id=scope_id,
                department=department,
                top=top,
                measure=resolved_measure,
                session_factory=async_session_factory,
            )

    try:
        data = await _cached_read(cache_key, _compute, db=db)
    except ValueError as exc:
        raise AppException(
            code="UNKNOWN_SCOPE_ID",
            message=str(exc),
            status_code=400,
        ) from exc
    return success_response(
        data={
            "start": start_dt.isoformat(),
            "end": end_dt.isoformat(),
            "group_by": group_by,
            "measure": resolved_measure,
            **data,
        },
        request=request,
    )


@router.get("/forecast")
async def forecast_endpoint(
    request: Request,
    integration_id: str | None = None,
    department: str | None = Query(None, max_length=128),
    measure: str | None = _MEASURE_QUERY,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    resolved_measure = _resolve_measure_or_403(measure, user)
    _validate_optional_integration_id(integration_id)
    data = await forecast_month_end(
        db, integration_id=integration_id, department=department,
        measure=resolved_measure,
    )
    return success_response(
        data={"measure": resolved_measure, **data}, request=request,
    )


@router.get("/anomalies")
async def anomalies_endpoint(
    request: Request,
    days: int = Query(30, ge=7, le=120),
    integration_id: str | None = None,
    department: str | None = Query(None, max_length=128),
    min_impact_usd: float = Query(50.0, ge=0, le=1_000_000),
    z_threshold: float = Query(2.5, ge=0.5, le=10.0),
    # Round-49: hard upper bound matches ``/resources`` (max 500) so a
    # mis-specified client query can't pull 10 000+ rows. Default 200
    # is generous for a single dashboard view; the deterministic-
    # tiebreaker sort guarantees the same caller-visible top-N
    # regardless of truncation.
    limit: int = Query(200, ge=1, le=500),
    measure: str | None = _MEASURE_QUERY,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    resolved_measure = _resolve_measure_or_403(measure, user)
    _validate_optional_integration_id(integration_id)
    data = await detect_anomalies(
        db,
        days=days,
        integration_id=integration_id,
        department=department,
        min_impact_usd=min_impact_usd,
        z_threshold=z_threshold,
        limit=limit,
        measure=resolved_measure,
    )
    return success_response(
        data={
            "days": days,
            "z_threshold": z_threshold,
            "min_impact_usd": min_impact_usd,
            "limit": limit,
            "measure": resolved_measure,
            "anomalies": data,
        },
        request=request,
    )


@router.get("/subordinates")
async def subordinates_breakdown_endpoint(
    request: Request,
    start: date | None = None,
    end: date | None = None,
    integration_id: str | None = None,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Manager-scoped cost breakdown: self + every direct/transitive
    report. Powers the ``/my/cloud/department-cost`` page.

    Authorization: every active authenticated employee may call this.
    The response is automatically scoped to (caller + caller's
    subordinates) — a non-manager sees only their own allocated
    facts (equivalent to "my own personal cost summary"). No
    ``can_view_finance`` gate — the data the caller can see here is,
    by construction, **their own scope only**.

    Leaver guard: the BFS helper drops ``employment_status='left'``
    descendants but DOES include the root unconditionally — a
    leaver whose JWT hasn't expired could otherwise pull their own
    historical spend. We refuse the request explicitly when the
    caller is left so the platform-wide post-leave data lockout
    holds end-to-end. (The runtime ``require_active_employee_for_
    mutations`` dependency on this router is mutation-only; this is
    a read endpoint, hence the explicit guard here.)

    The subordinate population is resolved via the CANONICAL core
    manager-tree resolver so the scope semantics stay consistent across
    asset / cost / future surfaces. Skip-level reports (A → B → C; A
    asking sees C) are included.
    """
    # #1806: call the canonical ``app.core.person_scope`` resolver
    # DIRECTLY, not the asset domain's ROUTER-PRIVATE
    # ``_resolve_subordinate_ids`` (a cross-domain import of a
    # ``_``-prefixed router symbol — renaming/moving it in the asset
    # router would 500 this cloud-cost endpoint, and it silently couples
    # cmdb to asset routing internals in violation of Non-Negotiable Rule
    # #3). ``_resolve_subordinate_ids`` was itself only a thin
    # ``{root} ∪ subordinate_person_ids(..., ACTIVE_ONLY)`` wrapper, so
    # this reproduces its exact contract with no behaviour change.
    from app.core.person_scope import (
        SubordinateScope,
        subordinate_person_ids,
    )
    from app.identity.person_models import IdentityPerson

    # Refuse left + leaver_pending employees so a still-valid JWT can't
    # pull cost history during the offboarding window. ``user`` is the
    # lightweight ``AuthUser`` carrying role names + id — not the full
    # IdentityPerson — so we re-fetch the row to check
    # ``employment_status`` against the live state. Matches the
    # ``is_active_employee`` convention used elsewhere for leaver_pending
    # read lockout (audit P2-15).
    caller_person = await db.get(IdentityPerson, user.id)
    if caller_person is None or caller_person.employment_status in (
        "left", "leaver_pending",
    ):
        raise ForbiddenError()

    person_ids = {
        user.id,
        *await subordinate_person_ids(
            db, user.id, scope=SubordinateScope.ACTIVE_ONLY,
        ),
    }
    # Round-22 audit: tighten the default cap on `/subordinates`
    # (vs the 365-day cap on admin endpoints). Each call runs four
    # aggregation scans on ``cmdb_cost_facts``; a manager hitting
    # this endpoint at 365 days with 100 reports issues 400 scans on
    # a multi-million-row table. A 92-day (≈ one quarter) cap is the
    # right shape for the "my org's spend last quarter" use case
    # and bounds DB load.
    start_dt, end_dt = _parse_window(start, end, max_days=92)
    # Validate ``integration_id`` shape at the boundary so an
    # arbitrary string can't be passed into the SQL filter. Missing
    # validation here meant a 1 MiB query-string flooded the log.
    if integration_id is not None:
        import uuid as _uuid_local
        try:
            _uuid_local.UUID(integration_id)
        except ValueError:
            # ``ValidationError(message)`` only — its constructor pins
            # ``code="VALIDATION_ERROR"`` already. Passing a custom
            # ``code=`` kwarg raises ``TypeError`` at runtime, defeating
            # the validator. (Round-2 review caught the regression.)
            raise ValidationError(
                "integration_id must be a UUID",
            )
    data = await subordinates_cost_breakdown(
        db,
        person_ids=person_ids,
        start=start_dt,
        end=end_dt,
        integration_id=integration_id,
    )
    return success_response(
        data={
            "start": start_dt.isoformat(),
            "end": end_dt.isoformat(),
            "scope_person_count": len(person_ids),
            **data,
        },
        request=request,
    )


# ── Allocation rules CRUD ──────────────────────────────────────────


class AllocationRulePayload(BaseModel):
    """Body for create. All match-* fields are optional; omitting
    one means "match any". At least one assign-* field is required
    (the service layer enforces this AFTER composing the rule).
    """

    model_config = {"extra": "forbid"}

    name: str = Field(..., min_length=1, max_length=255)
    priority: int = Field(100, ge=1, le=9999)
    match_integration_ids: list[str] | None = None
    match_provider: str | None = Field(None, max_length=32)
    match_resource_group: str | None = Field(None, max_length=256)
    match_service: str | None = Field(None, max_length=128)
    match_tag_key: str | None = Field(None, max_length=128)
    match_tag_value: str | None = Field(None, max_length=256)
    assign_person_id: str | None = Field(
        None,
        # ``IdentityPerson.id`` is UUID-typed. Without a pattern the
        # service layer's ``session.get(IdentityPerson, "alice")``
        # raises an opaque asyncmy ``ValueError`` → FastAPI 500. Reject
        # at the boundary with a clean 422 instead.
        pattern=r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    )
    assign_department: str | None = Field(None, max_length=256)
    assign_cost_center: str | None = Field(None, max_length=64)
    enabled: bool = True
    description: str | None = Field(None, max_length=1000)


class AllocationRulePatchPayload(BaseModel):
    """Body for PATCH — every field optional. ``extra='forbid'``
    rejects sneaky field-spoofing (``created_by_id`` / ``id`` /
    ``updated_at``) with a 422 instead of silently dropping them.
    """

    model_config = {"extra": "forbid"}

    name: str | None = Field(None, min_length=1, max_length=255)
    priority: int | None = Field(None, ge=1, le=9999)
    match_integration_ids: list[str] | None = None
    match_provider: str | None = Field(None, max_length=32)
    match_resource_group: str | None = Field(None, max_length=256)
    match_service: str | None = Field(None, max_length=128)
    match_tag_key: str | None = Field(None, max_length=128)
    match_tag_value: str | None = Field(None, max_length=256)
    assign_person_id: str | None = Field(
        None,
        # ``IdentityPerson.id`` is UUID-typed. Without a pattern the
        # service layer's ``session.get(IdentityPerson, "alice")``
        # raises an opaque asyncmy ``ValueError`` → FastAPI 500. Reject
        # at the boundary with a clean 422 instead.
        pattern=r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    )
    assign_department: str | None = Field(None, max_length=256)
    assign_cost_center: str | None = Field(None, max_length=64)
    enabled: bool | None = None
    description: str | None = Field(None, max_length=1000)


@router.get("/allocation-rules")
async def list_allocation_rules_endpoint(
    request: Request,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    rules = await list_allocation_rules(db)
    return success_response(data={"rules": rules}, request=request)


@router.post("/allocation-rules")
async def create_allocation_rule_endpoint(
    request: Request,
    payload: AllocationRulePayload,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    try:
        rule = await create_allocation_rule(
            db,
            payload=payload.model_dump(exclude_unset=True),
            actor_id=str(user.id) if user.id else None,
        )
    except ValueError as e:
        raise AppException(
            code="INVALID_RULE", message=str(e), status_code=400,
        ) from e
    audit = AuditService(db)
    await audit.log(
        trace_id=_trace_id(request),
        actor_type="user",
        actor_user_id=user.id,
        action="cost.allocation_rule.created",
        status="success",
        resource_type="cmdb_cost_allocation_rule",
        resource_id=rule["id"],
        detail={
            "name": rule["name"],
            "priority": rule["priority"],
            "match_provider": rule.get("match_provider"),
            "match_resource_group": rule.get("match_resource_group"),
            "match_service": rule.get("match_service"),
            "match_tag_key": rule.get("match_tag_key"),
            "assign_person_id": rule.get("assign_person_id"),
            "assign_department": rule.get("assign_department"),
            "assign_cost_center": rule.get("assign_cost_center"),
        },
    )
    return success_response(data=rule, request=request)


@router.patch("/allocation-rules/{rule_id}")
async def update_allocation_rule_endpoint(
    request: Request,
    rule_id: Annotated[str, Path(
        # UUID-shape regex; rejects path-injection / typo'd ids with
        # 422 instead of letting the SELECT raise an opaque DB error.
        pattern=r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    )],
    payload: AllocationRulePatchPayload,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    diff_payload = payload.model_dump(exclude_unset=True)
    # Round-49: audit is written atomically inside the service so a
    # post-commit failure can't lose the pre/post change diff. The
    # router just threads trace_id + IP + UA through.
    try:
        rule = await update_allocation_rule(
            db,
            rule_id=rule_id,
            payload=diff_payload,
            actor_id=str(user.id) if user.id else None,
            trace_id=_trace_id(request),
            # Round-49: use get_real_ip (X-Forwarded-For aware) so the
            # forensic ip_address column the same PR was hardening
            # records the real client IP, not nginx's loopback peer.
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("user-agent"),
        )
    except LookupError as e:
        raise NotFoundError("CmdbCostAllocationRule", rule_id) from e
    except ValueError as e:
        raise AppException(
            code="INVALID_RULE", message=str(e), status_code=400,
        ) from e
    return success_response(data=rule, request=request)


@router.delete("/allocation-rules/{rule_id}")
async def delete_allocation_rule_endpoint(
    request: Request,
    rule_id: Annotated[str, Path(
        # UUID-shape regex; rejects path-injection / typo'd ids with
        # 422 instead of letting the SELECT raise an opaque DB error.
        pattern=r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    )],
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    # Round-49: service captures full pre-delete snapshot into audit
    # ``detail`` in the SAME transaction as the delete so the row can
    # be forensically reconstructed even after the cmdb_cost_allocation
    # _rules row is gone.
    try:
        snapshot = await delete_allocation_rule(
            db,
            rule_id=rule_id,
            actor_id=str(user.id) if user.id else None,
            trace_id=_trace_id(request),
            # Round-49: get_real_ip aware of X-Forwarded-For.
            ip_address=get_real_ip(request),
            user_agent=request.headers.get("user-agent"),
        )
    except LookupError as e:
        raise NotFoundError("CmdbCostAllocationRule", rule_id) from e
    return success_response(
        data={"deleted": rule_id, "snapshot": snapshot}, request=request,
    )


@router.post("/allocation-rules/recompute")
async def recompute_allocation_endpoint(
    request: Request,
    days: int = Query(90, ge=1, le=365),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    # Serialize with the nightly cron and MCP tool via the same lock
    # key. Without this, two admins double-clicking the "重算分摊"
    # button, or the cron tick firing while an admin holds the button,
    # re-stamp the same 90-day fact table in parallel and race on
    # ON-DUPLICATE-KEY-UPDATE writes — non-deterministic
    # ``allocation_rule_id`` / ``cost_owner_person_id`` per row.
    # ttl matches the cron handler (1800s) so the lock can never expire
    # mid-recompute on a large re-stamp and let a second holder in; a
    # client disconnect cancels the coroutine and releases the lock via
    # the ``async with`` unwind, so the long ttl only bounds hard-death
    # recovery.
    from app.core.distributed_lock import LockBusy, distributed_lock

    try:
        async with distributed_lock(
            "cost_recompute_allocation", ttl=1800,
        ):
            summary = await recompute_allocation(db, days=days)
    except LockBusy as exc:
        # 409 Conflict: another holder is mid-recompute. The button
        # already shows a confirm dialog, so admins re-clicking gets
        # a clean "wait, in-progress" instead of a duplicate run.
        from app.core.exceptions import ConflictError
        raise ConflictError(
            "Another recompute_allocation run is in progress; "
            "wait for it to finish before retrying.",
        ) from exc
    audit = AuditService(db)
    await audit.log(
        trace_id=_trace_id(request),
        actor_type="user",
        actor_user_id=user.id,
        action="cost.allocation.recomputed",
        status="success",
        resource_type="cmdb_cost_facts",
        resource_id=None,
        detail={"days": days, **summary},
    )
    return success_response(
        data={"days": days, **summary}, request=request,
    )


# ── Backfill trigger (admin) ───────────────────────────────────────


class BackfillPayload(BaseModel):
    """Body for the manual backfill endpoint.

    ``extra='forbid'`` rejects unknown fields (a hand-rolled client
    typo'ing ``integration`` instead of ``integration_id`` would
    otherwise silently 422 with no actionable message). The UUID
    pattern on ``integration_id`` rejects malformed strings with
    422 instead of letting ``uuid.UUID(...)`` raise a 500 deep in
    ``sync_cost_facts_for_integration``.
    """

    model_config = {"extra": "forbid"}

    integration_id: str = Field(
        ...,
        pattern=r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
    )
    start_date: date
    end_date: date


@router.post("/backfill")
async def backfill_endpoint(
    request: Request,
    payload: BackfillPayload,
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Trigger an out-of-band cost-facts ingestion for ``[start, end]``.

    Runs synchronously in the request — caller is responsible for
    bounding the window (most clouds tolerate ≤90 days; AWS resource-
    level only goes back 14 days). For long backfills, prefer
    multiple smaller ranges or schedule via the cron config dict.
    """
    if not can_manage_cloud(user.role_names):
        raise ForbiddenError()
    if payload.start_date > payload.end_date:
        raise AppException(
            code="BAD_DATE_RANGE",
            message="start_date must be <= end_date",
            status_code=400,
        )
    # ``.days`` on a timedelta is end-exclusive; we want the
    # inclusive day count to match the documented "365-day backfill"
    # contract (Jan 1 → Jan 1 next year is 365 timedelta days but
    # 366 inclusive calendar days). Reject anything beyond that.
    if (payload.end_date - payload.start_date).days >= 365:
        raise AppException(
            code="WINDOW_TOO_LARGE",
            message="max 365-day backfill per call",
            status_code=400,
        )
    # Lower bound on ``start_date``: cloud billing APIs cap retention
    # at the tightest of:
    #   * AWS Cost Explorer: 14 months default (~425 days). Older data
    #     simply isn't accessible via GetCostAndUsage.
    #   * Azure Cost Management: 13 months for most subscription types
    #     (~395 days).
    #   * Aliyun BSS: 6-12 months depending on plan.
    #   * GCP BigQuery export: bounded by the export table's own
    #     retention (admin-controlled, but typically 12-24 months).
    # An admin foot-gun like ``start_date=1970-01-01`` would pass the
    # ``> end_date`` and ``< 365`` gates trivially while still spamming
    # every cloud's billing API with multi-decade queries that mostly
    # return empty. Bound to 14 months (the AWS retention floor) so
    # the backfill stays inside ALL vendors' retention windows; older
    # data is unreachable anyway and the empty-row noise just burns
    # vendor API quota.
    today = datetime.now(UTC).date()
    earliest_allowed = today - timedelta(days=425)
    if payload.start_date < earliest_allowed:
        raise AppException(
            code="WINDOW_TOO_OLD",
            message=(
                f"start_date {payload.start_date} is before the supported "
                f"backfill horizon ({earliest_allowed}, ~14 months). "
                "Cloud billing APIs (AWS CE / Azure CM / Aliyun BSS) "
                "do not retain detail older than this."
            ),
            status_code=400,
        )
    # Take the same per-integration inner lock the cron handler uses
    # (``cloud_sync:cost_facts:{integration_id}``) so an admin-clicked
    # backfill cannot race the 04:xx tick on the SAME integration —
    # without this both paths would enter
    # ``sync_cost_facts_for_integration`` concurrently, double-call the
    # vendor billing API (BigQuery / Cost Explorer / BSS jobs are billed
    # per query / quota'd), and double-write upserts that ON DUP KEY
    # UPDATE only resolves at the row level (driver-side calls are
    # already burnt). LockBusy → return a 409 so the admin can retry
    # after the cron tick clears. The key is per-integration (not
    # handler-wide) so a backfill on integration A doesn't block the
    # cron from running on integration B simultaneously.
    from app.core.database import async_session_factory
    from app.core.distributed_lock import LockBusy, distributed_lock
    detail_base: dict[str, Any] = {
        "start_date": payload.start_date.isoformat(),
        "end_date": payload.end_date.isoformat(),
    }

    async def _audit_failure(*, status: str, extra: dict[str, Any]) -> None:
        # Failure-path audits MUST use a fresh session — the request-
        # scoped ``db`` from ``Depends(get_db)`` is rolled back on the
        # outer ``raise`` (``app/core/database.py:get_db``), so an audit
        # row ``flush()``ed onto it would never persist. Round-2 review
        # caught this — without the fresh session "always audit on
        # every exit path" was a no-op for the LockBusy and generic-
        # exception branches.
        async with async_session_factory() as audit_session:
            await AuditService(audit_session).log(
                trace_id=_trace_id(request),
                actor_type="user",
                actor_user_id=user.id,
                action="cost.backfill.executed",
                status=status,
                resource_type="integration",
                resource_id=payload.integration_id,
                detail={**detail_base, **extra},
            )
            await audit_session.commit()

    try:
        async with distributed_lock(
            f"cloud_sync:cost_facts:{payload.integration_id}", ttl=900,
        ):
            result = await sync_cost_facts_for_integration(
                payload.integration_id,
                start_date=payload.start_date,
                end_date=payload.end_date,
            )
    except LockBusy as exc:
        await _audit_failure(
            status="conflict",
            extra={"reason": "cost_facts_sync busy"},
        )
        raise AppException(
            code="COST_FACTS_SYNC_BUSY",
            message=(
                "Another cost_facts run is in progress (cron or backfill). "
                "Retry once the run finishes."
            ),
            status_code=409,
        ) from exc
    except Exception as exc:
        # Always audit, even when the inner sync throws — without this
        # branch the admin clicks "backfill", the driver crashes mid-
        # run, and there's no audit_logs row attesting the action ever
        # took place. Re-raise so FastAPI maps to 500 and the failure
        # surface stays consistent. ``LockBusy`` was already caught
        # above; this branch only handles unexpected errors from the
        # inner sync.
        # Pre-launch P1 fix (Round 45 follow-up, CR-211 follow-up):
        # scrub the raw cloud-SDK exception text before persisting
        # into audit. boto3 ClientError / AWS API errors / Aliyun
        # OpenAPI errors can echo signature fragments + session
        # tokens; without scrubbing they land in audit_logs.detail
        # for any admin to read.
        from app.integrations.service import _scrub_message  # noqa: PLC0415
        await _audit_failure(
            status="error",
            extra={"error": _scrub_message(str(exc))[:500]},
        )
        raise
    # Success-path audit uses the request session; commits via get_db.
    # Normalize the inner status (which can be "ok" / "partial_failure"
    # / "failed" / "not_supported" / "skipped") into the fixed audit
    # enum. The cron handler explicitly treats ``ok``, ``not_supported``
    # (driver doesn't implement ``list_cost_facts``), and ``skipped``
    # (disabled / missing integration) as non-failure outcomes
    # ([handlers.py:run_cost_facts_sync](server/app/scheduled_jobs/handlers.py));
    # the backfill audit must use the same mapping or analyst dashboards
    # will show backfills against stub drivers as failures while crons
    # against the same drivers look healthy. Round-3 review caught this.
    inner_status = str(result.get("status", "unknown"))
    audit_status = (
        "ok"
        if inner_status in ("ok", "not_supported", "skipped")
        else "error"
    )
    await AuditService(db).log(
        trace_id=_trace_id(request),
        actor_type="user",
        actor_user_id=user.id,
        action="cost.backfill.executed",
        status=audit_status,
        resource_type="integration",
        resource_id=payload.integration_id,
        detail={
            **detail_base,
            "upserted": result.get("upserted"),
            "inner_status": inner_status,
        },
    )
    return success_response(data=result, request=request)


# ── CSV export ─────────────────────────────────────────────────────


@router.get("/export.csv")
async def export_csv_endpoint(
    request: Request,
    start: date | None = None,
    end: date | None = None,
    integration_id: str | None = None,
    department: str | None = Query(None, max_length=128),
    measure: str | None = _MEASURE_QUERY,
    user: AuthUser = Depends(get_current_user),
):
    """Stream a CSV dump of fact rows in the given window.

    Stream-yielded rather than materialised because a 365-day × 100k-
    row export can comfortably exceed 50 MB and FastAPI ``Response``
    buffers the full body before sending the first byte. ``asyncmy``
    streams rows lazily, so memory stays flat regardless of window
    size. Caller's ``Authorization: Bearer ...`` header is required —
    a plain browser navigation will 401; the frontend must trigger
    the download via fetch + Blob to attach the JWT.

    No ``db`` dependency: ``stream_facts_csv`` opens its own session
    so the cursor stays live for the entire response body without
    relying on FastAPI's dependency-cleanup ordering vs streaming.

    Deliberately NOT behind ``_cost_fanout_admission`` (2026-07-29
    review): an export holds exactly ONE pooled connection (the stream
    session) but holds it for the whole download — minutes, not seconds.
    Sharing the fan-out gate would let four downloads lock out the
    interactive dashboard for their entire duration and make the 20 s
    admission timeout start 503-ing ``/breakdown`` behind a slow client;
    and a private gate cannot be released reliably around a
    ``StreamingResponse`` (the slot must be taken before the response is
    returned so a rejection can still be an honest 503, but released from
    the body generator — and a client that disconnects before the body is
    iterated never runs that ``finally``, leaking the slot for the life
    of the process). The bound that matters here is the per-export
    connection count staying at 1; that is pinned by test
    ``test_export_csv_holds_exactly_one_pooled_session``.
    """
    if not can_view_finance(user.role_names):
        raise ForbiddenError()
    resolved_measure = _resolve_measure_or_403(measure, user)
    if integration_id is not None:
        try:
            uuid.UUID(integration_id)
        except (TypeError, ValueError) as exc:
            raise AppException(
                code="BAD_INTEGRATION_ID",
                message="integration_id must be a valid UUID",
                status_code=400,
            ) from exc
    start_dt, end_dt = _parse_window(start, end, max_days=365)
    # Audit BEFORE streaming AND on a fresh, immediately-committed
    # session — a CSV dump is bulk cost-data exfil and the trace must
    # persist even if the download aborts mid-stream. The request-
    # scoped ``db`` only commits in ``get_db`` AFTER the StreamingResponse
    # finishes; if the client closes the tab mid-download, ``get_db``
    # rolls back and the audit row vanishes (Round-3 review caught
    # this as the same bug class as the backfill exception-path audit
    # the Round-2 review fixed).
    from app.core.database import async_session_factory
    async with async_session_factory() as audit_session:
        await AuditService(audit_session).log(
            trace_id=_trace_id(request),
            actor_type="user",
            actor_user_id=user.id,
            action="cost.facts.exported",
            status="ok",
            resource_type="cost_facts",
            resource_id=integration_id,
            detail={
                "start_date": start_dt.isoformat(),
                "end_date": end_dt.isoformat(),
                "integration_id": integration_id,
                "department": department,
                "measure": resolved_measure,
            },
        )
        await audit_session.commit()
    filename = f"cost-facts-{start_dt.isoformat()}-{end_dt.isoformat()}.csv"
    return StreamingResponse(
        stream_facts_csv(
            start=start_dt,
            end=end_dt,
            integration_id=integration_id,
            department=department,
            measure=resolved_measure,
        ),
        media_type="text/csv; charset=utf-8",
        headers={
            "Content-Disposition": f'attachment; filename="{filename}"',
            # Defence in depth — discourage browsers from sniffing
            # the response as anything other than CSV.
            "X-Content-Type-Options": "nosniff",
        },
    )