"""Budget alerting for the ``cmdb.billing_sync`` handler.

Threshold lives in ``platform_settings`` so admins can adjust without
code / deploy:

  * ``billing.budget_usd.default`` — org-wide monthly cap (float USD).
  * ``billing.budget_usd.<integration_slug>`` — optional per-account
    override. Looked up first; if absent, the default is used.

口径说明（2026-07 成本口径改造）：本模块的月度预算卡发到 Feishu
**告警 webhook**（IT/云管理/财务的运维群，属实付口径保密圈），刻意
保持发票口径（``cmdb_billing_monthly``，与云厂商账单可对账）。
Owner 侧（研发可见）的预算/波动规则走 ``lifecycle_service`` 的
``scope.alert_rules`` 路径，输入已切业务折前成本（list）——两条通道
口径不同是设计决定，不是漂移。

Alerts fire via the primary Feishu integration's ``alert_webhook_url``
if configured. Errors never propagate — an alert failure must not
cause the billing row write to be rolled back, which is why this is
wrapped in try/except by the caller too.

Idempotency (migration 116, 2026-05-05): the cron runs daily, so a
sustained over-budget account would fire 30 alerts/month without
deduplication. ``cmdb_billing_alerts_dedup`` records the last alert
per ``(integration_id, month)``. Decision matrix:

* No prior row → send + INSERT.
* Prior row, ``last_alerted_at`` < 24h → SKIP silently.
* Prior row, ``last_alerted_at`` ≥ 24h AND ``total_usd`` ≥ 110% of
  ``last_alerted_amount_usd`` → resend + UPDATE.
* Prior row, ``last_alerted_at`` ≥ 24h AND no material change → SKIP.

Audit row (``cmdb.billing_alert_sent``) is written on every actual send
so finance / SecOps can post-hoc reconstruct who got pinged when.
"""
from __future__ import annotations

import math
import uuid
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Any

import httpx
from sqlalchemy import delete, select

from app.audit.service import AuditService
from app.core.database import async_session_factory
from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.models import CmdbBillingAlertsDedup
from app.integrations.compat import get_feishu_config
from app.integrations.models import PlatformSetting
from app.integrations.service import get_platform_setting

logger = get_safe_logger(__name__)


# Idempotency parameters — tuned for a daily cron + sustained-overrun
# scenario. 24h cooldown + 10% materiality bar means a flat-line
# overrun fires once; a 10%+ jump fires again next day.
_DEDUP_COOLDOWN = timedelta(hours=24)
_DEDUP_MATERIAL_INCREASE = Decimal("1.10")  # 10% above last-alerted total


def _coerce_budget_value(raw: object) -> float | None:
    """Coerce a ``platform_settings.value`` JSON cell to a budget USD
    float, or ``None`` if the value is not a finite positive number.

    Round-27 pre-launch (2026-05-25): mirrors ``fx_rate._coerce_float``'s
    ``isinstance(raw, bool): return None`` guard. ``bool`` is a Python
    subclass of ``int``, so a platform_setting accidentally written as
    JSON ``true`` would otherwise ``float(True) -> 1.0`` and silently
    set a $1 monthly budget that fires every cron tick. We reject
    booleans + non-finite + non-positive values explicitly.
    """
    if isinstance(raw, bool):
        return None
    if isinstance(raw, (int, float)):
        if not math.isfinite(raw):
            return None
        return float(raw) if raw > 0 else None
    if isinstance(raw, str):
        try:
            v = float(raw.strip())
        except (TypeError, ValueError):
            return None
        if not math.isfinite(v):
            return None
        return v if v > 0 else None
    if isinstance(raw, Decimal):
        try:
            v = float(raw)
        except (TypeError, ValueError):
            return None
        return v if math.isfinite(v) and v > 0 else None
    return None


async def _lookup_budget(integration_slug: str | None) -> float | None:
    """Return the effective monthly USD budget or None when disabled.

    All coercion runs through ``_coerce_budget_value`` so booleans,
    NaN, infinity and non-positive values are rejected the same way at
    both lookup tiers.
    """
    async with async_session_factory() as session:
        if integration_slug:
            per_acct = await session.get(
                PlatformSetting, f"billing.budget_usd.{integration_slug}",
            )
            if per_acct is not None:
                coerced = _coerce_budget_value(per_acct.value)
                if coerced is not None:
                    return coerced
        default = await session.get(PlatformSetting, "billing.budget_usd.default")
        if default is None:
            return None
        return _coerce_budget_value(default.value)


async def _resolve_slug(integration_id: str) -> str | None:
    """Resolve integration slug for budget lookup.

    Returns ``None`` if the integration row is missing OR disabled —
    Round-22 audit hardening: ``run_billing_sync`` already gates on
    ``Integration.enabled`` at start, but an admin can toggle
    ``enabled=False`` between that gate and our call (the cron tick
    holds the row in memory, the admin acts via the API). We MUST
    NOT fire an alert for a tenant the admin just disabled. The
    caller short-circuits on ``None`` slug → ``no-budget-configured``
    return.
    """
    import uuid as _uuid

    from app.integrations.models import Integration

    async with async_session_factory() as session:
        row = await session.get(Integration, _uuid.UUID(integration_id))
        if row is None or not row.enabled:
            return None
        return row.slug


async def _check_and_update_dedup(
    *,
    integration_id: str,
    month_key: str,
    total_usd: float,
    budget_usd: float,
) -> tuple[bool, str]:
    """Read-only decision: should an alert fire for this run?

    Returns ``(should_send, reason)``. ``reason`` is a short tag used
    for logging when the call is skipped — the caller maps it onto a
    return-status string. NOTE: this function NO LONGER writes the
    dedup row. The write is split out into :func:`_record_dedup_send`
    so it can run AFTER the webhook POST succeeds — otherwise a
    transient webhook 5xx would bake in 24h of "cooldown-active"
    silences while the budget breach is real and growing.
    (Pre-launch round 2 security P0.)
    """
    now = datetime.now(UTC)
    total_dec = Decimal(str(total_usd))

    async with async_session_factory() as session:
        existing = (
            await session.execute(
                select(CmdbBillingAlertsDedup)
                .where(CmdbBillingAlertsDedup.integration_id == integration_id)
                .where(CmdbBillingAlertsDedup.month == month_key)
            )
        ).scalar_one_or_none()

        if existing is None:
            return True, "first-alert"

        # Coerce naive timestamps to UTC for safe comparison — MySQL
        # DATETIME loses tz info on read even when the ORM declares
        # ``DateTime(timezone=True)`` (same family bug as billing_service
        # had to fix in 2026-05-01).
        last_at = existing.last_alerted_at
        if last_at is not None and last_at.tzinfo is None:
            last_at = last_at.replace(tzinfo=UTC)
        if last_at is not None and (now - last_at) < _DEDUP_COOLDOWN:
            return False, "cooldown-active"

        # ≥ 24h since last alert. Only resend if the overrun has
        # materially worsened — otherwise a flat-line over-budget
        # account would produce a daily ping forever.
        last_amount = existing.last_alerted_amount_usd or Decimal("0")
        if last_amount <= 0:
            # Corrupted baseline — treat as "send to recover the
            # legitimate baseline" so a future legitimate alert can
            # overwrite the zero row. Pre-launch round-3 audit P1:
            # previously this returned ``False`` (silenced forever),
            # which meant one bad UPDATE / migration error locked the
            # alerting pipeline for that integration permanently.
            # Re-sending here re-anchors ``last_alerted_amount_usd``
            # on the next ``_record_dedup_send`` call and the 24h
            # cooldown resumes from the legitimate spend value.
            logger.warning(
                "cmdb.billing_alert: zero baseline detected for "
                "integration=%s month=%s — re-firing to recover",
                integration_id, month_key,
            )
            return True, "recovering-from-zero-baseline"
        ratio = total_dec / last_amount
        if ratio >= _DEDUP_MATERIAL_INCREASE:
            return True, "material-increase"

        # Round-22 audit follow-up: material-increase ratio compares
        # against the LAST alerted amount, not the budget. A slow drift
        # from $150 → $180 → $200 → ... → $2000 each stays under the
        # +10% threshold day-over-day and silences the channel even
        # though the integration is now 20× over its budget. Override
        # the silence when total crosses a NEW integer multiple of the
        # configured budget since the last alert (2×, 3×, 5×, 10×,
        # 20×, 50×, 100×). The integer-multiple ladder is coarse
        # enough to avoid spam while still catching real runaway
        # spend.
        from decimal import Decimal as _D
        budget_dec_local = _D(str(budget_usd)) if budget_usd > 0 else _D("0")
        if budget_dec_local > 0:
            milestones = (_D("2"), _D("3"), _D("5"), _D("10"), _D("20"),
                          _D("50"), _D("100"))
            current_multiple = total_dec / budget_dec_local
            last_multiple = last_amount / budget_dec_local
            for m in milestones:
                if last_multiple < m <= current_multiple:
                    return True, f"crossed-{m}x-budget"

        return False, "no-material-increase"


async def _record_dedup_send(
    integration_id: str,
    month_key: str,
    total_usd: float,
    budget_usd: float,
) -> None:
    """Persist the dedup row AFTER a successful webhook POST.

    Split out from ``_check_and_update_dedup`` so a webhook failure
    leaves no "we just alerted" row behind. Otherwise the next cron
    run would hit ``cooldown-active`` and stay silent for 24h while
    the original alert never reached operators.

    Concurrency: two cron runs racing on the same (integration_id,
    month) will both INSERT — the UNIQUE index guarantees the loser
    raises IntegrityError. We catch it and fall through to UPDATE.
    """
    from sqlalchemy.exc import IntegrityError

    now = datetime.now(UTC)
    total_dec = Decimal(str(total_usd))
    budget_dec = Decimal(str(budget_usd))

    async with async_session_factory() as session:
        existing = (
            await session.execute(
                select(CmdbBillingAlertsDedup)
                .where(CmdbBillingAlertsDedup.integration_id == integration_id)
                .where(CmdbBillingAlertsDedup.month == month_key)
            )
        ).scalar_one_or_none()
        if existing is None:
            session.add(CmdbBillingAlertsDedup(
                id=uuid.uuid4(),
                integration_id=integration_id,
                month=month_key,
                last_alerted_at=now,
                last_alerted_amount_usd=total_dec,
                last_alerted_budget_usd=budget_dec,
            ))
            try:
                await session.commit()
                return
            except IntegrityError:
                # Concurrent run won the INSERT. The winning caller's
                # dedup row carries the authoritative snapshot they
                # claimed against. Round-22: previously we re-SELECT
                # + UPDATE with OUR ``total_dec``, which overwrites
                # the winning caller's amount with a value sampled
                # from a different cron tick. If the two callers
                # captured `total_usd` at slightly different times,
                # the dedup row would record the loser's number while
                # the alert FEISHU CARD shows the winner's number —
                # a forensic mismatch nobody can untangle later.
                # Just roll back and return; the winner's row stands.
                await session.rollback()
                return
        existing.last_alerted_at = now
        existing.last_alerted_amount_usd = total_dec
        existing.last_alerted_budget_usd = budget_dec
        await session.commit()


async def _record_alert_audit(
    *,
    integration_id: str,
    integration_name: str,
    month_key: str,
    total_usd: float,
    budget_usd: float,
    dedup_reason: str,
) -> None:
    """Append a ``cmdb.billing_alert_sent`` audit row.

    Best-effort — an audit-write failure must not block the alert flow,
    so any error here is logged and swallowed."""
    try:
        async with async_session_factory() as session:
            audit = AuditService(session)
            await audit.log(
                trace_id=f"cmdb.billing_alert.{integration_id}.{month_key}",
                actor_type="system",
                action="cmdb.billing_alert_sent",
                status="success",
                resource_type="cmdb_integration",
                resource_id=integration_id,
                detail={
                    "integration_name": integration_name,
                    "month": month_key,
                    "total_usd": total_usd,
                    "budget_usd": budget_usd,
                    "dedup_reason": dedup_reason,
                },
            )
            await session.commit()
    except Exception:  # noqa: BLE001
        logger.exception("cmdb.billing_alert_sent audit write failed (non-fatal)")


async def maybe_send_budget_alert(
    *,
    integration_id: str,
    integration_name: str,
    month: date,
    total_usd: float,
) -> dict[str, Any]:
    """Compare total vs threshold; push a Feishu alert on exceed.

    Idempotent across daily cron runs — see module docstring. Returns
    one of:

    * ``no-budget-configured`` — no threshold set for this account.
    * ``within-budget`` — total under threshold; nothing to do.
    * ``exceeded-deduped`` — over budget but the dedup window says skip
      (24h cooldown active or no material increase since last alert).
    * ``exceeded-no-webhook`` — over budget but no Feishu webhook
      configured; logged at INFO so admins notice the missing config.
    * ``exceeded-webhook-rejected`` — webhook URL failed SSRF guard.
    * ``exceeded-post-failed`` — webhook request raised; error logged.
    * ``exceeded-alerted`` — alert posted, dedup row written, audit row
      written.
    """
    slug = await _resolve_slug(integration_id)
    if slug is None:
        # Round-22 review-2 follow-up: ``_resolve_slug`` returns None
        # when the integration row is missing OR disabled. We MUST
        # short-circuit BEFORE ``_lookup_budget(None)`` — that path
        # falls through to the global ``billing.budget_usd.default``
        # row and would silently fire the alert against the default
        # budget for a tenant Helix no longer manages.
        return {"status": "integration-disabled-or-missing"}
    budget = await _lookup_budget(slug)
    if budget is None or budget <= 0:
        return {"status": "no-budget-configured"}
    if total_usd <= budget:
        return {"status": "within-budget", "total_usd": total_usd, "budget": budget}

    month_key = f"{month:%Y-%m}"
    # Claim-first dedup to close the dedup-after-POST race:
    #   1. Acquire a SHORT-TTL distributed lock (5s).
    #   2. Inside the lock: SELECT existing dedup row + decide
    #      should_send. If yes, INSERT a placeholder dedup row that
    #      reserves the slot. Release the lock.
    #   3. POST the webhook OUTSIDE the lock (so Feishu latency can't
    #      stall concurrent callers — they see the claim row and
    #      short-circuit on the cooldown check).
    #   4. On POST success: UPDATE the placeholder with the final
    #      amount + write audit.
    #   5. On POST failure: DELETE the placeholder so the next caller
    #      can re-fire — a transient Feishu blip must NOT bake in 24h
    #      of silence on a real budget breach.
    #
    # The lock window is now microseconds (single SELECT + INSERT), so
    # the previous concern about holding a lock across HTTP is gone.
    # Concurrent callers see LockBusy → back off; either they're
    # racing on the SAME alert (winning caller will commit the claim,
    # losing caller's next attempt sees it on cooldown) or they're
    # racing on different keys (different lock_key, no contention).
    from app.core.distributed_lock import LockBusy, distributed_lock

    lock_key = f"cmdb_billing_alert:{integration_id}:{month_key}"
    try:
        async with distributed_lock(lock_key, ttl=5):
            should_send, dedup_reason = await _check_and_update_dedup(
                integration_id=integration_id,
                month_key=month_key,
                total_usd=total_usd,
                budget_usd=budget,
            )
            if not should_send:
                return {
                    "status": "exceeded-deduped",
                    "reason": dedup_reason,
                    "total_usd": total_usd,
                    "budget": budget,
                }
            # Claim the slot atomically — INSERT placeholder dedup row
            # that fixes the (integration, month) so concurrent callers
            # see it on cooldown until POST resolves.
            await _record_dedup_send(
                integration_id=integration_id,
                month_key=month_key,
                total_usd=total_usd,
                budget_usd=budget,
            )
    except LockBusy:
        # Concurrent caller is inside the claim window (≤5s). Back off
        # so we don't queue; the winner is already committing the
        # claim. Our next tick will either find the claim row on
        # cooldown (no-op) or the winner's POST failed and DELETEd it
        # (we re-fire then).
        return {
            "status": "exceeded-deduped",
            "reason": "concurrent-alert-in-flight",
            "total_usd": total_usd,
            "budget": budget,
        }

    # POST + audit happen OUTSIDE the lock. ``_send_budget_alert_post``
    # rolls back the dedup row on POST failure so the next caller can
    # legitimately re-fire.
    return await _send_budget_alert_post(
        integration_id=integration_id,
        integration_name=integration_name,
        month=month,
        month_key=month_key,
        total_usd=total_usd,
        budget=budget,
        dedup_reason=dedup_reason,
    )


async def _delete_dedup_row(integration_id: str, month_key: str) -> None:
    """Roll back the placeholder dedup row on POST failure.

    Best-effort: if the DELETE itself fails, the worst-case is one
    cooldown window of silence on a transient Feishu blip. Log and
    swallow.
    """
    try:
        async with async_session_factory() as session:
            await session.execute(
                delete(CmdbBillingAlertsDedup).where(
                    CmdbBillingAlertsDedup.integration_id == integration_id,
                    CmdbBillingAlertsDedup.month == month_key,
                )
            )
            await session.commit()
    except Exception:  # noqa: BLE001
        logger.exception(
            "cmdb.billing_alert: failed to roll back dedup row for "
            "integration=%s month=%s — next 24h may be silent on this "
            "(integration, month) tuple",
            integration_id, month_key,
        )


async def _send_budget_alert_post(
    *,
    integration_id: str,
    integration_name: str,
    month: date,
    month_key: str,
    total_usd: float,
    budget: float,
    dedup_reason: str,
) -> dict[str, Any]:
    """POST the webhook + write audit. Called AFTER the dedup row is
    already claimed (inside the lock window in
    ``maybe_send_budget_alert``). On any POST failure, roll back the
    claim so the next cron tick can re-fire.
    """
    # Round-27 pre-launch (2026-05-25): defense-in-depth — the only
    # production caller (``maybe_send_budget_alert``) gates ``budget > 0``
    # before invoking, but a future direct caller (MCP tool / test
    # fixture / admin-only re-trigger endpoint) without that gate would
    # ``ZeroDivisionError`` on line ``(total_usd - budget) / budget``.
    # Refuse rather than emit a nonsensical card or crash mid-send.
    if budget <= 0:
        raise ValueError(
            f"_send_budget_alert_post: budget must be > 0 (got {budget!r}); "
            "caller must enforce gating before invoking.",
        )

    # Round-22 audit follow-up: the Feishu card amount is sampled from
    # ``cmdb_billing_monthly.total_cost_usd`` at cron-tick time, while
    # the admin dashboard at ``/admin/cloud/billing`` aggregates
    # ``cmdb_cloud_billing_daily`` for the same month. When the global
    # FX-rate refresh ticks between the two reads, finance opens the
    # card ($10,500) and the dashboard ($10,420) and assumes Helix is
    # lying. Anchor the card to a single source-of-truth row + an
    # explicit fetched-at stamp + a deep link so the operator can
    # cross-check without guessing.
    now_for_audit = datetime.now(UTC)
    # ``/admin/cloud/billing`` is a redirect-only stub
    # (``<Navigate to="/admin/cloud/overview?tab=billing" replace />``)
    # whose hard-coded querystring REPLACES any incoming params — i.e.
    # ``?month=…&integration_id=…`` get stripped on the bounce. Point
    # the card straight at the real page so finance lands on the
    # alert's month + account; CloudOverviewPage's billing tab consumes
    # both ``month`` and ``integration_id`` from search params.
    #
    # Round-3 logic audit MEDIUM: prepend ``app.frontend_base_url`` so
    # the URL renders as a clickable target in Feishu chat (relative
    # paths starting with ``/`` are not auto-linkified). Falls back to
    # the bare path when the setting is unset / non-http(s) — same
    # safe-omit posture as ``offboarding/service._notify_transfer_accepted_it``.
    portal_url = await get_platform_setting("app.frontend_base_url", default="")
    deep_link_path = (
        f"/admin/cloud/overview?tab=billing&month={month_key}"
        f"&integration_id={integration_id}"
    )
    if portal_url and str(portal_url).startswith(("http://", "https://")):
        deep_link = f"{portal_url}{deep_link_path}"
    else:
        deep_link = deep_link_path
    # Round-51 S2-Feishu: Feishu group-bot text msg_type interprets
    # ``<at user_id="all"></at>`` tags. ``integration_name`` is the
    # admin-set alias (auto-derived from cloud-side fields per memory
    # ``cmdb/aws — alias auto-derived``). A maliciously / accidentally
    # set alias like ``prod <at user_id="all"></at>`` would otherwise
    # @all the entire Feishu group on every monthly budget alert. LF
    # injection in the alias would also let an attacker forge footer
    # text. Sanitise at the call site — strip CR/LF and HTML brackets.
    from app.notifications.text_safety import safe_envelope_text
    safe_integration_name = safe_envelope_text(integration_name, limit=80)
    safe_integration_name = (
        safe_integration_name.replace("<", "&lt;").replace(">", "&gt;")
    )
    # 双口径保密红线(cost-measure-surface-sweep.md;2026-07-30 第二批复查
    # P1-1)。这条推送的收件人是**飞书群成员**——群成员资格代码无从校验,
    # 结构上不可能是保密圈的子集,而 `_invoice_measure_recipient_ids` 的
    # 按收件人渲染(告警卡那两族用的)在群 webhook 上没有着力点。金额来自
    # `cmdb_billing_monthly.total_cost_usd`,该表**没有 list 列**,所以这个
    # 数恒为实付(见 `masking.AUDIT_ACTION_COST_BASIS` 里同 action 的分类);
    # 群里任何非圈成员拿它去除自己合法可读的折前月合计,即得折扣率。
    #
    # 于是走「写侧不落」的同一条修法:超出**百分比**(无量纲,正是「为什么
    # 这条告警会发」要的那个数)+ 对账 deep link,绝对金额让收件人去有
    # measure 门禁的账单页看。审计行(`_record_alert_audit`)仍存完整金额
    # 供圈内取证,读侧由 masking 按 action 分类脱敏。
    over_pct = (total_usd - budget) / budget * 100
    msg = (
        f"💰 月度账单超预算告警\n"
        f"云账号: {safe_integration_name}\n"
        f"账单月: {month_key}\n"
        f"超出预算: {over_pct:.1f}%\n"
        f"触发原因: {dedup_reason}\n"
        f"数据源: cmdb_billing_monthly (fetched "
        f"{now_for_audit:%Y-%m-%d %H:%M UTC})\n"
        f"金额为实付口径,仅财务 / 云管理员可见 —— 请在对账明细页查看: "
        f"{deep_link}\n\n"
        f"如对帐单有任何疑问，请联系 IT 团队进行确认。"
    )
    try:
        feishu_cfg = await get_feishu_config()
        webhook_url = str(feishu_cfg.get("alert_webhook_url") or "")
    except Exception:  # noqa: BLE001
        webhook_url = ""
    if not webhook_url:
        # Roll back the dedup claim — no webhook configured means the
        # next admin who fixes the config should immediately re-fire,
        # not wait 24h on a cooldown for an alert that was never sent.
        await _delete_dedup_row(integration_id, month_key)
        # Redact: log only correlation identifiers, NOT the rendered
        # message body (which includes the absolute USD figures). If
        # logs are forwarded to a central aggregator with broader read
        # ACLs than the app, the previous shape leaked monthly spend
        # per integration to any log searcher. Audit row + dedup row
        # carry the full detail; the operational log just needs to
        # show "config drift detected on integration X".
        logger.info(
            "billing budget exceeded but no Feishu alert_webhook_url "
            "configured: integration_id=%s month=%s",
            integration_id, month_key,
        )
        return {"status": "exceeded-no-webhook", "total_usd": total_usd, "budget": budget}
    # Pre-launch hardening 2026-05-04: SSRF guard on the admin-pasted
    # Feishu webhook URL. Without this an admin who pastes a URL on
    # the internal network (metadata service, internal admin portal)
    # would tunnel the alert payload (org-internal usage figures) to
    # that target.
    try:
        from app.integrations._url_safety import (
            UnsafeURLError,
            safe_async_http_transport,
        )
        transport = await safe_async_http_transport(
            webhook_url,
            allowed_suffixes=("feishu.cn", "larksuite.com"),
        )
    except UnsafeURLError as exc:
        # Same rollback rationale as the no-webhook path above.
        await _delete_dedup_row(integration_id, month_key)
        # Narrow exception so genuine SSRF rejections are
        # distinguishable from config-corruption / urlparse edge cases
        # — the previous bare ``except Exception`` masked everything as
        # "rejected" and admins lost the diagnostic.
        logger.warning(
            "Feishu billing alert webhook URL rejected by SSRF guard: %s",
            exc.message,
        )
        return {"status": "exceeded-webhook-rejected"}
    try:
        # ``trust_env=False`` blocks HTTP_PROXY env-var injection — a
        # compromised sidecar / debug override could otherwise tunnel
        # this alert (with the bearer-equivalent webhook secret in
        # the URL) through an attacker-controlled proxy.
        async with httpx.AsyncClient(
            timeout=15, trust_env=False, transport=transport,
        ) as client:
            resp = await client.post(
                webhook_url,
                json={"msg_type": "text", "content": {"text": msg}},
            )
            result = resp.json()
        # Treat ANY failure indicator as a failure — modern Feishu/Lark
        # bot endpoints return ``{"code": 9499, "msg": "..."}`` with no
        # ``StatusCode`` key, so the previous AND-short-circuit
        # (``StatusCode not in (0,None) AND code not in (0,None)``)
        # silently treated such failures as success — wrote a dedup
        # row, audited "alert_sent", and then silenced the budget
        # alert for 24h while finance never received it.
        # (Round-17 audit.)
        ok = (
            resp.status_code == 200
            and result.get("StatusCode") in (0, None)
            and result.get("code") in (0, None)
        )
        if not ok:
            logger.warning(
                "Feishu billing alert non-zero status (%s): %s",
                resp.status_code, result,
            )
            # Webhook returned non-zero — roll back the claim so the
            # next caller can re-fire instead of being silenced by the
            # cooldown. (Round 2 security P0 + round-2 self-review.)
            await _delete_dedup_row(integration_id, month_key)
            return {"status": "exceeded-post-failed"}
    except Exception:  # noqa: BLE001
        logger.exception("Feishu billing alert post failed")
        await _delete_dedup_row(integration_id, month_key)
        return {"status": "exceeded-post-failed"}

    # Webhook succeeded — the claim row was already inserted inside
    # the lock window in ``maybe_send_budget_alert``. Audit-only here.
    await _record_alert_audit(
        integration_id=integration_id,
        integration_name=integration_name,
        month_key=month_key,
        total_usd=total_usd,
        budget_usd=budget,
        dedup_reason=dedup_reason,
    )
    return {
        "status": "exceeded-alerted",
        "reason": dedup_reason,
        "total_usd": total_usd,
        "budget": budget,
    }


__all__ = ["maybe_send_budget_alert"]